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

nautilus_hyperliquid/websocket/
handler.rs
