nautilus_hyperliquid/websocket/
handler.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! WebSocket message handler for Hyperliquid.
17
18use std::{
19    collections::VecDeque,
20    sync::{
21        Arc,
22        atomic::{AtomicBool, Ordering},
23    },
24};
25
26use ahash::{AHashMap, AHashSet};
27use nautilus_common::cache::fifo::FifoCache;
28use nautilus_core::{
29    AtomicTime, MUTEX_POISONED, Params, nanos::UnixNanos, time::get_atomic_clock_realtime,
30};
31use nautilus_model::{
32    data::{BarType, CustomData, Data, DataType},
33    identifiers::{AccountId, InstrumentId},
34    instruments::{Instrument, InstrumentAny},
35    types::Price,
36};
37use nautilus_network::{
38    RECONNECTED,
39    retry::{RetryManager, create_websocket_retry_manager},
40    websocket::{SubscriptionState, WebSocketClient},
41};
42use rust_decimal::Decimal;
43use tokio_tungstenite::tungstenite::Message;
44use ustr::Ustr;
45
46use super::{
47    client::{AssetContextDataType, CloidCache},
48    enums::HyperliquidWsChannel,
49    error::HyperliquidWsError,
50    messages::{
51        CandleData, ExecutionReport, HyperliquidWsMessage, HyperliquidWsRequest, NautilusWsMessage,
52        PostRequest, SubscriptionRequest, WsActiveAssetCtxData, WsAllDexsAssetCtxsData,
53        WsUserEventData,
54    },
55    parse::{
56        parse_ws_asset_context, parse_ws_candle, parse_ws_fill_report, parse_ws_open_interest,
57        parse_ws_order_book_deltas, parse_ws_order_book_depth10, parse_ws_order_status_report,
58        parse_ws_public_trade, parse_ws_quote_tick, parse_ws_trade_tick,
59    },
60    post::PostRouter,
61    trades::TradeStreamUses,
62};
63use crate::data_types::{
64    HyperliquidAllDexsAssetCtxs, HyperliquidAllMids, HyperliquidDexAssetCtx,
65    HyperliquidImpactPrices,
66};
67
68/// Commands sent from the outer client to the inner message handler.
69#[derive(Debug)]
70#[expect(
71    clippy::large_enum_variant,
72    reason = "Commands are ephemeral and immediately consumed"
73)]
74#[allow(private_interfaces)]
75pub enum HandlerCommand {
76    /// Set the WebSocketClient for the handler to use.
77    SetClient(WebSocketClient),
78    /// Disconnect the WebSocket connection.
79    Disconnect,
80    /// Subscribe to the given subscriptions.
81    Subscribe {
82        subscriptions: Vec<SubscriptionRequest>,
83    },
84    /// Unsubscribe from the given subscriptions.
85    Unsubscribe {
86        subscriptions: Vec<SubscriptionRequest>,
87    },
88    /// Send a WebSocket post request.
89    Post { id: u64, request: PostRequest },
90    /// Initialize the instruments cache with the given instruments.
91    InitializeInstruments(Vec<InstrumentAny>),
92    /// Update a single instrument in the cache.
93    UpdateInstrument(InstrumentAny),
94    /// Add a bar type mapping for candle parsing.
95    AddBarType { key: String, bar_type: BarType },
96    /// Remove a bar type mapping.
97    RemoveBarType { key: String },
98    /// Update asset context subscriptions for a coin.
99    UpdateAssetContextSubs {
100        coin: Ustr,
101        data_types: AHashSet<AssetContextDataType>,
102    },
103    /// Update the logical consumers of a `trades` stream for a coin.
104    UpdateTradeSubs { coin: Ustr, uses: TradeStreamUses },
105    /// Cache the ordered instrument IDs needed to normalize `allDexsAssetCtxs`.
106    CacheAllDexAssetCtxsInstrumentIds(AHashMap<Ustr, Vec<Option<InstrumentId>>>),
107    /// Cache spot fill coin mappings for instrument lookup.
108    CacheSpotFillCoins(AHashMap<Ustr, Ustr>),
109    /// Flag whether the `l2Book` stream for `coin` should also be emitted
110    /// as [`NautilusWsMessage::Depth10`] snapshots.
111    SetDepth10Sub { coin: Ustr, subscribed: bool },
112}
113
114#[derive(Default)]
115struct AssetContextCaches {
116    mark_price: AHashMap<Ustr, Decimal>,
117    index_price: AHashMap<Ustr, Decimal>,
118    funding_rate: AHashMap<Ustr, Decimal>,
119    open_interest: AHashMap<Ustr, Decimal>,
120}
121
122impl AssetContextCaches {
123    fn clear(&mut self, coin: Ustr, data_type: AssetContextDataType) {
124        match data_type {
125            AssetContextDataType::MarkPrice => {
126                self.mark_price.remove(&coin);
127            }
128            AssetContextDataType::IndexPrice => {
129                self.index_price.remove(&coin);
130            }
131            AssetContextDataType::FundingRate => {
132                self.funding_rate.remove(&coin);
133            }
134            AssetContextDataType::OpenInterest => {
135                self.open_interest.remove(&coin);
136            }
137        }
138    }
139
140    fn clear_removed(
141        &mut self,
142        coin: Ustr,
143        previous_data_types: Option<&AHashSet<AssetContextDataType>>,
144        next_data_types: &AHashSet<AssetContextDataType>,
145    ) {
146        let Some(previous_data_types) = previous_data_types else {
147            return;
148        };
149
150        for data_type in previous_data_types {
151            if !next_data_types.contains(data_type) {
152                self.clear(coin, *data_type);
153            }
154        }
155    }
156}
157
158pub(super) struct FeedHandler {
159    clock: &'static AtomicTime,
160    signal: Arc<AtomicBool>,
161    client: Option<WebSocketClient>,
162    cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
163    raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
164    out_tx: tokio::sync::mpsc::UnboundedSender<NautilusWsMessage>,
165    account_id: Option<AccountId>,
166    subscriptions: SubscriptionState,
167    post_router: Arc<PostRouter>,
168    retry_manager: RetryManager<HyperliquidWsError>,
169    message_buffer: VecDeque<NautilusWsMessage>,
170    instruments: AHashMap<Ustr, InstrumentAny>,
171    cloid_cache: CloidCache,
172    bar_types_cache: AHashMap<String, BarType>,
173    bar_cache: AHashMap<String, CandleData>,
174    asset_context_subs: AHashMap<Ustr, AHashSet<AssetContextDataType>>,
175    trade_subs: AHashMap<Ustr, TradeStreamUses>,
176    all_dex_asset_ctxs_instrument_ids: AHashMap<Ustr, Vec<Option<InstrumentId>>>,
177    depth10_subs: AHashSet<Ustr>,
178    processed_trade_ids: FifoCache<u64, 10_000>,
179    processed_public_trade_ids: FifoCache<(Ustr, u64), 10_000>,
180    asset_context_caches: AssetContextCaches,
181}
182
183impl FeedHandler {
184    /// Creates a new [`FeedHandler`] instance.
185    #[allow(
186        clippy::too_many_arguments,
187        reason = "constructs the handler from independent runtime channels and caches"
188    )]
189    pub(super) fn new(
190        signal: Arc<AtomicBool>,
191        cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
192        raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
193        out_tx: tokio::sync::mpsc::UnboundedSender<NautilusWsMessage>,
194        account_id: Option<AccountId>,
195        subscriptions: SubscriptionState,
196        cloid_cache: CloidCache,
197        post_router: Arc<PostRouter>,
198    ) -> Self {
199        Self {
200            clock: get_atomic_clock_realtime(),
201            signal,
202            client: None,
203            cmd_rx,
204            raw_rx,
205            out_tx,
206            account_id,
207            subscriptions,
208            post_router,
209            retry_manager: create_websocket_retry_manager(),
210            message_buffer: VecDeque::new(),
211            instruments: AHashMap::new(),
212            cloid_cache,
213            bar_types_cache: AHashMap::new(),
214            bar_cache: AHashMap::new(),
215            asset_context_subs: AHashMap::new(),
216            trade_subs: AHashMap::new(),
217            all_dex_asset_ctxs_instrument_ids: AHashMap::new(),
218            depth10_subs: AHashSet::new(),
219            processed_trade_ids: FifoCache::new(),
220            processed_public_trade_ids: FifoCache::new(),
221            asset_context_caches: AssetContextCaches::default(),
222        }
223    }
224
225    /// Send a message to the output channel.
226    pub(super) fn send(&self, msg: NautilusWsMessage) -> Result<(), String> {
227        self.out_tx
228            .send(msg)
229            .map_err(|e| format!("Failed to send message: {e}"))
230    }
231
232    /// Check if the handler has received a stop signal.
233    pub(super) fn is_stopped(&self) -> bool {
234        self.signal.load(Ordering::Relaxed)
235    }
236
237    async fn send_with_retry(&self, payload: String) -> anyhow::Result<()> {
238        if let Some(client) = &self.client {
239            self.retry_manager
240                .execute_with_retry(
241                    "websocket_send",
242                    || {
243                        let payload = payload.clone();
244                        async move {
245                            client.send_text(payload, None).await.map_err(|e| {
246                                HyperliquidWsError::ClientError(format!("Send failed: {e}"))
247                            })
248                        }
249                    },
250                    should_retry_hyperliquid_error,
251                    create_hyperliquid_timeout_error,
252                )
253                .await
254                .map_err(|e| anyhow::anyhow!("{e}"))
255        } else {
256            Err(anyhow::anyhow!("No WebSocket client available"))
257        }
258    }
259
260    pub(super) async fn next(&mut self) -> Option<NautilusWsMessage> {
261        if let Some(msg) = self.message_buffer.pop_front() {
262            return Some(msg);
263        }
264
265        loop {
266            tokio::select! {
267                Some(cmd) = self.cmd_rx.recv() => {
268                    match cmd {
269                        HandlerCommand::SetClient(client) => {
270                            log::debug!("Setting WebSocket client in handler");
271                            self.client = Some(client);
272                        }
273                        HandlerCommand::Disconnect => {
274                            log::debug!("Handler received disconnect command");
275
276                            if let Some(ref client) = self.client {
277                                client.disconnect().await;
278                            }
279                            self.signal.store(true, Ordering::SeqCst);
280                            return None;
281                        }
282                        HandlerCommand::Subscribe { subscriptions } => {
283                            for subscription in subscriptions {
284                                let key = subscription_to_key(&subscription);
285                                self.subscriptions.mark_subscribe(&key);
286
287                                let request = HyperliquidWsRequest::Subscribe { subscription };
288                                match serde_json::to_string(&request) {
289                                    Ok(payload) => {
290                                        log::debug!("Sending subscribe payload: {payload}");
291                                        if let Err(e) = self.send_with_retry(payload).await {
292                                            log::error!("Error subscribing to {key}: {e}");
293                                            self.subscriptions.mark_failure(&key);
294                                        }
295                                    }
296                                    Err(e) => {
297                                        log::error!("Error serializing subscription for {key}: {e}");
298                                        self.subscriptions.mark_failure(&key);
299                                    }
300                                }
301                            }
302                        }
303                        HandlerCommand::Unsubscribe { subscriptions } => {
304                            for subscription in subscriptions {
305                                let key = subscription_to_key(&subscription);
306                                self.subscriptions.mark_unsubscribe(&key);
307
308                                let request = HyperliquidWsRequest::Unsubscribe { subscription };
309                                match serde_json::to_string(&request) {
310                                    Ok(payload) => {
311                                        log::debug!("Sending unsubscribe payload: {payload}");
312                                        if let Err(e) = self.send_with_retry(payload).await {
313                                            log::error!("Error unsubscribing from {key}: {e}");
314                                        }
315                                    }
316                                    Err(e) => {
317                                        log::error!("Error serializing unsubscription for {key}: {e}");
318                                    }
319                                }
320                            }
321                        }
322                        HandlerCommand::Post { id, request } => {
323                            let request = HyperliquidWsRequest::Post { id, request };
324                            match serde_json::to_string(&request) {
325                                Ok(payload) => {
326                                    log::debug!("Sending post payload: id={id}");
327                                    if let Err(e) = self.send_with_retry(payload).await {
328                                        log::error!("Error sending post request id={id}: {e}");
329                                        self.post_router.cancel(id).await;
330                                    }
331                                }
332                                Err(e) => {
333                                    log::error!("Error serializing post request id={id}: {e}");
334                                    self.post_router.cancel(id).await;
335                                }
336                            }
337                        }
338                        HandlerCommand::InitializeInstruments(instruments) => {
339                            for inst in instruments {
340                                let coin = inst.raw_symbol().inner();
341                                self.instruments.insert(coin, inst);
342                            }
343                        }
344                        HandlerCommand::UpdateInstrument(inst) => {
345                            let coin = inst.raw_symbol().inner();
346                            self.instruments.insert(coin, inst);
347                        }
348                        HandlerCommand::AddBarType { key, bar_type } => {
349                            self.bar_types_cache.insert(key, bar_type);
350                        }
351                        HandlerCommand::RemoveBarType { key } => {
352                            self.bar_types_cache.remove(&key);
353                            self.bar_cache.remove(&key);
354                        }
355                        HandlerCommand::UpdateAssetContextSubs { coin, data_types } => {
356                            let previous_data_types = self.asset_context_subs.get(&coin).cloned();
357                            self.asset_context_caches.clear_removed(
358                                coin,
359                                previous_data_types.as_ref(),
360                                &data_types,
361                            );
362
363                            if data_types.is_empty() {
364                                self.asset_context_subs.remove(&coin);
365                            } else {
366                                self.asset_context_subs.insert(coin, data_types);
367                            }
368                        }
369                        HandlerCommand::UpdateTradeSubs { coin, uses } => {
370                            if uses.is_empty() {
371                                self.trade_subs.remove(&coin);
372                            } else {
373                                self.trade_subs.insert(coin, uses);
374                            }
375                        }
376                        HandlerCommand::CacheAllDexAssetCtxsInstrumentIds(mappings) => {
377                            self.all_dex_asset_ctxs_instrument_ids = mappings;
378                        }
379                        HandlerCommand::CacheSpotFillCoins(_) => {
380                            // No longer needed - raw_symbol now contains the proper format
381                        }
382                        HandlerCommand::SetDepth10Sub { coin, subscribed } => {
383                            if subscribed {
384                                self.depth10_subs.insert(coin);
385                            } else {
386                                self.depth10_subs.remove(&coin);
387                            }
388                        }
389                    }
390                }
391
392                Some(raw_msg) = self.raw_rx.recv() => {
393                    match raw_msg {
394                        Message::Text(text) => {
395                            if text == RECONNECTED {
396                                log::info!("Received RECONNECTED sentinel");
397                                return Some(NautilusWsMessage::Reconnected);
398                            }
399
400                            match serde_json::from_str::<HyperliquidWsMessage>(&text) {
401                                Ok(msg) => {
402                                    if let HyperliquidWsMessage::Post { data } = msg {
403                                        self.post_router.complete(data).await;
404                                        continue;
405                                    }
406
407                                    let ts_init = self.clock.get_time_ns();
408                                    let all_mids_data_types =
409                                        Self::all_mids_data_types(&self.subscriptions);
410
411                                    let nautilus_msgs = Self::parse_to_nautilus_messages(
412                                        msg,
413                                        &self.instruments,
414                                        &self.cloid_cache,
415                                        &self.bar_types_cache,
416                                        self.account_id,
417                                        ts_init,
418                                        &self.asset_context_subs,
419                                        &self.trade_subs,
420                                        &self.depth10_subs,
421                                        &mut self.processed_trade_ids,
422                                        &mut self.processed_public_trade_ids,
423                                        &mut self.asset_context_caches,
424                                        &mut self.bar_cache,
425                                        &self.all_dex_asset_ctxs_instrument_ids,
426                                        &all_mids_data_types,
427                                    );
428
429                                    if !nautilus_msgs.is_empty() {
430                                        let mut iter = nautilus_msgs.into_iter();
431                                        let first = iter.next().unwrap();
432                                        self.message_buffer.extend(iter);
433                                        return Some(first);
434                                    }
435                                }
436                                Err(e) => {
437                                    log::error!("Error parsing WebSocket message: {e}, text: {text}");
438                                }
439                            }
440                        }
441                        Message::Ping(data) => {
442                            if let Some(ref client) = self.client
443                                && let Err(e) = client.send_pong(data.to_vec()).await {
444                                log::error!("Error sending pong: {e}");
445                            }
446                        }
447                        Message::Close(_) => {
448                            log::debug!("Received WebSocket close frame");
449                            return None;
450                        }
451                        _ => {}
452                    }
453                }
454
455                else => {
456                    log::debug!("Handler shutting down: stream ended or command channel closed");
457                    return None;
458                }
459            }
460        }
461    }
462
463    #[expect(clippy::too_many_arguments)]
464    fn parse_to_nautilus_messages(
465        msg: HyperliquidWsMessage,
466        instruments: &AHashMap<Ustr, InstrumentAny>,
467        cloid_cache: &CloidCache,
468        bar_types: &AHashMap<String, BarType>,
469        account_id: Option<AccountId>,
470        ts_init: UnixNanos,
471        asset_context_subs: &AHashMap<Ustr, AHashSet<AssetContextDataType>>,
472        trade_subs: &AHashMap<Ustr, TradeStreamUses>,
473        depth10_subs: &AHashSet<Ustr>,
474        processed_trade_ids: &mut FifoCache<u64, 10_000>,
475        processed_public_trade_ids: &mut FifoCache<(Ustr, u64), 10_000>,
476        asset_context_caches: &mut AssetContextCaches,
477        bar_cache: &mut AHashMap<String, CandleData>,
478        all_dex_asset_ctxs_instrument_ids: &AHashMap<Ustr, Vec<Option<InstrumentId>>>,
479        all_mids_data_types: &[DataType],
480    ) -> Vec<NautilusWsMessage> {
481        let mut result = Vec::new();
482
483        match msg {
484            HyperliquidWsMessage::OrderUpdates { data } => {
485                if let Some(account_id) = account_id
486                    && let Some(msg) = Self::handle_order_updates(
487                        &data,
488                        instruments,
489                        cloid_cache,
490                        account_id,
491                        ts_init,
492                    )
493                {
494                    result.push(msg);
495                }
496            }
497            HyperliquidWsMessage::UserEvents { data } | HyperliquidWsMessage::User { data } => {
498                // Process fills from userEvents channel (userFills channel is redundant)
499                match data {
500                    WsUserEventData::Fills { fills } => {
501                        log::debug!("Received {} fill(s) from userEvents channel", fills.len());
502                        for fill in &fills {
503                            log::debug!(
504                                "Fill: oid={}, coin={}, side={:?}, sz={}, px={}",
505                                fill.oid,
506                                fill.coin,
507                                fill.side,
508                                fill.sz,
509                                fill.px
510                            );
511                        }
512
513                        if let Some(account_id) = account_id {
514                            log::debug!("Processing fills with account_id={account_id}");
515
516                            if let Some(msg) = Self::handle_user_fills(
517                                &fills,
518                                instruments,
519                                cloid_cache,
520                                account_id,
521                                ts_init,
522                                processed_trade_ids,
523                            ) {
524                                log::debug!("Successfully created fill message");
525                                result.push(msg);
526                            } else {
527                                log::debug!("handle_user_fills returned None (no new fills)");
528                            }
529                        } else {
530                            log::warn!("Cannot process fills: account_id is None");
531                        }
532                    }
533                    WsUserEventData::Liquidation { liquidation } => {
534                        log::warn!(
535                            "Liquidation event: lid={}, liquidator={}, liquidated_user={}, ntl_pos={}, account_value={}",
536                            liquidation.lid,
537                            liquidation.liquidator,
538                            liquidation.liquidated_user,
539                            liquidation.liquidated_ntl_pos,
540                            liquidation.liquidated_account_value,
541                        );
542                    }
543                    _ => {
544                        log::debug!("Received non-fill user event: {data:?}");
545                    }
546                }
547            }
548            HyperliquidWsMessage::UserFills { data } => {
549                // UserFills channel is redundant with userEvents, but handle it for
550                // backwards compatibility if explicitly subscribed
551                if let Some(account_id) = account_id
552                    && let Some(msg) = Self::handle_user_fills(
553                        &data.fills,
554                        instruments,
555                        cloid_cache,
556                        account_id,
557                        ts_init,
558                        processed_trade_ids,
559                    )
560                {
561                    result.push(msg);
562                }
563            }
564            HyperliquidWsMessage::Trades { data } => {
565                result.extend(Self::handle_trades(
566                    &data,
567                    instruments,
568                    trade_subs,
569                    processed_public_trade_ids,
570                    ts_init,
571                ));
572            }
573            HyperliquidWsMessage::AllMids { data } => {
574                let mut mids = std::collections::HashMap::with_capacity(
575                    data.mids.len().min(instruments.len()),
576                );
577
578                for (coin, mid_str) in &data.mids {
579                    if let Some(instrument) = instruments.get(coin) {
580                        match mid_str.parse::<Price>() {
581                            Ok(price) => {
582                                mids.insert(instrument.id(), price);
583                            }
584                            Err(e) => {
585                                log::warn!("Failed to parse mid price for {coin}: {e}");
586                            }
587                        }
588                    } else {
589                        log::debug!("No instrument found for coin: {coin}");
590                    }
591                }
592
593                if !mids.is_empty() {
594                    // Take instead of clone on the last subscriber
595                    let last_idx = all_mids_data_types.len().saturating_sub(1);
596                    for (i, data_type) in all_mids_data_types.iter().enumerate() {
597                        let mids_for_this = if i == last_idx {
598                            std::mem::take(&mut mids)
599                        } else {
600                            mids.clone()
601                        };
602                        let all_mids = HyperliquidAllMids::new(mids_for_this, ts_init, ts_init);
603                        result.push(NautilusWsMessage::CustomData(Data::Custom(
604                            CustomData::new(Arc::new(all_mids), data_type.clone()),
605                        )));
606                    }
607                }
608            }
609            HyperliquidWsMessage::AllDexsAssetCtxs { data } => {
610                if let Some(msg) = Self::handle_all_dexs_asset_ctxs(
611                    data,
612                    all_dex_asset_ctxs_instrument_ids,
613                    ts_init,
614                ) {
615                    result.push(msg);
616                }
617            }
618            HyperliquidWsMessage::Bbo { data } => {
619                if let Some(msg) = Self::handle_bbo(&data, instruments, ts_init) {
620                    result.push(msg);
621                }
622            }
623            HyperliquidWsMessage::L2Book { data } => {
624                result.extend(Self::handle_l2_book(
625                    &data,
626                    instruments,
627                    depth10_subs,
628                    ts_init,
629                ));
630            }
631            HyperliquidWsMessage::Candle { data } => {
632                if let Some(msg) =
633                    Self::handle_candle(&data, instruments, bar_types, bar_cache, ts_init)
634                {
635                    result.push(msg);
636                }
637            }
638            HyperliquidWsMessage::ActiveAssetCtx { data }
639            | HyperliquidWsMessage::ActiveSpotAssetCtx { data } => {
640                result.extend(Self::handle_asset_context(
641                    &data,
642                    instruments,
643                    asset_context_subs,
644                    asset_context_caches,
645                    ts_init,
646                ));
647            }
648            HyperliquidWsMessage::Error { data } => {
649                log::warn!("Received error from Hyperliquid WebSocket: {data}");
650            }
651            // Ignore other message types (subscription confirmations, etc)
652            _ => {}
653        }
654
655        result
656    }
657
658    fn handle_order_updates(
659        data: &[super::messages::WsOrderData],
660        instruments: &AHashMap<Ustr, InstrumentAny>,
661        cloid_cache: &CloidCache,
662        account_id: AccountId,
663        ts_init: UnixNanos,
664    ) -> Option<NautilusWsMessage> {
665        let mut exec_reports = Vec::new();
666
667        for order_update in data {
668            let instrument = instruments.get(&order_update.order.coin);
669
670            if let Some(instrument) = instrument {
671                match parse_ws_order_status_report(order_update, instrument, account_id, ts_init) {
672                    Ok(mut report) => {
673                        // Resolve cloid to real client_order_id if cached
674                        if let Some(cloid) = &order_update.order.cloid {
675                            let cloid_ustr = Ustr::from(cloid.as_str());
676                            let resolved = cloid_cache
677                                .lock()
678                                .expect(MUTEX_POISONED)
679                                .get(&cloid_ustr)
680                                .copied();
681
682                            if let Some(real_client_order_id) = resolved {
683                                log::debug!("Resolved cloid {cloid} -> {real_client_order_id}");
684                                report.client_order_id = Some(real_client_order_id);
685                            }
686                        }
687                        exec_reports.push(ExecutionReport::Order(report));
688                    }
689                    Err(e) => {
690                        log::error!("Error parsing order update: {e}");
691                    }
692                }
693            } else {
694                log::debug!("No instrument found for coin: {}", order_update.order.coin);
695            }
696        }
697
698        if exec_reports.is_empty() {
699            None
700        } else {
701            Some(NautilusWsMessage::ExecutionReports(exec_reports))
702        }
703    }
704
705    fn handle_user_fills(
706        fills: &[super::messages::WsFillData],
707        instruments: &AHashMap<Ustr, InstrumentAny>,
708        cloid_cache: &CloidCache,
709        account_id: AccountId,
710        ts_init: UnixNanos,
711        processed_trade_ids: &mut FifoCache<u64, 10_000>,
712    ) -> Option<NautilusWsMessage> {
713        let mut exec_reports = Vec::new();
714
715        for fill in fills {
716            if processed_trade_ids.contains(&fill.tid) {
717                log::debug!("Skipping duplicate fill: tid={}", fill.tid);
718                continue;
719            }
720
721            let instrument = instruments.get(&fill.coin);
722
723            if let Some(instrument) = instrument {
724                log::debug!("Found instrument for fill coin={}", fill.coin);
725                match parse_ws_fill_report(fill, instrument, account_id, ts_init) {
726                    Ok(mut report) => {
727                        // Mark processed only after successful parse
728                        processed_trade_ids.add(fill.tid);
729
730                        if let Some(cloid) = &fill.cloid {
731                            let cloid_ustr = Ustr::from(cloid.as_str());
732                            let resolved = cloid_cache
733                                .lock()
734                                .expect(MUTEX_POISONED)
735                                .get(&cloid_ustr)
736                                .copied();
737
738                            if let Some(real_client_order_id) = resolved {
739                                log::debug!(
740                                    "Resolved fill cloid {cloid} -> {real_client_order_id}"
741                                );
742                                report.client_order_id = Some(real_client_order_id);
743                            }
744                        }
745                        log::debug!(
746                            "Parsed fill report: venue_order_id={:?}, trade_id={:?}",
747                            report.venue_order_id,
748                            report.trade_id
749                        );
750                        exec_reports.push(ExecutionReport::Fill(report));
751                    }
752                    Err(e) => {
753                        log::error!("Error parsing fill: {e}");
754                    }
755                }
756            } else {
757                // Not marked as processed so fill is retried if instrument loads later
758                log::warn!("No instrument found for fill coin={}", fill.coin);
759            }
760        }
761
762        if exec_reports.is_empty() {
763            None
764        } else {
765            Some(NautilusWsMessage::ExecutionReports(exec_reports))
766        }
767    }
768
769    fn handle_trades(
770        data: &[super::messages::WsTradeData],
771        instruments: &AHashMap<Ustr, InstrumentAny>,
772        trade_subs: &AHashMap<Ustr, TradeStreamUses>,
773        processed_public_trade_ids: &mut FifoCache<(Ustr, u64), 10_000>,
774        ts_init: UnixNanos,
775    ) -> Vec<NautilusWsMessage> {
776        let mut trade_ticks = Vec::new();
777        let mut public_trades = Vec::new();
778
779        for trade in data {
780            if let Some(instrument) = instruments.get(&trade.coin) {
781                let uses = trade_subs.get(&trade.coin).copied().unwrap_or_default();
782
783                if uses.ticks {
784                    match parse_ws_trade_tick(trade, instrument, ts_init) {
785                        Ok(tick) => trade_ticks.push(tick),
786                        Err(e) => {
787                            log::error!("Error parsing trade tick: {e}");
788                        }
789                    }
790                }
791
792                if uses.public_trades {
793                    let trade_key = (trade.coin, trade.tid);
794                    if processed_public_trade_ids.contains(&trade_key) {
795                        log::debug!(
796                            "Skipping replayed public trade: coin={}, tid={}",
797                            trade.coin,
798                            trade.tid
799                        );
800                        continue;
801                    }
802
803                    match parse_ws_public_trade(trade, instrument, ts_init) {
804                        Ok(trade) => {
805                            processed_public_trade_ids.add(trade_key);
806                            public_trades.push(trade);
807                        }
808                        Err(e) => {
809                            log::error!("Error parsing public trade: {e}");
810                        }
811                    }
812                }
813            } else {
814                log::debug!("No instrument found for coin: {}", trade.coin);
815            }
816        }
817
818        let mut result = Vec::with_capacity(1 + public_trades.len());
819        if !trade_ticks.is_empty() {
820            result.push(NautilusWsMessage::Trades(trade_ticks));
821        }
822        result.extend(public_trades.into_iter().map(|trade| {
823            let instrument_id = trade.instrument_id;
824            NautilusWsMessage::CustomData(Data::Custom(CustomData::new(
825                Arc::new(trade),
826                Self::public_trade_data_type(instrument_id),
827            )))
828        }));
829        result
830    }
831
832    fn handle_bbo(
833        data: &super::messages::WsBboData,
834        instruments: &AHashMap<Ustr, InstrumentAny>,
835        ts_init: UnixNanos,
836    ) -> Option<NautilusWsMessage> {
837        if let Some(instrument) = instruments.get(&data.coin) {
838            match parse_ws_quote_tick(data, instrument, ts_init) {
839                Ok(quote_tick) => Some(NautilusWsMessage::Quote(quote_tick)),
840                Err(e) => {
841                    log::error!("Error parsing quote tick: {e}");
842                    None
843                }
844            }
845        } else {
846            log::debug!("No instrument found for coin: {}", data.coin);
847            None
848        }
849    }
850
851    fn handle_l2_book(
852        data: &super::messages::WsBookData,
853        instruments: &AHashMap<Ustr, InstrumentAny>,
854        depth10_subs: &AHashSet<Ustr>,
855        ts_init: UnixNanos,
856    ) -> Vec<NautilusWsMessage> {
857        let mut out = Vec::new();
858
859        let Some(instrument) = instruments.get(&data.coin) else {
860            log::debug!("No instrument found for coin: {}", data.coin);
861            return out;
862        };
863
864        match parse_ws_order_book_deltas(data, instrument, ts_init) {
865            Ok(deltas) => out.push(NautilusWsMessage::Deltas(deltas)),
866            Err(e) => log::error!("Error parsing order book deltas: {e}"),
867        }
868
869        if depth10_subs.contains(&data.coin) {
870            match parse_ws_order_book_depth10(data, instrument, ts_init) {
871                Ok(depth) => out.push(NautilusWsMessage::Depth10(Box::new(depth))),
872                Err(e) => log::error!("Error parsing order book depth10: {e}"),
873            }
874        }
875
876        out
877    }
878
879    fn handle_candle(
880        data: &CandleData,
881        instruments: &AHashMap<Ustr, InstrumentAny>,
882        bar_types: &AHashMap<String, BarType>,
883        bar_cache: &mut AHashMap<String, CandleData>,
884        ts_init: UnixNanos,
885    ) -> Option<NautilusWsMessage> {
886        let key = format!("candle:{}:{}", data.s, data.i);
887
888        let mut closed_bar = None;
889
890        if let Some(cached) = bar_cache.get(&key) {
891            // Emit cached bar when close_time changes, indicating the previous period closed
892            if cached.close_time != data.close_time {
893                log::debug!(
894                    "Bar period changed for {}: prev_close_time={}, new_close_time={}",
895                    data.s,
896                    cached.close_time,
897                    data.close_time
898                );
899                closed_bar = Some(cached.clone());
900            }
901        }
902
903        bar_cache.insert(key.clone(), data.clone());
904
905        if let Some(closed_data) = closed_bar {
906            if let Some(bar_type) = bar_types.get(&key) {
907                if let Some(instrument) = instruments.get(&data.s) {
908                    match parse_ws_candle(&closed_data, instrument, bar_type, ts_init) {
909                        Ok(bar) => return Some(NautilusWsMessage::Candle(bar)),
910                        Err(e) => {
911                            log::error!("Error parsing closed candle: {e}");
912                        }
913                    }
914                } else {
915                    log::debug!("No instrument found for coin: {}", data.s);
916                }
917            } else {
918                log::debug!("No bar type found for key: {key}");
919            }
920        }
921
922        None
923    }
924
925    fn handle_asset_context(
926        data: &WsActiveAssetCtxData,
927        instruments: &AHashMap<Ustr, InstrumentAny>,
928        asset_context_subs: &AHashMap<Ustr, AHashSet<AssetContextDataType>>,
929        asset_context_caches: &mut AssetContextCaches,
930        ts_init: UnixNanos,
931    ) -> Vec<NautilusWsMessage> {
932        let mut result = Vec::new();
933
934        let coin = match data {
935            WsActiveAssetCtxData::Perp { coin, .. } => coin,
936            WsActiveAssetCtxData::Spot { coin, .. } => coin,
937        };
938
939        if let Some(instrument) = instruments.get(coin) {
940            let (mark_px, oracle_px, funding, open_interest) = match data {
941                WsActiveAssetCtxData::Perp { ctx, .. } => (
942                    &ctx.shared.mark_px,
943                    Some(&ctx.oracle_px),
944                    Some(&ctx.funding),
945                    Some(&ctx.open_interest),
946                ),
947                WsActiveAssetCtxData::Spot { ctx, .. } => (&ctx.shared.mark_px, None, None, None),
948            };
949
950            let mark_changed = asset_context_caches.mark_price.get(coin) != Some(mark_px);
951            let index_changed =
952                oracle_px.is_some_and(|px| asset_context_caches.index_price.get(coin) != Some(px));
953            let funding_changed = funding
954                .is_some_and(|rate| asset_context_caches.funding_rate.get(coin) != Some(rate));
955            let open_interest_changed = open_interest
956                .is_some_and(|value| asset_context_caches.open_interest.get(coin) != Some(value));
957
958            let subscribed_types = asset_context_subs.get(coin);
959
960            if mark_changed || index_changed || funding_changed {
961                match parse_ws_asset_context(data, instrument, ts_init) {
962                    Ok((mark_price, index_price, funding_rate)) => {
963                        if mark_changed
964                            && subscribed_types
965                                .is_some_and(|s| s.contains(&AssetContextDataType::MarkPrice))
966                        {
967                            asset_context_caches.mark_price.insert(*coin, *mark_px);
968                            result.push(NautilusWsMessage::MarkPrice(mark_price));
969                        }
970
971                        if index_changed
972                            && subscribed_types
973                                .is_some_and(|s| s.contains(&AssetContextDataType::IndexPrice))
974                        {
975                            if let Some(px) = oracle_px {
976                                asset_context_caches.index_price.insert(*coin, *px);
977                            }
978
979                            if let Some(index) = index_price {
980                                result.push(NautilusWsMessage::IndexPrice(index));
981                            }
982                        }
983
984                        if funding_changed
985                            && subscribed_types
986                                .is_some_and(|s| s.contains(&AssetContextDataType::FundingRate))
987                        {
988                            if let Some(rate) = funding {
989                                asset_context_caches.funding_rate.insert(*coin, *rate);
990                            }
991
992                            if let Some(funding) = funding_rate {
993                                result.push(NautilusWsMessage::FundingRate(funding));
994                            }
995                        }
996                    }
997                    Err(e) => {
998                        log::error!("Error parsing asset context: {e}");
999                    }
1000                }
1001            }
1002
1003            if let Some(value) = open_interest
1004                && open_interest_changed
1005                && subscribed_types.is_some_and(|s| s.contains(&AssetContextDataType::OpenInterest))
1006            {
1007                match parse_ws_open_interest(*value, instrument, ts_init) {
1008                    Ok(open_interest_data) => {
1009                        asset_context_caches.open_interest.insert(*coin, *value);
1010
1011                        let data_type =
1012                            Self::open_interest_data_type(open_interest_data.instrument_id);
1013                        result.push(NautilusWsMessage::CustomData(Data::Custom(
1014                            CustomData::new(Arc::new(open_interest_data), data_type),
1015                        )));
1016                    }
1017                    Err(e) => {
1018                        log::error!("Error parsing open interest: {e}");
1019                    }
1020                }
1021            }
1022        } else {
1023            log::debug!("No instrument found for coin: {coin}");
1024        }
1025
1026        result
1027    }
1028
1029    fn handle_all_dexs_asset_ctxs(
1030        data: WsAllDexsAssetCtxsData,
1031        all_dex_asset_ctxs_instrument_ids: &AHashMap<Ustr, Vec<Option<InstrumentId>>>,
1032        ts_init: UnixNanos,
1033    ) -> Option<NautilusWsMessage> {
1034        let mut entries = Vec::new();
1035
1036        for (dex, ctxs) in data.ctxs {
1037            let dex_key = Ustr::from(dex.as_str());
1038            let Some(instrument_ids) = all_dex_asset_ctxs_instrument_ids.get(&dex_key) else {
1039                log::warn!("Missing Hyperliquid allDexsAssetCtxs mapping for dex='{dex}'");
1040                continue;
1041            };
1042
1043            if ctxs.len() != instrument_ids.len() {
1044                // Mapping is built once at bootstrap, so a count change means the universe
1045                // drifted and positional alignment can no longer be trusted.
1046                log::warn!(
1047                    "Hyperliquid allDexsAssetCtxs count mismatch for dex='{dex}': received {} contexts but cached {} instrument IDs (reconnect to refresh)",
1048                    ctxs.len(),
1049                    instrument_ids.len()
1050                );
1051            }
1052
1053            for (index, ctx) in ctxs.into_iter().enumerate() {
1054                let Some(Some(instrument_id)) = instrument_ids.get(index).copied() else {
1055                    log::warn!(
1056                        "Missing Hyperliquid allDexsAssetCtxs instrument mapping for dex='{dex}' index={index}"
1057                    );
1058                    continue;
1059                };
1060
1061                match Self::normalize_all_dex_asset_ctx_entry(&dex, instrument_id, ctx) {
1062                    Ok(entry) => entries.push(entry),
1063                    Err(e) => {
1064                        log::warn!(
1065                            "Failed to normalize Hyperliquid allDexsAssetCtxs entry dex='{dex}' index={index}: {e}"
1066                        );
1067                    }
1068                }
1069            }
1070        }
1071
1072        if entries.is_empty() {
1073            return None;
1074        }
1075
1076        let payload = HyperliquidAllDexsAssetCtxs::new(entries, ts_init, ts_init);
1077        let data_type = DataType::new("HyperliquidAllDexsAssetCtxs", None, None);
1078        Some(NautilusWsMessage::CustomData(Data::Custom(
1079            CustomData::new(Arc::new(payload), data_type),
1080        )))
1081    }
1082
1083    fn normalize_all_dex_asset_ctx_entry(
1084        dex: &str,
1085        instrument_id: InstrumentId,
1086        ctx: super::messages::PerpsAssetCtx,
1087    ) -> anyhow::Result<HyperliquidDexAssetCtx> {
1088        let mark_price = Price::from_decimal(ctx.shared.mark_px).map_err(anyhow::Error::msg)?;
1089        let oracle_price = Price::from_decimal(ctx.oracle_px).map_err(anyhow::Error::msg)?;
1090        let prev_day_price =
1091            Price::from_decimal(ctx.shared.prev_day_px).map_err(anyhow::Error::msg)?;
1092        let mid_price = ctx
1093            .shared
1094            .mid_px
1095            .map(|value| Price::from_decimal(value).map_err(anyhow::Error::msg))
1096            .transpose()?;
1097        let funding_rate = ctx.funding;
1098        let open_interest = ctx.open_interest;
1099        let premium = ctx.premium;
1100        let day_ntl_volume = ctx.shared.day_ntl_vlm;
1101        let day_base_volume = ctx
1102            .shared
1103            .day_base_vlm
1104            .ok_or_else(|| anyhow::anyhow!("missing dayBaseVlm"))?;
1105        let impact_prices = match ctx.shared.impact_pxs {
1106            Some(values) => match values.as_slice() {
1107                [bid, ask] => Some(HyperliquidImpactPrices {
1108                    bid: bid.parse::<Price>().map_err(anyhow::Error::msg)?,
1109                    ask: ask.parse::<Price>().map_err(anyhow::Error::msg)?,
1110                }),
1111                other => {
1112                    anyhow::bail!("expected 2 impact prices, received {}", other.len());
1113                }
1114            },
1115            None => None,
1116        };
1117
1118        Ok(HyperliquidDexAssetCtx {
1119            dex: dex.to_string(),
1120            instrument_id,
1121            mark_price,
1122            oracle_price,
1123            prev_day_price,
1124            mid_price,
1125            impact_prices,
1126            funding_rate,
1127            open_interest,
1128            premium,
1129            day_ntl_volume,
1130            day_base_volume,
1131        })
1132    }
1133
1134    fn all_mids_data_types(subscriptions: &SubscriptionState) -> Vec<DataType> {
1135        let mut topics = subscriptions.all_topics();
1136        topics.sort_unstable();
1137        topics.dedup();
1138
1139        let all_mids_channel = HyperliquidWsChannel::AllMids.as_str();
1140        let all_mids_prefix = format!("{all_mids_channel}:");
1141        let mut data_types = Vec::new();
1142
1143        for topic in topics {
1144            if topic == all_mids_channel {
1145                data_types.push(DataType::new("HyperliquidAllMids", None, None));
1146            } else if let Some(dex) = topic.strip_prefix(&all_mids_prefix) {
1147                let mut metadata = Params::new();
1148                metadata.insert(
1149                    "dex".to_string(),
1150                    serde_json::Value::String(dex.to_string()),
1151                );
1152                data_types.push(DataType::new("HyperliquidAllMids", Some(metadata), None));
1153            }
1154        }
1155
1156        if data_types.is_empty() {
1157            data_types.push(DataType::new("HyperliquidAllMids", None, None));
1158        }
1159
1160        data_types
1161    }
1162
1163    fn open_interest_data_type(instrument_id: InstrumentId) -> DataType {
1164        let mut metadata = Params::new();
1165        metadata.insert(
1166            "instrument_id".to_string(),
1167            serde_json::Value::String(instrument_id.to_string()),
1168        );
1169        DataType::new(
1170            "HyperliquidOpenInterest",
1171            Some(metadata),
1172            Some(instrument_id.to_string()),
1173        )
1174    }
1175
1176    fn public_trade_data_type(instrument_id: InstrumentId) -> DataType {
1177        let mut metadata = Params::new();
1178        metadata.insert(
1179            "instrument_id".to_string(),
1180            serde_json::Value::String(instrument_id.to_string()),
1181        );
1182        DataType::new(
1183            "HyperliquidPublicTrade",
1184            Some(metadata),
1185            Some(instrument_id.to_string()),
1186        )
1187    }
1188}
1189
1190pub(crate) fn subscription_to_key(sub: &SubscriptionRequest) -> String {
1191    match sub {
1192        SubscriptionRequest::AllMids { dex } => {
1193            if let Some(dex_name) = dex {
1194                format!("{}:{dex_name}", HyperliquidWsChannel::AllMids.as_str())
1195            } else {
1196                HyperliquidWsChannel::AllMids.as_str().to_string()
1197            }
1198        }
1199        SubscriptionRequest::AllDexsAssetCtxs => {
1200            HyperliquidWsChannel::AllDexsAssetCtxs.as_str().to_string()
1201        }
1202        SubscriptionRequest::Notification { user } => {
1203            format!("{}:{user}", HyperliquidWsChannel::Notification.as_str())
1204        }
1205        SubscriptionRequest::WebData2 { user } => {
1206            format!("{}:{user}", HyperliquidWsChannel::WebData2.as_str())
1207        }
1208        SubscriptionRequest::Candle { coin, interval } => {
1209            format!(
1210                "{}:{coin}:{}",
1211                HyperliquidWsChannel::Candle.as_str(),
1212                interval.as_str()
1213            )
1214        }
1215        SubscriptionRequest::L2Book { coin, .. } => {
1216            format!("{}:{coin}", HyperliquidWsChannel::L2Book.as_str())
1217        }
1218        SubscriptionRequest::Trades { coin } => {
1219            format!("{}:{coin}", HyperliquidWsChannel::Trades.as_str())
1220        }
1221        SubscriptionRequest::OrderUpdates { user } => {
1222            format!("{}:{user}", HyperliquidWsChannel::OrderUpdates.as_str())
1223        }
1224        SubscriptionRequest::UserEvents { user } => {
1225            format!("{}:{user}", HyperliquidWsChannel::UserEvents.as_str())
1226        }
1227        SubscriptionRequest::UserFills { user, .. } => {
1228            format!("{}:{user}", HyperliquidWsChannel::UserFills.as_str())
1229        }
1230        SubscriptionRequest::UserFundings { user } => {
1231            format!("{}:{user}", HyperliquidWsChannel::UserFundings.as_str())
1232        }
1233        SubscriptionRequest::UserNonFundingLedgerUpdates { user } => {
1234            format!(
1235                "{}:{user}",
1236                HyperliquidWsChannel::UserNonFundingLedgerUpdates.as_str()
1237            )
1238        }
1239        SubscriptionRequest::ActiveAssetCtx { coin } => {
1240            format!("{}:{coin}", HyperliquidWsChannel::ActiveAssetCtx.as_str())
1241        }
1242        SubscriptionRequest::ActiveSpotAssetCtx { coin } => {
1243            format!(
1244                "{}:{coin}",
1245                HyperliquidWsChannel::ActiveSpotAssetCtx.as_str()
1246            )
1247        }
1248        SubscriptionRequest::ActiveAssetData { user, coin } => {
1249            format!(
1250                "{}:{user}:{coin}",
1251                HyperliquidWsChannel::ActiveAssetData.as_str()
1252            )
1253        }
1254        SubscriptionRequest::UserTwapSliceFills { user } => {
1255            format!(
1256                "{}:{user}",
1257                HyperliquidWsChannel::UserTwapSliceFills.as_str()
1258            )
1259        }
1260        SubscriptionRequest::UserTwapHistory { user } => {
1261            format!("{}:{user}", HyperliquidWsChannel::UserTwapHistory.as_str())
1262        }
1263        SubscriptionRequest::Bbo { coin } => {
1264            format!("{}:{coin}", HyperliquidWsChannel::Bbo.as_str())
1265        }
1266    }
1267}
1268
1269/// Determines whether a Hyperliquid WebSocket error should trigger a retry.
1270pub(crate) fn should_retry_hyperliquid_error(error: &HyperliquidWsError) -> bool {
1271    match error {
1272        HyperliquidWsError::TungsteniteError(_) => true,
1273        HyperliquidWsError::ClientError(msg) => {
1274            let msg_lower = msg.to_lowercase();
1275            msg_lower.contains("timeout")
1276                || msg_lower.contains("timed out")
1277                || msg_lower.contains("connection")
1278                || msg_lower.contains("network")
1279        }
1280        _ => false,
1281    }
1282}
1283
1284/// Creates a timeout error for Hyperliquid retry logic.
1285pub(crate) fn create_hyperliquid_timeout_error(msg: String) -> HyperliquidWsError {
1286    HyperliquidWsError::ClientError(msg)
1287}
1288
1289#[cfg(test)]
1290mod tests {
1291    use std::{
1292        sync::{Arc, Mutex, atomic::AtomicBool},
1293        time::Duration,
1294    };
1295
1296    use ahash::{AHashMap, AHashSet};
1297    use nautilus_common::cache::fifo::FifoCacheMap;
1298    use nautilus_core::nanos::UnixNanos;
1299    use nautilus_model::{
1300        data::Data,
1301        identifiers::{ClientOrderId, InstrumentId, Symbol},
1302        instruments::{CryptoPerpetual, Instrument, InstrumentAny},
1303        types::{Currency, Price, Quantity},
1304    };
1305    use nautilus_network::websocket::SubscriptionState;
1306    use rstest::rstest;
1307    use rust_decimal::Decimal;
1308    use rust_decimal_macros::dec;
1309    use serde_json::json;
1310    use ustr::Ustr;
1311
1312    use super::{
1313        super::{
1314            client::{AssetContextDataType, CLOID_CACHE_CAPACITY, CloidCache},
1315            messages::{
1316                NautilusWsMessage, PerpsAssetCtx, PostRequest, SharedAssetCtx, SpotAssetCtx,
1317                WsActiveAssetCtxData, WsAllDexsAssetCtxsData, WsBookData, WsLevelData,
1318            },
1319            post::PostRouter,
1320        },
1321        AssetContextCaches, FeedHandler, HandlerCommand,
1322    };
1323    use crate::{
1324        common::consts::HYPERLIQUID_VENUE,
1325        data_types::{HyperliquidAllDexsAssetCtxs, HyperliquidOpenInterest},
1326    };
1327
1328    fn btc_perp() -> InstrumentAny {
1329        InstrumentAny::CryptoPerpetual(CryptoPerpetual::new(
1330            InstrumentId::new(Symbol::new("BTC-PERP"), *HYPERLIQUID_VENUE),
1331            Symbol::new("BTC-PERP"),
1332            Currency::from("BTC"),
1333            Currency::from("USDC"),
1334            Currency::from("USDC"),
1335            false,
1336            2,
1337            3,
1338            Price::from("0.01"),
1339            Quantity::from("0.001"),
1340            None,
1341            None,
1342            None,
1343            None,
1344            None,
1345            None,
1346            None,
1347            None,
1348            None,
1349            None,
1350            None,
1351            None,
1352            None,
1353            None,
1354            UnixNanos::default(),
1355            UnixNanos::default(),
1356        ))
1357    }
1358
1359    fn one_level_book() -> WsBookData {
1360        WsBookData {
1361            coin: Ustr::from("BTC"),
1362            levels: [
1363                vec![WsLevelData {
1364                    px: dec!(100.00),
1365                    sz: dec!(1.0),
1366                    n: 1,
1367                }],
1368                vec![WsLevelData {
1369                    px: dec!(100.01),
1370                    sz: dec!(1.0),
1371                    n: 1,
1372                }],
1373            ],
1374            time: 1_700_000_000_000,
1375        }
1376    }
1377
1378    fn btc_active_spot_asset_ctx() -> WsActiveAssetCtxData {
1379        WsActiveAssetCtxData::Spot {
1380            coin: Ustr::from("BTC"),
1381            ctx: SpotAssetCtx {
1382                shared: SharedAssetCtx {
1383                    day_ntl_vlm: dec!(1000000.0),
1384                    prev_day_px: dec!(49000.0),
1385                    mark_px: dec!(50000.0),
1386                    mid_px: Some(dec!(50001.0)),
1387                    impact_pxs: None,
1388                    day_base_vlm: Some(dec!(100.0)),
1389                },
1390                circulating_supply: dec!(19000000.0),
1391            },
1392        }
1393    }
1394
1395    fn btc_active_asset_ctx(open_interest: Decimal) -> WsActiveAssetCtxData {
1396        WsActiveAssetCtxData::Perp {
1397            coin: Ustr::from("BTC"),
1398            ctx: PerpsAssetCtx {
1399                shared: SharedAssetCtx {
1400                    day_ntl_vlm: dec!(1000000.0),
1401                    prev_day_px: dec!(49000.0),
1402                    mark_px: dec!(50000.0),
1403                    mid_px: Some(dec!(50001.0)),
1404                    impact_pxs: Some(vec!["50000.0".to_string(), "50002.0".to_string()]),
1405                    day_base_vlm: Some(dec!(100.0)),
1406                },
1407                funding: dec!(0.0001),
1408                open_interest,
1409                oracle_px: dec!(50005.0),
1410                premium: Some(dec!(-0.0001)),
1411            },
1412        }
1413    }
1414
1415    fn sample_all_dexs_asset_ctxs() -> WsAllDexsAssetCtxsData {
1416        let raw = include_str!("../../test_data/ws_all_dexs_asset_ctxs.json");
1417        let msg: super::super::messages::HyperliquidWsMessage =
1418            serde_json::from_str(raw).expect("expected valid allDexsAssetCtxs fixture");
1419
1420        let super::super::messages::HyperliquidWsMessage::AllDexsAssetCtxs { data } = msg else {
1421            panic!("expected allDexsAssetCtxs fixture message");
1422        };
1423
1424        let default_entry = data
1425            .ctxs
1426            .iter()
1427            .find(|(dex, _)| dex.is_empty())
1428            .and_then(|(dex, ctxs)| ctxs.first().cloned().map(|ctx| (dex.clone(), vec![ctx])))
1429            .expect("expected default dex sample");
1430        let xyz_entry = data
1431            .ctxs
1432            .iter()
1433            .find(|(dex, _)| dex == "xyz")
1434            .and_then(|(dex, ctxs)| ctxs.first().cloned().map(|ctx| (dex.clone(), vec![ctx])))
1435            .expect("expected xyz dex sample");
1436
1437        WsAllDexsAssetCtxsData {
1438            ctxs: vec![default_entry, xyz_entry],
1439        }
1440    }
1441
1442    #[tokio::test]
1443    async fn post_send_failure_cancels_router_waiter() {
1444        let signal = Arc::new(AtomicBool::new(false));
1445        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
1446        let (raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
1447        let (out_tx, _out_rx) = tokio::sync::mpsc::unbounded_channel();
1448        let post_router = PostRouter::new();
1449        let cloid_cache: CloidCache = Arc::new(Mutex::new(FifoCacheMap::<
1450            Ustr,
1451            ClientOrderId,
1452            CLOID_CACHE_CAPACITY,
1453        >::new()));
1454        let mut handler = FeedHandler::new(
1455            signal,
1456            cmd_rx,
1457            raw_rx,
1458            out_tx,
1459            None,
1460            SubscriptionState::new(':'),
1461            cloid_cache,
1462            Arc::clone(&post_router),
1463        );
1464
1465        let id = 99;
1466        let rx = post_router.register(id).await.unwrap();
1467
1468        let task = tokio::spawn(async move { handler.next().await });
1469
1470        cmd_tx
1471            .send(HandlerCommand::Post {
1472                id,
1473                request: PostRequest::Info {
1474                    payload: json!({"type": "userRateLimit", "user": "0x123"}),
1475                },
1476            })
1477            .unwrap();
1478        drop(cmd_tx);
1479        drop(raw_tx);
1480
1481        let closed = tokio::time::timeout(Duration::from_millis(100), rx)
1482            .await
1483            .expect("post waiter should close without waiting for post timeout");
1484        assert!(closed.is_err(), "post router cancel must close the waiter");
1485        let _rx = post_router
1486            .register(id)
1487            .await
1488            .expect("post id should be reusable after cancellation");
1489        assert!(task.await.unwrap().is_none());
1490    }
1491
1492    #[rstest]
1493    fn handle_l2_book_emits_deltas_only_when_not_in_depth10_subs() {
1494        let mut instruments = AHashMap::new();
1495        instruments.insert(Ustr::from("BTC"), btc_perp());
1496        let depth10_subs = AHashSet::<Ustr>::new();
1497
1498        let msgs = FeedHandler::handle_l2_book(
1499            &one_level_book(),
1500            &instruments,
1501            &depth10_subs,
1502            UnixNanos::default(),
1503        );
1504
1505        assert_eq!(msgs.len(), 1);
1506        assert!(matches!(msgs[0], NautilusWsMessage::Deltas(_)));
1507    }
1508
1509    #[rstest]
1510    fn handle_l2_book_emits_deltas_and_depth10_when_coin_in_subs() {
1511        let mut instruments = AHashMap::new();
1512        instruments.insert(Ustr::from("BTC"), btc_perp());
1513        let mut depth10_subs = AHashSet::<Ustr>::new();
1514        depth10_subs.insert(Ustr::from("BTC"));
1515
1516        let msgs = FeedHandler::handle_l2_book(
1517            &one_level_book(),
1518            &instruments,
1519            &depth10_subs,
1520            UnixNanos::default(),
1521        );
1522
1523        assert_eq!(msgs.len(), 2);
1524        assert!(matches!(msgs[0], NautilusWsMessage::Deltas(_)));
1525        assert!(matches!(msgs[1], NautilusWsMessage::Depth10(_)));
1526    }
1527
1528    #[rstest]
1529    fn handle_l2_book_returns_empty_when_instrument_unknown() {
1530        let instruments = AHashMap::<Ustr, InstrumentAny>::new();
1531        let depth10_subs = AHashSet::<Ustr>::new();
1532
1533        let msgs = FeedHandler::handle_l2_book(
1534            &one_level_book(),
1535            &instruments,
1536            &depth10_subs,
1537            UnixNanos::default(),
1538        );
1539
1540        assert!(msgs.is_empty());
1541    }
1542
1543    #[rstest]
1544    fn handle_asset_context_emits_open_interest_custom_data_when_subscribed() {
1545        let instrument = btc_perp();
1546        let instrument_id = instrument.id();
1547        let mut instruments = AHashMap::new();
1548        instruments.insert(Ustr::from("BTC"), instrument);
1549
1550        let mut asset_context_subs = AHashMap::new();
1551        asset_context_subs.insert(
1552            Ustr::from("BTC"),
1553            AHashSet::from_iter([AssetContextDataType::OpenInterest]),
1554        );
1555
1556        let mut asset_context_caches = AssetContextCaches::default();
1557
1558        let msgs = FeedHandler::handle_asset_context(
1559            &btc_active_asset_ctx(dec!(100000.0)),
1560            &instruments,
1561            &asset_context_subs,
1562            &mut asset_context_caches,
1563            UnixNanos::default(),
1564        );
1565
1566        assert_eq!(msgs.len(), 1);
1567
1568        match &msgs[0] {
1569            NautilusWsMessage::CustomData(Data::Custom(custom)) => {
1570                let open_interest = custom
1571                    .data
1572                    .as_any()
1573                    .downcast_ref::<HyperliquidOpenInterest>()
1574                    .expect("expected HyperliquidOpenInterest");
1575                assert_eq!(open_interest.instrument_id, instrument_id);
1576                assert_eq!(open_interest.open_interest.to_string(), "100000.0");
1577                assert_eq!(
1578                    custom
1579                        .data_type
1580                        .metadata()
1581                        .and_then(|metadata| metadata.get_str("instrument_id"))
1582                        .map(ToString::to_string),
1583                    Some(instrument_id.to_string()),
1584                );
1585            }
1586            other => panic!("unexpected message type: {other:?}"),
1587        }
1588    }
1589
1590    #[rstest]
1591    fn handle_all_dexs_asset_ctxs_emits_normalized_custom_data() {
1592        let mapping = AHashMap::from_iter([
1593            (
1594                Ustr::from(""),
1595                vec![Some(InstrumentId::from("BTC-USD-PERP.HYPERLIQUID"))],
1596            ),
1597            (
1598                Ustr::from("xyz"),
1599                vec![Some(InstrumentId::from("xyz:XYZ100-USD-PERP.HYPERLIQUID"))],
1600            ),
1601        ]);
1602
1603        let msg = FeedHandler::handle_all_dexs_asset_ctxs(
1604            sample_all_dexs_asset_ctxs(),
1605            &mapping,
1606            UnixNanos::default(),
1607        )
1608        .expect("expected custom data");
1609
1610        match msg {
1611            NautilusWsMessage::CustomData(Data::Custom(custom)) => {
1612                let payload = custom
1613                    .data
1614                    .as_any()
1615                    .downcast_ref::<HyperliquidAllDexsAssetCtxs>()
1616                    .expect("expected HyperliquidAllDexsAssetCtxs");
1617                assert_eq!(payload.entries.len(), 2);
1618                assert_eq!(
1619                    payload.entries[0].instrument_id,
1620                    InstrumentId::from("BTC-USD-PERP.HYPERLIQUID")
1621                );
1622                assert_eq!(payload.entries[1].dex, "xyz");
1623                assert_eq!(
1624                    payload.entries[1].instrument_id,
1625                    InstrumentId::from("xyz:XYZ100-USD-PERP.HYPERLIQUID")
1626                );
1627                assert_eq!(payload.entries[0].mark_price.to_string(), "77562.0");
1628                assert_eq!(payload.entries[1].day_base_volume.to_string(), "5135.2458");
1629            }
1630            other => panic!("expected custom data, found {other:?}"),
1631        }
1632    }
1633
1634    #[rstest]
1635    fn handle_all_dexs_asset_ctxs_preserves_index_alignment_when_mappings_are_missing() {
1636        let data = WsAllDexsAssetCtxsData {
1637            ctxs: vec![(
1638                String::new(),
1639                vec![
1640                    PerpsAssetCtx {
1641                        shared: SharedAssetCtx {
1642                            day_ntl_vlm: dec!(1516669192.1953897476),
1643                            prev_day_px: dec!(76317.0),
1644                            mark_px: dec!(77562.0),
1645                            mid_px: Some(dec!(77558.5)),
1646                            impact_pxs: Some(vec!["77558.0".to_string(), "77559.0".to_string()]),
1647                            day_base_vlm: Some(dec!(19707.77457)),
1648                        },
1649                        funding: dec!(-0.0000015186),
1650                        open_interest: dec!(27353.17682),
1651                        oracle_px: dec!(77605.0),
1652                        premium: Some(dec!(-0.0005927453)),
1653                    },
1654                    PerpsAssetCtx {
1655                        shared: SharedAssetCtx {
1656                            day_ntl_vlm: dec!(591989409.9392402172),
1657                            prev_day_px: dec!(2094.6),
1658                            mark_px: dec!(2123.7),
1659                            mid_px: Some(dec!(2123.95)),
1660                            impact_pxs: Some(vec!["2123.65".to_string(), "2124.0".to_string()]),
1661                            day_base_vlm: Some(dec!(281686.8234999999)),
1662                        },
1663                        funding: dec!(0.0000125),
1664                        open_interest: dec!(605822.2557999999),
1665                        oracle_px: dec!(2124.6),
1666                        premium: Some(dec!(-0.0002824061)),
1667                    },
1668                ],
1669            )],
1670        };
1671
1672        let mapping = AHashMap::from_iter([(
1673            Ustr::from(""),
1674            vec![None, Some(InstrumentId::from("ETH-USD-PERP.HYPERLIQUID"))],
1675        )]);
1676
1677        let msg = FeedHandler::handle_all_dexs_asset_ctxs(data, &mapping, UnixNanos::default())
1678            .expect("expected custom data");
1679
1680        match msg {
1681            NautilusWsMessage::CustomData(Data::Custom(custom)) => {
1682                let payload = custom
1683                    .data
1684                    .as_any()
1685                    .downcast_ref::<HyperliquidAllDexsAssetCtxs>()
1686                    .expect("expected HyperliquidAllDexsAssetCtxs");
1687                assert_eq!(payload.entries.len(), 1);
1688                assert_eq!(
1689                    payload.entries[0].instrument_id,
1690                    InstrumentId::from("ETH-USD-PERP.HYPERLIQUID")
1691                );
1692                assert_eq!(payload.entries[0].mark_price.to_string(), "2123.7");
1693            }
1694            other => panic!("expected custom data, found {other:?}"),
1695        }
1696    }
1697
1698    #[rstest]
1699    fn handle_asset_context_skips_open_interest_for_spot_payload() {
1700        let instrument = btc_perp();
1701        let mut instruments = AHashMap::new();
1702        instruments.insert(Ustr::from("BTC"), instrument);
1703
1704        let mut asset_context_subs = AHashMap::new();
1705        asset_context_subs.insert(
1706            Ustr::from("BTC"),
1707            AHashSet::from_iter([AssetContextDataType::OpenInterest]),
1708        );
1709
1710        let mut asset_context_caches = AssetContextCaches::default();
1711
1712        let msgs = FeedHandler::handle_asset_context(
1713            &btc_active_spot_asset_ctx(),
1714            &instruments,
1715            &asset_context_subs,
1716            &mut asset_context_caches,
1717            UnixNanos::default(),
1718        );
1719
1720        assert!(msgs.is_empty());
1721        assert!(asset_context_caches.open_interest.is_empty());
1722    }
1723
1724    #[rstest]
1725    fn handle_asset_context_suppresses_unchanged_open_interest() {
1726        let instrument = btc_perp();
1727        let mut instruments = AHashMap::new();
1728        instruments.insert(Ustr::from("BTC"), instrument);
1729
1730        let mut asset_context_subs = AHashMap::new();
1731        asset_context_subs.insert(
1732            Ustr::from("BTC"),
1733            AHashSet::from_iter([AssetContextDataType::OpenInterest]),
1734        );
1735
1736        let mut asset_context_caches = AssetContextCaches::default();
1737
1738        let first = FeedHandler::handle_asset_context(
1739            &btc_active_asset_ctx(dec!(100000.0)),
1740            &instruments,
1741            &asset_context_subs,
1742            &mut asset_context_caches,
1743            UnixNanos::default(),
1744        );
1745        let second = FeedHandler::handle_asset_context(
1746            &btc_active_asset_ctx(dec!(100000.0)),
1747            &instruments,
1748            &asset_context_subs,
1749            &mut asset_context_caches,
1750            UnixNanos::default(),
1751        );
1752
1753        assert_eq!(first.len(), 1);
1754        assert!(second.is_empty());
1755    }
1756
1757    #[rstest]
1758    fn asset_context_caches_clear_removed_data_types() {
1759        let coin = Ustr::from("BTC");
1760        let mut caches = AssetContextCaches::default();
1761        caches.mark_price.insert(coin, dec!(98455.5));
1762        caches.index_price.insert(coin, dec!(98460.0));
1763        caches.funding_rate.insert(coin, dec!(0.0001));
1764        caches.open_interest.insert(coin, dec!(1500.0));
1765
1766        let previous_data_types = AHashSet::from_iter([
1767            AssetContextDataType::MarkPrice,
1768            AssetContextDataType::IndexPrice,
1769            AssetContextDataType::FundingRate,
1770            AssetContextDataType::OpenInterest,
1771        ]);
1772        let next_data_types = AHashSet::from_iter([
1773            AssetContextDataType::MarkPrice,
1774            AssetContextDataType::FundingRate,
1775        ]);
1776
1777        caches.clear_removed(coin, Some(&previous_data_types), &next_data_types);
1778
1779        assert_eq!(caches.mark_price.get(&coin).copied(), Some(dec!(98455.5)));
1780        assert!(caches.index_price.get(&coin).is_none());
1781        assert_eq!(caches.funding_rate.get(&coin).copied(), Some(dec!(0.0001)));
1782        assert!(caches.open_interest.get(&coin).is_none());
1783    }
1784}
nautilus_hyperliquid/websocket/handler.rs

nautilus_hyperliquid/websocket/
handler.rs
