nautilus_deribit/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 Deribit.
17//!
18//! The handler runs in a dedicated Tokio task as the I/O boundary between the client
19//! orchestrator and the network layer. It exclusively owns the `WebSocketClient` and
20//! processes commands from the client via an unbounded channel.
21
22use std::{
23 collections::VecDeque,
24 sync::{
25 Arc, Mutex,
26 atomic::{AtomicBool, AtomicU64, Ordering},
27 },
28};
29
30use ahash::AHashMap;
31use nautilus_common::cache::fifo::FifoCache;
32use nautilus_core::{AtomicSet, AtomicTime, UUID4, UnixNanos, time::get_atomic_clock_realtime};
33use nautilus_model::{
34 data::{Bar, CustomData, Data, DataType, InstrumentStatus},
35 enums::MarketStatusAction,
36 events::{AccountState, OrderCancelRejected, OrderModifyRejected, OrderRejected},
37 identifiers::{
38 AccountId, ClientOrderId, InstrumentId, StrategyId, Symbol, TraderId, VenueOrderId,
39 },
40 instruments::{Instrument, InstrumentAny},
41};
42use nautilus_network::{
43 RECONNECTED,
44 retry::{RetryManager, create_websocket_retry_manager},
45 websocket::{AuthTracker, SubscriptionState, WebSocketClient},
46};
47use tokio_tungstenite::tungstenite::Message;
48use ustr::Ustr;
49
50use super::{
51 enums::{DeribitBookMsgType, DeribitHeartbeatType, DeribitWsChannel, DeribitWsMethod},
52 error::DeribitWsError,
53 messages::{
54 DeribitAuthResult, DeribitBookMsg, DeribitCancelAllByInstrumentParams, DeribitCancelParams,
55 DeribitChartMsg, DeribitEditParams, DeribitHeartbeatParams, DeribitInstrumentStateMsg,
56 DeribitJsonRpcRequest, DeribitOrderMsg, DeribitOrderParams, DeribitOrderResponse,
57 DeribitPerpetualMsg, DeribitPortfolioMsg, DeribitQuoteMsg, DeribitSubscribeParams,
58 DeribitTickerMsg, DeribitTradeMsg, DeribitUserTradeMsg, DeribitVolatilityIndexMsg,
59 DeribitWsMessage, NautilusWsMessage, parse_raw_message,
60 },
61 parse::{
62 OrderEventType, determine_order_event_type, parse_book_msg, parse_chart_msg,
63 parse_order_accepted, parse_order_canceled, parse_order_expired, parse_order_updated,
64 parse_perpetual_to_funding_rate, parse_quote_msg, parse_ticker_to_index_price,
65 parse_ticker_to_mark_price, parse_ticker_to_option_greeks, parse_trades_data,
66 parse_user_order_msg, parse_user_trade_msg, resolution_to_bar_type,
67 },
68};
69use crate::{
70 common::{
71 consts::{DERIBIT_POST_ONLY_ERROR_CODE, DERIBIT_RATE_LIMIT_KEY_ORDER, DERIBIT_VENUE},
72 enums::DeribitInstrumentState,
73 parse::{parse_portfolio_to_account_state, use_cost_for_bar_volume},
74 },
75 data_types::DeribitVolatilityIndex,
76};
77
78/// Type of pending request for request ID correlation.
79#[derive(Debug, Clone)]
80pub enum PendingRequestType {
81 /// Authentication request.
82 Authenticate,
83 /// Subscribe request with requested channels.
84 Subscribe { channels: Vec<String> },
85 /// Unsubscribe request with requested channels.
86 Unsubscribe { channels: Vec<String> },
87 /// Set heartbeat request.
88 SetHeartbeat,
89 /// Test/ping request (heartbeat response).
90 Test,
91 /// Buy order request.
92 Buy {
93 client_order_id: ClientOrderId,
94 trader_id: TraderId,
95 strategy_id: StrategyId,
96 instrument_id: InstrumentId,
97 },
98 /// Sell order request.
99 Sell {
100 client_order_id: ClientOrderId,
101 trader_id: TraderId,
102 strategy_id: StrategyId,
103 instrument_id: InstrumentId,
104 },
105 /// Edit order request.
106 Edit {
107 client_order_id: ClientOrderId,
108 trader_id: TraderId,
109 strategy_id: StrategyId,
110 instrument_id: InstrumentId,
111 },
112 /// Cancel order request.
113 Cancel {
114 client_order_id: ClientOrderId,
115 trader_id: TraderId,
116 strategy_id: StrategyId,
117 instrument_id: InstrumentId,
118 },
119 /// Cancel all orders by instrument request.
120 CancelAllByInstrument { instrument_id: InstrumentId },
121 /// Get order state request.
122 GetOrderState {
123 client_order_id: ClientOrderId,
124 trader_id: TraderId,
125 strategy_id: StrategyId,
126 instrument_id: InstrumentId,
127 },
128}
129
130/// Commands sent from the client to the handler.
131#[allow(missing_debug_implementations)]
132pub enum HandlerCommand {
133 /// Set the active WebSocket client.
134 SetClient(WebSocketClient),
135 /// Disconnect the WebSocket.
136 Disconnect,
137 /// Authenticate with credentials.
138 Authenticate {
139 /// Serialized auth params (DeribitAuthParams or DeribitRefreshTokenParams).
140 auth_params: serde_json::Value,
141 },
142 /// Enable heartbeat with interval.
143 SetHeartbeat { interval: u64 },
144 /// Initialize the instrument cache.
145 InitializeInstruments(Vec<InstrumentAny>),
146 /// Update a single instrument in the cache.
147 UpdateInstrument(Box<InstrumentAny>),
148 /// Subscribe to channels.
149 Subscribe { channels: Vec<String> },
150 /// Unsubscribe from channels.
151 Unsubscribe { channels: Vec<String> },
152 /// Submit a buy order.
153 Buy {
154 params: DeribitOrderParams,
155 client_order_id: ClientOrderId,
156 trader_id: TraderId,
157 strategy_id: StrategyId,
158 instrument_id: InstrumentId,
159 },
160 /// Submit a sell order.
161 Sell {
162 params: DeribitOrderParams,
163 client_order_id: ClientOrderId,
164 trader_id: TraderId,
165 strategy_id: StrategyId,
166 instrument_id: InstrumentId,
167 },
168 /// Edit an existing order.
169 Edit {
170 params: DeribitEditParams,
171 client_order_id: ClientOrderId,
172 trader_id: TraderId,
173 strategy_id: StrategyId,
174 instrument_id: InstrumentId,
175 },
176 /// Cancel an existing order.
177 Cancel {
178 params: DeribitCancelParams,
179 client_order_id: ClientOrderId,
180 trader_id: TraderId,
181 strategy_id: StrategyId,
182 instrument_id: InstrumentId,
183 },
184 /// Cancel all orders by instrument.
185 CancelAllByInstrument {
186 params: DeribitCancelAllByInstrumentParams,
187 instrument_id: InstrumentId,
188 },
189 /// Get order state.
190 GetOrderState {
191 order_id: String,
192 client_order_id: ClientOrderId,
193 trader_id: TraderId,
194 strategy_id: StrategyId,
195 instrument_id: InstrumentId,
196 },
197}
198
199/// Context for an order submitted via this handler.
200///
201/// Stores the original trader/strategy/client IDs from the buy/sell command
202/// so they can be used when processing user.orders subscription updates.
203#[derive(Debug, Clone)]
204pub struct OrderContext {
205 pub client_order_id: ClientOrderId,
206 pub trader_id: TraderId,
207 pub strategy_id: StrategyId,
208 pub instrument_id: InstrumentId,
209}
210
211/// Deribit WebSocket feed handler.
212///
213/// Runs in a dedicated Tokio task, processing commands and raw WebSocket messages.
214#[allow(missing_debug_implementations)]
215pub struct DeribitWsFeedHandler {
216 clock: &'static AtomicTime,
217 signal: Arc<AtomicBool>,
218 inner: Option<WebSocketClient>,
219 cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
220 raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
221 out_tx: tokio::sync::mpsc::UnboundedSender<NautilusWsMessage>,
222 auth_tracker: AuthTracker,
223 subscriptions_state: SubscriptionState,
224 retry_manager: RetryManager<DeribitWsError>,
225 instruments_cache: AHashMap<Ustr, InstrumentAny>,
226 option_greeks_subs: Arc<AtomicSet<InstrumentId>>,
227 mark_price_subs: Arc<AtomicSet<InstrumentId>>,
228 index_price_subs: Arc<AtomicSet<InstrumentId>>,
229 request_id_counter: AtomicU64,
230 pending_requests: AHashMap<u64, PendingRequestType>,
231 account_id: Option<AccountId>,
232 order_contexts: AHashMap<VenueOrderId, OrderContext>,
233 emitted_accepted: FifoCache<VenueOrderId, 10_000>,
234 terminal_orders: FifoCache<ClientOrderId, 10_000>,
235 pending_bars: AHashMap<String, Bar>,
236 bars_timestamp_on_close: bool,
237 last_account_states: AHashMap<String, AccountState>,
238 book_sequence: AHashMap<Ustr, u64>,
239 pending_book_resync: Vec<String>,
240 pending_outgoing: VecDeque<NautilusWsMessage>,
241 subscribe_errors: Arc<Mutex<Vec<String>>>,
242}
243
244impl DeribitWsFeedHandler {
245 /// Creates a new feed handler.
246 #[expect(clippy::too_many_arguments)]
247 #[must_use]
248 pub fn new(
249 signal: Arc<AtomicBool>,
250 cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
251 raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
252 out_tx: tokio::sync::mpsc::UnboundedSender<NautilusWsMessage>,
253 auth_tracker: AuthTracker,
254 subscriptions_state: SubscriptionState,
255 option_greeks_subs: Arc<AtomicSet<InstrumentId>>,
256 mark_price_subs: Arc<AtomicSet<InstrumentId>>,
257 index_price_subs: Arc<AtomicSet<InstrumentId>>,
258 account_id: Option<AccountId>,
259 bars_timestamp_on_close: bool,
260 subscribe_errors: Arc<Mutex<Vec<String>>>,
261 ) -> Self {
262 Self {
263 clock: get_atomic_clock_realtime(),
264 signal,
265 inner: None,
266 cmd_rx,
267 raw_rx,
268 out_tx,
269 auth_tracker,
270 subscriptions_state,
271 retry_manager: create_websocket_retry_manager(),
272 instruments_cache: AHashMap::new(),
273 option_greeks_subs,
274 mark_price_subs,
275 index_price_subs,
276 request_id_counter: AtomicU64::new(1),
277 pending_requests: AHashMap::new(),
278 account_id,
279 order_contexts: AHashMap::new(),
280 emitted_accepted: FifoCache::new(),
281 terminal_orders: FifoCache::new(),
282 pending_bars: AHashMap::new(),
283 bars_timestamp_on_close,
284 last_account_states: AHashMap::new(),
285 book_sequence: AHashMap::new(),
286 pending_book_resync: Vec::new(),
287 pending_outgoing: VecDeque::new(),
288 subscribe_errors,
289 }
290 }
291
292 /// Sets the account ID for order/fill reports.
293 pub fn set_account_id(&mut self, account_id: AccountId) {
294 self.account_id = Some(account_id);
295 }
296
297 /// Returns the account ID.
298 #[must_use]
299 pub fn account_id(&self) -> Option<AccountId> {
300 self.account_id
301 }
302
303 fn clear_state(&mut self) {
304 let pending_count = self.pending_requests.len();
305 let emitted_count = self.emitted_accepted.len();
306 let bars_count = self.pending_bars.len();
307 let account_count = self.last_account_states.len();
308 let book_count = self.book_sequence.len();
309 let outgoing_count = self.pending_outgoing.len();
310
311 self.pending_requests.clear();
312 self.emitted_accepted.clear();
313 self.pending_bars.clear();
314 self.last_account_states.clear();
315 self.book_sequence.clear();
316 self.pending_book_resync.clear();
317 self.pending_outgoing.clear();
318
319 log::debug!(
320 "Reset state: pending_requests={pending_count}, emitted_accepted={emitted_count}, \
321 pending_bars={bars_count}, account_states={account_count}, book_sequence={book_count}, \
322 pending_outgoing={outgoing_count}"
323 );
324 }
325
326 /// Generates a unique request ID.
327 fn next_request_id(&self) -> u64 {
328 self.request_id_counter.fetch_add(1, Ordering::Relaxed)
329 }
330
331 /// Returns the current timestamp.
332 fn ts_init(&self) -> UnixNanos {
333 self.clock.get_time_ns()
334 }
335
336 /// Checks if there's a pending buy/sell request for the given client_order_id.
337 ///
338 /// This is used to avoid emitting duplicate OrderAccepted events from the
339 /// user.orders subscription when the response path will also emit an event.
340 fn is_pending_order(&self, client_order_id: &ClientOrderId) -> bool {
341 self.pending_requests.values().any(|req| match req {
342 PendingRequestType::Buy {
343 client_order_id: id,
344 ..
345 }
346 | PendingRequestType::Sell {
347 client_order_id: id,
348 ..
349 } => id == client_order_id,
350 _ => false,
351 })
352 }
353
354 /// Gets the OrderContext from a pending buy/sell request by client_order_id.
355 ///
356 /// Returns None if no pending request found.
357 fn get_pending_order_context(&self, client_order_id: &ClientOrderId) -> Option<OrderContext> {
358 for req in self.pending_requests.values() {
359 match req {
360 PendingRequestType::Buy {
361 client_order_id: id,
362 trader_id,
363 strategy_id,
364 instrument_id,
365 }
366 | PendingRequestType::Sell {
367 client_order_id: id,
368 trader_id,
369 strategy_id,
370 instrument_id,
371 } if id == client_order_id => {
372 return Some(OrderContext {
373 client_order_id: *id,
374 trader_id: *trader_id,
375 strategy_id: *strategy_id,
376 instrument_id: *instrument_id,
377 });
378 }
379 _ => {}
380 }
381 }
382 None
383 }
384
385 async fn send_tracked_request(
386 &mut self,
387 request_id: u64,
388 payload: Result<String, DeribitWsError>,
389 rate_limit_keys: Option<&[Ustr]>,
390 ) -> Result<(), DeribitWsError> {
391 let payload = match payload {
392 Ok(p) => p,
393 Err(e) => {
394 self.pending_requests.remove(&request_id);
395 return Err(e);
396 }
397 };
398 self.send_with_retry(payload, rate_limit_keys).await
399 }
400
401 /// Sends a message over the WebSocket with retry logic.
402 async fn send_with_retry(
403 &self,
404 payload: String,
405 rate_limit_keys: Option<&[Ustr]>,
406 ) -> Result<(), DeribitWsError> {
407 if let Some(client) = &self.inner {
408 let keys_owned: Option<Vec<Ustr>> = rate_limit_keys.map(|k| k.to_vec());
409 self.retry_manager
410 .execute_with_retry(
411 "websocket_send",
412 || {
413 let payload = payload.clone();
414 let keys = keys_owned.clone();
415 async move {
416 client
417 .send_text(payload, keys.as_deref())
418 .await
419 .map_err(|e| DeribitWsError::Send(e.to_string()))
420 }
421 },
422 |e| matches!(e, DeribitWsError::Send(_)),
423 DeribitWsError::Timeout,
424 )
425 .await
426 } else {
427 Err(DeribitWsError::NotConnected)
428 }
429 }
430
431 /// Handles a subscribe command.
432 ///
433 /// Note: The client has already called `mark_subscribe` before sending this command.
434 async fn handle_subscribe(&mut self, channels: Vec<String>) -> Result<(), DeribitWsError> {
435 let request_id = self.next_request_id();
436
437 // Track this request for response correlation
438 self.pending_requests.insert(
439 request_id,
440 PendingRequestType::Subscribe {
441 channels: channels.clone(),
442 },
443 );
444
445 // Deribit requires private/subscribe for authenticated channels
446 let method = if channels
447 .iter()
448 .any(|ch| DeribitWsChannel::requires_auth(ch))
449 {
450 DeribitWsMethod::PrivateSubscribe
451 } else {
452 DeribitWsMethod::PublicSubscribe
453 };
454
455 let request = DeribitJsonRpcRequest::new(
456 request_id,
457 method.as_method_str(),
458 DeribitSubscribeParams {
459 channels: channels.clone(),
460 },
461 );
462
463 let payload =
464 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
465
466 log::debug!("Subscribing to channels: request_id={request_id}, channels={channels:?}");
467 self.send_tracked_request(request_id, payload, None).await
468 }
469
470 /// Handles an unsubscribe command.
471 async fn handle_unsubscribe(&mut self, channels: Vec<String>) -> Result<(), DeribitWsError> {
472 let request_id = self.next_request_id();
473
474 // Track this request for response correlation
475 self.pending_requests.insert(
476 request_id,
477 PendingRequestType::Unsubscribe {
478 channels: channels.clone(),
479 },
480 );
481
482 let method = if channels
483 .iter()
484 .any(|ch| DeribitWsChannel::requires_auth(ch))
485 {
486 DeribitWsMethod::PrivateUnsubscribe
487 } else {
488 DeribitWsMethod::PublicUnsubscribe
489 };
490
491 let request = DeribitJsonRpcRequest::new(
492 request_id,
493 method.as_method_str(),
494 DeribitSubscribeParams {
495 channels: channels.clone(),
496 },
497 );
498
499 let payload =
500 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
501
502 log::debug!("Unsubscribing from channels: request_id={request_id}, channels={channels:?}");
503 self.send_tracked_request(request_id, payload, None).await
504 }
505
506 /// Handles enabling heartbeat.
507 async fn handle_set_heartbeat(&mut self, interval: u64) -> Result<(), DeribitWsError> {
508 let request_id = self.next_request_id();
509
510 // Track this request for response correlation
511 self.pending_requests
512 .insert(request_id, PendingRequestType::SetHeartbeat);
513
514 let request = DeribitJsonRpcRequest::new(
515 request_id,
516 DeribitWsMethod::SetHeartbeat.as_method_str(),
517 DeribitHeartbeatParams { interval },
518 );
519
520 let payload =
521 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
522
523 log::debug!(
524 "Enabling heartbeat with interval: request_id={request_id}, interval={interval} seconds"
525 );
526 self.send_tracked_request(request_id, payload, None).await
527 }
528
529 /// Responds to a heartbeat test_request.
530 async fn handle_heartbeat_test_request(&mut self) -> Result<(), DeribitWsError> {
531 let request_id = self.next_request_id();
532
533 // Track this request for response correlation
534 self.pending_requests
535 .insert(request_id, PendingRequestType::Test);
536
537 let request = DeribitJsonRpcRequest::new(
538 request_id,
539 DeribitWsMethod::Test.as_method_str(),
540 serde_json::json!({}),
541 );
542
543 let payload =
544 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
545
546 log::trace!("Responding to heartbeat test_request: request_id={request_id}");
547 self.send_tracked_request(request_id, payload, None).await
548 }
549
550 /// Handles a buy order command.
551 async fn handle_buy(
552 &mut self,
553 params: DeribitOrderParams,
554 client_order_id: ClientOrderId,
555 trader_id: TraderId,
556 strategy_id: StrategyId,
557 instrument_id: InstrumentId,
558 ) -> Result<(), DeribitWsError> {
559 let request_id = self.next_request_id();
560
561 self.pending_requests.insert(
562 request_id,
563 PendingRequestType::Buy {
564 client_order_id,
565 trader_id,
566 strategy_id,
567 instrument_id,
568 },
569 );
570
571 let request =
572 DeribitJsonRpcRequest::new(request_id, DeribitWsMethod::Buy.as_method_str(), params);
573
574 let payload =
575 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
576
577 log::debug!("Sending buy order: request_id={request_id}");
578 self.send_tracked_request(
579 request_id,
580 payload,
581 Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()),
582 )
583 .await
584 }
585
586 /// Handles a sell order command.
587 async fn handle_sell(
588 &mut self,
589 params: DeribitOrderParams,
590 client_order_id: ClientOrderId,
591 trader_id: TraderId,
592 strategy_id: StrategyId,
593 instrument_id: InstrumentId,
594 ) -> Result<(), DeribitWsError> {
595 let request_id = self.next_request_id();
596
597 self.pending_requests.insert(
598 request_id,
599 PendingRequestType::Sell {
600 client_order_id,
601 trader_id,
602 strategy_id,
603 instrument_id,
604 },
605 );
606
607 let request =
608 DeribitJsonRpcRequest::new(request_id, DeribitWsMethod::Sell.as_method_str(), params);
609
610 let payload =
611 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
612
613 log::debug!("Sending sell order: request_id={request_id}");
614 self.send_tracked_request(
615 request_id,
616 payload,
617 Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()),
618 )
619 .await
620 }
621
622 /// Handles an edit order command.
623 async fn handle_edit(
624 &mut self,
625 params: DeribitEditParams,
626 client_order_id: ClientOrderId,
627 trader_id: TraderId,
628 strategy_id: StrategyId,
629 instrument_id: InstrumentId,
630 ) -> Result<(), DeribitWsError> {
631 let request_id = self.next_request_id();
632 let order_id = params.order_id.clone();
633
634 self.pending_requests.insert(
635 request_id,
636 PendingRequestType::Edit {
637 client_order_id,
638 trader_id,
639 strategy_id,
640 instrument_id,
641 },
642 );
643
644 let request =
645 DeribitJsonRpcRequest::new(request_id, DeribitWsMethod::Edit.as_method_str(), params);
646
647 let payload =
648 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
649
650 log::debug!("Sending edit order: request_id={request_id}, order_id={order_id}");
651 self.send_tracked_request(
652 request_id,
653 payload,
654 Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()),
655 )
656 .await
657 }
658
659 /// Handles a cancel order command.
660 async fn handle_cancel(
661 &mut self,
662 params: DeribitCancelParams,
663 client_order_id: ClientOrderId,
664 trader_id: TraderId,
665 strategy_id: StrategyId,
666 instrument_id: InstrumentId,
667 ) -> Result<(), DeribitWsError> {
668 let request_id = self.next_request_id();
669 let order_id = params.order_id.clone();
670
671 self.pending_requests.insert(
672 request_id,
673 PendingRequestType::Cancel {
674 client_order_id,
675 trader_id,
676 strategy_id,
677 instrument_id,
678 },
679 );
680
681 let request =
682 DeribitJsonRpcRequest::new(request_id, DeribitWsMethod::Cancel.as_method_str(), params);
683
684 let payload =
685 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
686
687 log::debug!("Sending cancel order: request_id={request_id}, order_id={order_id}");
688 self.send_tracked_request(
689 request_id,
690 payload,
691 Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()),
692 )
693 .await
694 }
695
696 /// Handles cancel all orders by instrument command.
697 async fn handle_cancel_all_by_instrument(
698 &mut self,
699 params: DeribitCancelAllByInstrumentParams,
700 instrument_id: InstrumentId,
701 ) -> Result<(), DeribitWsError> {
702 let request_id = self.next_request_id();
703 let instrument_name = params.instrument_name.clone();
704
705 // Track this request for response correlation
706 self.pending_requests.insert(
707 request_id,
708 PendingRequestType::CancelAllByInstrument { instrument_id },
709 );
710
711 let request = DeribitJsonRpcRequest::new(
712 request_id,
713 DeribitWsMethod::CancelAllByInstrument.as_method_str(),
714 params,
715 );
716
717 let payload =
718 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
719
720 log::debug!(
721 "Sending cancel_all_by_instrument: request_id={request_id}, instrument={instrument_name}"
722 );
723 self.send_tracked_request(
724 request_id,
725 payload,
726 Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()),
727 )
728 .await
729 }
730
731 /// Handles get order state command.
732 async fn handle_get_order_state(
733 &mut self,
734 order_id: String,
735 client_order_id: ClientOrderId,
736 trader_id: TraderId,
737 strategy_id: StrategyId,
738 instrument_id: InstrumentId,
739 ) -> Result<(), DeribitWsError> {
740 let request_id = self.next_request_id();
741
742 // Track this request for response correlation
743 self.pending_requests.insert(
744 request_id,
745 PendingRequestType::GetOrderState {
746 client_order_id,
747 trader_id,
748 strategy_id,
749 instrument_id,
750 },
751 );
752
753 let params = serde_json::json!({
754 "order_id": order_id
755 });
756
757 let request = DeribitJsonRpcRequest::new(
758 request_id,
759 DeribitWsMethod::GetOrderState.as_method_str(),
760 params,
761 );
762
763 let payload =
764 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
765
766 log::debug!("Sending get_order_state: request_id={request_id}, order_id={order_id}");
767 self.send_tracked_request(
768 request_id,
769 payload,
770 Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()),
771 )
772 .await
773 }
774
775 /// Processes a command from the client.
776 async fn process_command(&mut self, cmd: HandlerCommand) {
777 match cmd {
778 HandlerCommand::SetClient(client) => {
779 log::debug!("Setting WebSocket client");
780 self.inner = Some(client);
781 }
782 HandlerCommand::Disconnect => {
783 log::debug!("Disconnecting WebSocket");
784
785 if let Some(client) = self.inner.take() {
786 client.disconnect().await;
787 }
788 }
789 HandlerCommand::Authenticate { auth_params } => {
790 let request_id = self.next_request_id();
791 log::debug!("Authenticating: request_id={request_id}");
792
793 // Track this request for response correlation
794 self.pending_requests
795 .insert(request_id, PendingRequestType::Authenticate);
796
797 let request = DeribitJsonRpcRequest::new(
798 request_id,
799 DeribitWsMethod::PublicAuth.as_method_str(),
800 auth_params,
801 );
802
803 match serde_json::to_string(&request) {
804 Ok(payload) => {
805 if let Err(e) = self.send_with_retry(payload, None).await {
806 self.pending_requests.remove(&request_id);
807 log::error!("Authentication send failed: {e}");
808 self.auth_tracker.fail(format!("Send failed: {e}"));
809 }
810 }
811 Err(e) => {
812 self.pending_requests.remove(&request_id);
813 log::error!("Failed to serialize auth request: {e}");
814 self.auth_tracker.fail(format!("Serialization failed: {e}"));
815 }
816 }
817 }
818 HandlerCommand::SetHeartbeat { interval } => {
819 if let Err(e) = self.handle_set_heartbeat(interval).await {
820 log::error!("Set heartbeat failed: {e}");
821 }
822 }
823 HandlerCommand::InitializeInstruments(instruments) => {
824 log::debug!("Handler received {} instruments", instruments.len());
825 self.instruments_cache.clear();
826 for inst in instruments {
827 self.instruments_cache
828 .insert(inst.raw_symbol().inner(), inst);
829 }
830 }
831 HandlerCommand::UpdateInstrument(instrument) => {
832 log::trace!("Updating instrument: {}", instrument.raw_symbol());
833 self.instruments_cache
834 .insert(instrument.raw_symbol().inner(), *instrument);
835 }
836 HandlerCommand::Subscribe { channels } => {
837 if let Err(e) = self.handle_subscribe(channels).await {
838 log::error!("Subscribe failed: {e}");
839 }
840 }
841 HandlerCommand::Unsubscribe { channels } => {
842 // User-initiated unsubscribe cancels any pending book resync
843 // for these channels so we don't re-subscribe against user intent
844 self.pending_book_resync.retain(|ch| !channels.contains(ch));
845
846 if let Err(e) = self.handle_unsubscribe(channels).await {
847 log::error!("Unsubscribe failed: {e}");
848 }
849 }
850 HandlerCommand::Buy {
851 params,
852 client_order_id,
853 trader_id,
854 strategy_id,
855 instrument_id,
856 } => {
857 if let Err(e) = self
858 .handle_buy(
859 params,
860 client_order_id,
861 trader_id,
862 strategy_id,
863 instrument_id,
864 )
865 .await
866 {
867 log::error!("Buy order failed: {e}");
868 }
869 }
870 HandlerCommand::Sell {
871 params,
872 client_order_id,
873 trader_id,
874 strategy_id,
875 instrument_id,
876 } => {
877 if let Err(e) = self
878 .handle_sell(
879 params,
880 client_order_id,
881 trader_id,
882 strategy_id,
883 instrument_id,
884 )
885 .await
886 {
887 log::error!("Sell order failed: {e}");
888 }
889 }
890 HandlerCommand::Edit {
891 params,
892 client_order_id,
893 trader_id,
894 strategy_id,
895 instrument_id,
896 } => {
897 if let Err(e) = self
898 .handle_edit(
899 params,
900 client_order_id,
901 trader_id,
902 strategy_id,
903 instrument_id,
904 )
905 .await
906 {
907 log::error!("Edit order failed: {e}");
908 }
909 }
910 HandlerCommand::Cancel {
911 params,
912 client_order_id,
913 trader_id,
914 strategy_id,
915 instrument_id,
916 } => {
917 if let Err(e) = self
918 .handle_cancel(
919 params,
920 client_order_id,
921 trader_id,
922 strategy_id,
923 instrument_id,
924 )
925 .await
926 {
927 log::error!("Cancel order failed: {e}");
928 }
929 }
930 HandlerCommand::CancelAllByInstrument {
931 params,
932 instrument_id,
933 } => {
934 if let Err(e) = self
935 .handle_cancel_all_by_instrument(params, instrument_id)
936 .await
937 {
938 log::error!("Cancel all by instrument failed: {e}");
939 }
940 }
941 HandlerCommand::GetOrderState {
942 order_id,
943 client_order_id,
944 trader_id,
945 strategy_id,
946 instrument_id,
947 } => {
948 if let Err(e) = self
949 .handle_get_order_state(
950 order_id,
951 client_order_id,
952 trader_id,
953 strategy_id,
954 instrument_id,
955 )
956 .await
957 {
958 log::error!("Get order state failed: {e}");
959 }
960 }
961 }
962 }
963
964 /// Processes a raw WebSocket message.
965 async fn process_raw_message(&mut self, text: &str) -> Option<NautilusWsMessage> {
966 if text == RECONNECTED {
967 log::info!("Received reconnection signal");
968
969 self.auth_tracker.invalidate();
970 self.clear_state();
971
972 return Some(NautilusWsMessage::Reconnected);
973 }
974
975 // Parse the JSON-RPC message
976 let ws_msg = match parse_raw_message(text) {
977 Ok(msg) => msg,
978 Err(e) => {
979 log::warn!("Failed to parse message: {e}");
980 return None;
981 }
982 };
983
984 let ts_init = self.ts_init();
985
986 match ws_msg {
987 DeribitWsMessage::Response(response) => {
988 // Look up the request type by ID for explicit correlation
989 if let Some(request_id) = response.id
990 && let Some(request_type) = self.pending_requests.remove(&request_id)
991 {
992 match request_type {
993 PendingRequestType::Authenticate => {
994 if let Some(error) = &response.error {
995 let reason = format!(
996 "Authentication error code={}: {}",
997 error.code, error.message
998 );
999 log::error!(
1000 "Authentication failed: code={}, message={}, request_id={}",
1001 error.code,
1002 error.message,
1003 request_id
1004 );
1005 self.auth_tracker.fail(reason.clone());
1006 return Some(NautilusWsMessage::AuthenticationFailed(reason));
1007 } else if let Some(result) = &response.result {
1008 match serde_json::from_value::<DeribitAuthResult>(result.clone()) {
1009 Ok(auth_result) => {
1010 self.auth_tracker.succeed();
1011 log::debug!(
1012 "WebSocket authenticated successfully (request_id={}, scope={}, expires_in={}s)",
1013 request_id,
1014 auth_result.scope,
1015 auth_result.expires_in
1016 );
1017 return Some(NautilusWsMessage::Authenticated(Box::new(
1018 auth_result,
1019 )));
1020 }
1021 Err(e) => {
1022 let reason = format!("Failed to parse auth result: {e}");
1023 log::error!("{reason}: request_id={request_id}");
1024 self.auth_tracker.fail(reason.clone());
1025 return Some(NautilusWsMessage::AuthenticationFailed(
1026 reason,
1027 ));
1028 }
1029 }
1030 }
1031 }
1032 PendingRequestType::Subscribe { channels } => {
1033 if let Some(error) = &response.error {
1034 log::error!(
1035 "Subscribe failed: code={}, message={}, channels={:?}, request_id={}",
1036 error.code,
1037 error.message,
1038 channels,
1039 request_id
1040 );
1041
1042 if let Ok(mut errors) = self.subscribe_errors.lock() {
1043 errors.push(format!(
1044 "Subscribe rejected: code={}, message={}",
1045 error.code, error.message,
1046 ));
1047 }
1048 } else {
1049 // Confirm each channel in the subscription
1050 for ch in &channels {
1051 self.subscriptions_state.confirm_subscribe(ch);
1052 log::debug!("Subscription confirmed: {ch}");
1053 }
1054 }
1055 }
1056 PendingRequestType::Unsubscribe { channels } => {
1057 if let Some(error) = &response.error {
1058 log::error!(
1059 "Unsubscribe failed: code={}, message={}, channels={:?}, request_id={}",
1060 error.code,
1061 error.message,
1062 channels,
1063 request_id
1064 );
1065 } else {
1066 for ch in &channels {
1067 self.subscriptions_state.confirm_unsubscribe(ch);
1068 log::debug!("Unsubscription confirmed: {ch}");
1069 }
1070 }
1071
1072 // Resubscribe channels pending book resync (kept in
1073 // pending_book_resync until a fresh snapshot arrives)
1074 if !self.pending_book_resync.is_empty() {
1075 let resync: Vec<String> = channels
1076 .iter()
1077 .filter(|ch| self.pending_book_resync.contains(ch))
1078 .cloned()
1079 .collect();
1080
1081 if !resync.is_empty() {
1082 let _ = self.handle_subscribe(resync).await;
1083 }
1084 }
1085 }
1086 PendingRequestType::SetHeartbeat => {
1087 if let Some(error) = &response.error {
1088 log::error!(
1089 "Set heartbeat failed: code={}, message={}, request_id={}",
1090 error.code,
1091 error.message,
1092 request_id
1093 );
1094 } else {
1095 log::debug!("Heartbeat enabled (request_id={request_id})");
1096 }
1097 }
1098 PendingRequestType::Test => {
1099 if let Some(error) = &response.error {
1100 log::warn!(
1101 "Heartbeat test failed: code={}, message={}, request_id={}",
1102 error.code,
1103 error.message,
1104 request_id
1105 );
1106 } else {
1107 log::trace!(
1108 "Heartbeat test acknowledged (request_id={request_id})"
1109 );
1110 }
1111 }
1112 PendingRequestType::Cancel {
1113 client_order_id,
1114 trader_id,
1115 strategy_id,
1116 instrument_id,
1117 } => {
1118 if let Some(result) = &response.result {
1119 match serde_json::from_value::<DeribitOrderMsg>(result.clone()) {
1120 Ok(order_msg) => {
1121 let venue_order_id =
1122 VenueOrderId::new(order_msg.order_id.as_str());
1123 log::debug!(
1124 "Cancel confirmed: venue_order_id={venue_order_id}, \
1125 client_order_id={client_order_id}, state={}",
1126 order_msg.order_state
1127 );
1128
1129 // Emit OrderCanceled from the response path so we
1130 // do not lose the event during a reconnection gap.
1131 // Both paths check terminal_orders to suppress
1132 // duplicates regardless of which arrives first.
1133 if order_msg.order_state == "cancelled"
1134 && !self.terminal_orders.contains(&client_order_id)
1135 {
1136 let instrument_name_ustr =
1137 Ustr::from(order_msg.instrument_name.as_str());
1138
1139 if let Some(instrument) =
1140 self.instruments_cache.get(&instrument_name_ustr)
1141 && let Some(account_id) = self.account_id
1142 {
1143 let event = parse_order_canceled(
1144 &order_msg,
1145 instrument,
1146 account_id,
1147 trader_id,
1148 strategy_id,
1149 ts_init,
1150 );
1151 // Keep order_contexts so the subscription
1152 // path resolves the same client_order_id
1153 // and hits the terminal_orders dedup check
1154 self.terminal_orders.add(client_order_id);
1155 return Some(NautilusWsMessage::OrderCanceled(
1156 event,
1157 ));
1158 }
1159 }
1160 }
1161 Err(e) => {
1162 log::error!(
1163 "Failed to parse cancel response: request_id={request_id}, error={e}"
1164 );
1165 }
1166 }
1167 } else if let Some(error) = &response.error {
1168 log::error!(
1169 "Cancel rejected: code={}, message={}, client_order_id={}",
1170 error.code,
1171 error.message,
1172 client_order_id
1173 );
1174 return Some(NautilusWsMessage::OrderCancelRejected(
1175 OrderCancelRejected::new(
1176 trader_id,
1177 strategy_id,
1178 instrument_id,
1179 client_order_id,
1180 ustr::ustr(&format!(
1181 "code={}: {}",
1182 error.code, error.message
1183 )),
1184 UUID4::new(),
1185 ts_init,
1186 ts_init,
1187 false,
1188 None, // venue_order_id not available in error response
1189 self.account_id,
1190 ),
1191 ));
1192 }
1193 }
1194 PendingRequestType::CancelAllByInstrument { instrument_id } => {
1195 if let Some(result) = &response.result {
1196 match serde_json::from_value::<u64>(result.clone()) {
1197 Ok(count) => {
1198 log::debug!(
1199 "Cancelled {count} orders for instrument {instrument_id}"
1200 );
1201 // Individual order status updates come via user.orders subscription
1202 }
1203 Err(e) => {
1204 log::warn!("Failed to parse cancel_all response: {e}");
1205 }
1206 }
1207 } else if let Some(error) = &response.error {
1208 log::error!(
1209 "Cancel all by instrument rejected: code={}, message={}, instrument_id={}",
1210 error.code,
1211 error.message,
1212 instrument_id
1213 );
1214 }
1215 }
1216 PendingRequestType::Buy {
1217 client_order_id,
1218 trader_id,
1219 strategy_id,
1220 instrument_id,
1221 }
1222 | PendingRequestType::Sell {
1223 client_order_id,
1224 trader_id,
1225 strategy_id,
1226 instrument_id,
1227 } => {
1228 if let Some(result) = &response.result {
1229 match serde_json::from_value::<DeribitOrderResponse>(result.clone())
1230 {
1231 Ok(order_response) => {
1232 let venue_order_id_str = &order_response.order.order_id;
1233 let venue_order_id =
1234 VenueOrderId::new(venue_order_id_str.as_str());
1235 let order_state = &order_response.order.order_state;
1236 log::debug!(
1237 "Order response: venue_order_id={venue_order_id}, client_order_id={client_order_id}, state={order_state}"
1238 );
1239
1240 self.order_contexts.insert(
1241 venue_order_id,
1242 OrderContext {
1243 client_order_id,
1244 trader_id,
1245 strategy_id,
1246 instrument_id,
1247 },
1248 );
1249
1250 // Skip OrderAccepted if order already reached terminal state
1251 if self.terminal_orders.contains(&client_order_id) {
1252 log::debug!(
1253 "Skipping OrderAccepted for terminal order: client_order_id={client_order_id}"
1254 );
1255 self.emitted_accepted.add(venue_order_id);
1256 } else if order_state == "filled" {
1257 // Order went directly Submitted -> Filled (e.g., market orders)
1258 log::debug!(
1259 "Skipping OrderAccepted for already filled order: venue_order_id={venue_order_id}, client_order_id={client_order_id}"
1260 );
1261 self.terminal_orders.add(client_order_id);
1262 self.emitted_accepted.add(venue_order_id);
1263 } else {
1264 let instrument_name_ustr = Ustr::from(
1265 order_response.order.instrument_name.as_str(),
1266 );
1267
1268 if let Some(instrument) =
1269 self.instruments_cache.get(&instrument_name_ustr)
1270 {
1271 if let Some(account_id) = self.account_id {
1272 let event = parse_order_accepted(
1273 &order_response.order,
1274 instrument,
1275 account_id,
1276 trader_id,
1277 strategy_id,
1278 ts_init,
1279 );
1280 // Mark OrderAccepted as emitted to prevent duplicate from subscription
1281 self.emitted_accepted.add(venue_order_id);
1282 return Some(NautilusWsMessage::OrderAccepted(
1283 event,
1284 ));
1285 } else {
1286 log::warn!(
1287 "Cannot create OrderAccepted: account_id not set"
1288 );
1289 }
1290 } else {
1291 log::warn!(
1292 "Instrument {instrument_name_ustr} not found in cache for order response"
1293 );
1294 }
1295 }
1296 }
1297 Err(e) => {
1298 log::error!(
1299 "Failed to parse order response: request_id={request_id}, error={e}"
1300 );
1301 }
1302 }
1303 } else if let Some(error) = &response.error {
1304 let due_post_only = error.code == DERIBIT_POST_ONLY_ERROR_CODE;
1305 let reason = if let Some(data) = &error.data {
1306 format!(
1307 "code={}: {} (data: {})",
1308 error.code, error.message, data
1309 )
1310 } else {
1311 format!("code={}: {}", error.code, error.message)
1312 };
1313
1314 log::debug!(
1315 "Order rejected: {reason}, client_order_id={client_order_id}"
1316 );
1317 return Some(NautilusWsMessage::OrderRejected(OrderRejected::new(
1318 trader_id,
1319 strategy_id,
1320 instrument_id,
1321 client_order_id,
1322 self.account_id.unwrap_or(AccountId::new("DERIBIT-UNKNOWN")),
1323 ustr::ustr(&reason),
1324 UUID4::new(),
1325 ts_init,
1326 ts_init,
1327 false,
1328 due_post_only,
1329 )));
1330 }
1331 }
1332 PendingRequestType::Edit {
1333 client_order_id,
1334 trader_id,
1335 strategy_id,
1336 instrument_id,
1337 } => {
1338 if let Some(result) = &response.result {
1339 match serde_json::from_value::<DeribitOrderResponse>(result.clone())
1340 {
1341 Ok(order_response) => {
1342 let venue_order_id =
1343 VenueOrderId::new(&order_response.order.order_id);
1344 log::debug!(
1345 "Order updated: venue_order_id={}, client_order_id={}, state={}",
1346 venue_order_id,
1347 client_order_id,
1348 order_response.order.order_state
1349 );
1350
1351 self.order_contexts.insert(
1352 venue_order_id,
1353 OrderContext {
1354 client_order_id,
1355 trader_id,
1356 strategy_id,
1357 instrument_id,
1358 },
1359 );
1360
1361 let instrument_name_ustr = Ustr::from(
1362 order_response.order.instrument_name.as_str(),
1363 );
1364
1365 if let Some(instrument) =
1366 self.instruments_cache.get(&instrument_name_ustr)
1367 {
1368 if let Some(account_id) = self.account_id {
1369 let event = parse_order_updated(
1370 &order_response.order,
1371 instrument,
1372 account_id,
1373 trader_id,
1374 strategy_id,
1375 ts_init,
1376 );
1377 return Some(NautilusWsMessage::OrderUpdated(
1378 event,
1379 ));
1380 } else {
1381 log::warn!(
1382 "Cannot create OrderUpdated: account_id not set"
1383 );
1384 }
1385 } else {
1386 log::warn!(
1387 "Instrument {instrument_name_ustr} not found in cache for edit response"
1388 );
1389 }
1390 }
1391 Err(e) => {
1392 log::error!(
1393 "Failed to parse edit response: request_id={request_id}, error={e}"
1394 );
1395 }
1396 }
1397 } else if let Some(error) = &response.error {
1398 log::error!(
1399 "Order modify rejected: code={}, message={}, client_order_id={}",
1400 error.code,
1401 error.message,
1402 client_order_id
1403 );
1404 return Some(NautilusWsMessage::OrderModifyRejected(
1405 OrderModifyRejected::new(
1406 trader_id,
1407 strategy_id,
1408 instrument_id,
1409 client_order_id,
1410 ustr::ustr(&format!(
1411 "code={}: {}",
1412 error.code, error.message
1413 )),
1414 UUID4::new(),
1415 ts_init,
1416 ts_init,
1417 false,
1418 None, // venue_order_id not available
1419 self.account_id,
1420 ),
1421 ));
1422 }
1423 }
1424 PendingRequestType::GetOrderState {
1425 client_order_id,
1426 trader_id: _,
1427 strategy_id: _,
1428 instrument_id: _,
1429 } => {
1430 if let Some(result) = &response.result {
1431 match serde_json::from_value::<DeribitOrderMsg>(result.clone()) {
1432 Ok(order_msg) => {
1433 log::debug!(
1434 "Order state received: venue_order_id={}, client_order_id={}, state={}",
1435 order_msg.order_id,
1436 client_order_id,
1437 order_msg.order_state
1438 );
1439
1440 // Convert to OrderStatusReport
1441 let instrument_name_ustr = order_msg.instrument_name;
1442
1443 if let Some(instrument) =
1444 self.instruments_cache.get(&instrument_name_ustr)
1445 {
1446 if let Some(account_id) = self.account_id {
1447 match parse_user_order_msg(
1448 &order_msg, instrument, account_id, ts_init,
1449 ) {
1450 Ok(report) => {
1451 return Some(
1452 NautilusWsMessage::OrderStatusReports(
1453 vec![report],
1454 ),
1455 );
1456 }
1457 Err(e) => {
1458 log::warn!(
1459 "Failed to parse get_order_state response to report: {e}"
1460 );
1461 }
1462 }
1463 } else {
1464 log::warn!(
1465 "Cannot create OrderStatusReport: account_id not set"
1466 );
1467 }
1468 } else {
1469 log::warn!(
1470 "Instrument {instrument_name_ustr} not found in cache for get_order_state response"
1471 );
1472 }
1473 }
1474 Err(e) => {
1475 log::error!(
1476 "Failed to parse get_order_state response: request_id={request_id}, error={e}"
1477 );
1478 }
1479 }
1480 } else if let Some(error) = &response.error {
1481 log::error!(
1482 "Get order state failed: code={}, message={}, client_order_id={}",
1483 error.code,
1484 error.message,
1485 client_order_id
1486 );
1487 }
1488 }
1489 }
1490 } else if let Some(request_id) = response.id {
1491 // Response with ID but no matching pending request
1492 if let Some(error) = &response.error {
1493 // Log orphaned error response with all available context
1494 log::error!(
1495 "Deribit error for unknown request: code={}, message={}, request_id={}, data={:?}",
1496 error.code,
1497 error.message,
1498 request_id,
1499 error.data
1500 );
1501 return Some(NautilusWsMessage::Error(DeribitWsError::DeribitError {
1502 code: error.code,
1503 message: error.message.clone(),
1504 }));
1505 } else {
1506 // Success response but no pending request - likely already processed
1507 log::debug!(
1508 "Received response for unknown request_id={}, result present: {}",
1509 request_id,
1510 response.result.is_some()
1511 );
1512 }
1513 } else if let Some(error) = &response.error {
1514 // Error response with no ID (shouldn't happen in JSON-RPC 2.0, but handle it)
1515 log::error!(
1516 "Deribit error with no request_id: code={}, message={}, data={:?}",
1517 error.code,
1518 error.message,
1519 error.data
1520 );
1521 return Some(NautilusWsMessage::Error(DeribitWsError::DeribitError {
1522 code: error.code,
1523 message: error.message.clone(),
1524 }));
1525 }
1526 None
1527 }
1528 DeribitWsMessage::Notification(notification) => {
1529 let channel = ¬ification.params.channel;
1530 let data = ¬ification.params.data;
1531
1532 // Determine channel type and parse accordingly
1533 if let Some(channel_type) = DeribitWsChannel::from_channel_string(channel) {
1534 match channel_type {
1535 DeribitWsChannel::Trades => {
1536 // Parse trade messages
1537 match serde_json::from_value::<Vec<DeribitTradeMsg>>(data.clone()) {
1538 Ok(trades) => {
1539 log::debug!("Received {} trades", trades.len());
1540 let data_vec = parse_trades_data(
1541 &trades,
1542 &self.instruments_cache,
1543 ts_init,
1544 );
1545
1546 if data_vec.is_empty() && !trades.is_empty() {
1547 let missing: Vec<&Ustr> = trades
1548 .iter()
1549 .map(|t| &t.instrument_name)
1550 .filter(|name| {
1551 !self.instruments_cache.contains_key(name)
1552 })
1553 .collect();
1554
1555 if missing.is_empty() {
1556 log::warn!(
1557 "Received {} trades but parsed 0 (parse failures); cache size: {}",
1558 trades.len(),
1559 self.instruments_cache.len()
1560 );
1561 } else {
1562 log::warn!(
1563 "Trade message received but instrument(s) not found in cache: {:?} (cache size: {})",
1564 missing,
1565 self.instruments_cache.len()
1566 );
1567 }
1568 } else if !data_vec.is_empty() {
1569 log::debug!("Parsed {} trade ticks", data_vec.len());
1570 return Some(NautilusWsMessage::Data(data_vec));
1571 }
1572 }
1573 Err(e) => {
1574 log::warn!("Failed to deserialize trades: {e}");
1575 }
1576 }
1577 }
1578 DeribitWsChannel::Book => {
1579 // Parse order book messages
1580 match serde_json::from_value::<DeribitBookMsg>(data.clone()) {
1581 Ok(book_msg) => {
1582 if let Some(instrument) =
1583 self.instruments_cache.get(&book_msg.instrument_name)
1584 {
1585 let inst_name = book_msg.instrument_name.to_string();
1586 let awaiting_resync =
1587 self.pending_book_resync.iter().any(|ch| {
1588 ch.starts_with("book.")
1589 && ch
1590 .split('.')
1591 .nth(1)
1592 .is_some_and(|s| s == inst_name)
1593 });
1594
1595 if awaiting_resync
1596 && book_msg.msg_type == DeribitBookMsgType::Change
1597 {
1598 // Drop deltas while awaiting resync snapshot
1599 } else if awaiting_resync
1600 && book_msg.msg_type == DeribitBookMsgType::Snapshot
1601 {
1602 self.pending_book_resync.retain(|ch| {
1603 !(ch.starts_with("book.")
1604 && ch
1605 .split('.')
1606 .nth(1)
1607 .is_some_and(|s| s == inst_name))
1608 });
1609 self.book_sequence.insert(
1610 book_msg.instrument_name,
1611 book_msg.change_id,
1612 );
1613
1614 match parse_book_msg(&book_msg, instrument, ts_init) {
1615 Ok(deltas) => {
1616 return Some(NautilusWsMessage::Deltas(deltas));
1617 }
1618 Err(e) => {
1619 log::warn!("Failed to parse book message: {e}");
1620 }
1621 }
1622 } else if book_msg.msg_type == DeribitBookMsgType::Change
1623 && let Some(prev_id) = book_msg.prev_change_id
1624 && let Some(&last_id) =
1625 self.book_sequence.get(&book_msg.instrument_name)
1626 && prev_id != last_id
1627 {
1628 log::error!(
1629 "Book sequence gap for {}: expected prev_change_id={}, was {} \
1630 - dropping delta, forcing resync",
1631 book_msg.instrument_name,
1632 last_id,
1633 prev_id
1634 );
1635 self.book_sequence.remove(&book_msg.instrument_name);
1636
1637 let book_channels: Vec<String> = self
1638 .subscriptions_state
1639 .all_topics()
1640 .into_iter()
1641 .filter(|t| {
1642 t.starts_with("book.")
1643 && t.split('.')
1644 .nth(1)
1645 .is_some_and(|s| s == inst_name)
1646 })
1647 .collect();
1648
1649 if !book_channels.is_empty() {
1650 for ch in &book_channels {
1651 self.subscriptions_state.mark_failure(ch);
1652 }
1653 // Defer resubscribe until unsubscribe ack
1654 self.pending_book_resync
1655 .extend(book_channels.clone());
1656 let _ =
1657 self.handle_unsubscribe(book_channels).await;
1658 }
1659 } else {
1660 self.book_sequence.insert(
1661 book_msg.instrument_name,
1662 book_msg.change_id,
1663 );
1664
1665 match parse_book_msg(&book_msg, instrument, ts_init) {
1666 Ok(deltas) => {
1667 return Some(NautilusWsMessage::Deltas(deltas));
1668 }
1669 Err(e) => {
1670 log::warn!("Failed to parse book message: {e}");
1671 }
1672 }
1673 }
1674 } else {
1675 log::warn!(
1676 "Book message received but instrument '{}' not found in cache (cache size: {})",
1677 book_msg.instrument_name,
1678 self.instruments_cache.len()
1679 );
1680 }
1681 }
1682 Err(e) => {
1683 log::warn!(
1684 "Failed to deserialize book message: {e}, channel: {channel}"
1685 );
1686 }
1687 }
1688 }
1689 DeribitWsChannel::Ticker => {
1690 match serde_json::from_value::<DeribitTickerMsg>(data.clone()) {
1691 Ok(ticker_msg) => {
1692 if let Some(instrument) =
1693 self.instruments_cache.get(&ticker_msg.instrument_name)
1694 {
1695 // Emit OptionGreeks only if subscribed
1696 if self.option_greeks_subs.contains(&instrument.id())
1697 && let Some(option_greeks) =
1698 parse_ticker_to_option_greeks(
1699 &ticker_msg,
1700 instrument,
1701 ts_init,
1702 )
1703 {
1704 let _ = self.out_tx.send(
1705 NautilusWsMessage::OptionGreeks(option_greeks),
1706 );
1707 }
1708
1709 let instrument_id = instrument.id();
1710 let mut data_vec = Vec::new();
1711
1712 // Emit MarkPriceUpdate only if subscribed
1713 if self.mark_price_subs.contains(&instrument_id) {
1714 match parse_ticker_to_mark_price(
1715 &ticker_msg,
1716 instrument,
1717 ts_init,
1718 ) {
1719 Ok(mark_price) => {
1720 data_vec
1721 .push(Data::MarkPriceUpdate(mark_price));
1722 }
1723 Err(e) => {
1724 log::warn!("Failed to parse mark price: {e}");
1725 }
1726 }
1727 }
1728
1729 // Emit IndexPriceUpdate only if subscribed
1730 if self.index_price_subs.contains(&instrument_id) {
1731 match parse_ticker_to_index_price(
1732 &ticker_msg,
1733 instrument,
1734 ts_init,
1735 ) {
1736 Ok(index_price) => {
1737 data_vec
1738 .push(Data::IndexPriceUpdate(index_price));
1739 }
1740 Err(e) => {
1741 log::warn!("Failed to parse index price: {e}");
1742 }
1743 }
1744 }
1745
1746 if !data_vec.is_empty() {
1747 return Some(NautilusWsMessage::Data(data_vec));
1748 }
1749 } else {
1750 log::warn!(
1751 "Ticker message received but instrument '{}' not found in cache (cache size: {})",
1752 ticker_msg.instrument_name,
1753 self.instruments_cache.len()
1754 );
1755 }
1756 }
1757 Err(e) => {
1758 log::warn!(
1759 "Failed to deserialize ticker message: {e}, channel: {channel}"
1760 );
1761 }
1762 }
1763 }
1764 DeribitWsChannel::Perpetual => {
1765 // Parse perpetual channel for funding rate updates
1766 // This channel is dedicated to perpetual instruments and provides
1767 // the interest (funding) rate
1768 match serde_json::from_value::<DeribitPerpetualMsg>(data.clone()) {
1769 Ok(perpetual_msg) => {
1770 // Extract instrument name from channel: perpetual.{instrument}.{interval}
1771 let parts: Vec<&str> = channel.split('.').collect();
1772 if parts.len() >= 2 {
1773 let instrument_name = Ustr::from(parts[1]);
1774
1775 if let Some(instrument) =
1776 self.instruments_cache.get(&instrument_name)
1777 {
1778 let funding_rate = parse_perpetual_to_funding_rate(
1779 &perpetual_msg,
1780 instrument,
1781 ts_init,
1782 );
1783 return Some(NautilusWsMessage::FundingRates(vec![
1784 funding_rate,
1785 ]));
1786 } else {
1787 log::warn!(
1788 "Instrument {} not found in cache (cache size: {})",
1789 instrument_name,
1790 self.instruments_cache.len()
1791 );
1792 }
1793 }
1794 }
1795 Err(e) => {
1796 log::warn!(
1797 "Failed to deserialize perpetual message: {e}, data: {data}"
1798 );
1799 }
1800 }
1801 }
1802 DeribitWsChannel::Quote => {
1803 // Parse quote messages
1804 match serde_json::from_value::<DeribitQuoteMsg>(data.clone()) {
1805 Ok(quote_msg) => {
1806 if let Some(instrument) =
1807 self.instruments_cache.get("e_msg.instrument_name)
1808 {
1809 match parse_quote_msg("e_msg, instrument, ts_init) {
1810 Ok(quote) => {
1811 return Some(NautilusWsMessage::Data(vec![
1812 Data::Quote(quote),
1813 ]));
1814 }
1815 Err(e) => {
1816 log::warn!("Failed to parse quote message: {e}");
1817 }
1818 }
1819 } else {
1820 log::warn!(
1821 "Quote message received but instrument '{}' not found in cache (cache size: {})",
1822 quote_msg.instrument_name,
1823 self.instruments_cache.len()
1824 );
1825 }
1826 }
1827 Err(e) => {
1828 log::warn!(
1829 "Failed to deserialize quote message: {e}, channel: {channel}"
1830 );
1831 }
1832 }
1833 }
1834 DeribitWsChannel::VolatilityIndex => {
1835 match serde_json::from_value::<DeribitVolatilityIndexMsg>(data.clone())
1836 {
1837 Ok(msg) => {
1838 let ts_event = UnixNanos::from(msg.timestamp * 1_000_000);
1839 let mut metadata = nautilus_core::Params::new();
1840 metadata.insert(
1841 "index_name".to_string(),
1842 serde_json::Value::String(msg.index_name.clone()),
1843 );
1844 let data_type = DataType::new(
1845 "DeribitVolatilityIndex",
1846 Some(metadata),
1847 None,
1848 );
1849
1850 let dvol = DeribitVolatilityIndex::new(
1851 msg.index_name,
1852 msg.volatility,
1853 ts_event,
1854 ts_init,
1855 );
1856
1857 return Some(NautilusWsMessage::Data(vec![Data::Custom(
1858 CustomData::new(Arc::new(dvol), data_type),
1859 )]));
1860 }
1861 Err(e) => {
1862 log::warn!("Failed to deserialize volatility index: {e}");
1863 }
1864 }
1865 }
1866 DeribitWsChannel::InstrumentState => {
1867 match serde_json::from_value::<DeribitInstrumentStateMsg>(data.clone())
1868 {
1869 Ok(state_msg) => {
1870 log::debug!(
1871 "Instrument state change: {} -> {} (timestamp: {})",
1872 state_msg.instrument_name,
1873 state_msg.state,
1874 state_msg.timestamp
1875 );
1876
1877 let instrument_id = if let Some(instrument) =
1878 self.instruments_cache.get(&state_msg.instrument_name)
1879 {
1880 instrument.id()
1881 } else {
1882 log::debug!(
1883 "Instrument '{}' not in cache, constructing ID",
1884 state_msg.instrument_name
1885 );
1886 InstrumentId::new(
1887 Symbol::new(state_msg.instrument_name),
1888 *DERIBIT_VENUE,
1889 )
1890 };
1891
1892 let action = MarketStatusAction::from(state_msg.state);
1893 let is_trading =
1894 Some(state_msg.state == DeribitInstrumentState::Started);
1895 let ts_event = UnixNanos::from(state_msg.timestamp * 1_000_000);
1896 let status = InstrumentStatus::new(
1897 instrument_id,
1898 action,
1899 ts_event,
1900 ts_init,
1901 None,
1902 None,
1903 is_trading,
1904 None,
1905 None,
1906 );
1907 return Some(NautilusWsMessage::InstrumentStatus(status));
1908 }
1909 Err(e) => {
1910 log::warn!("Failed to parse instrument status message: {e}");
1911 }
1912 }
1913 }
1914 DeribitWsChannel::ChartTrades => {
1915 // Parse chart.trades messages into Bar objects using emit-on-next pattern.
1916 // Deribit sends updates for the current bar as it builds. We only emit
1917 // a bar when we receive a bar with a different timestamp, confirming
1918 // the previous bar is closed.
1919 if let Ok(chart_msg) =
1920 serde_json::from_value::<DeribitChartMsg>(data.clone())
1921 {
1922 // Extract instrument and resolution from channel
1923 // Channel format: chart.trades.{instrument}.{resolution}
1924 let parts: Vec<&str> = channel.split('.').collect();
1925 if parts.len() >= 4 {
1926 let instrument_name = Ustr::from(parts[2]);
1927 let resolution = parts[3];
1928
1929 if let Some(instrument) =
1930 self.instruments_cache.get(&instrument_name)
1931 {
1932 let instrument_id = instrument.id();
1933
1934 match resolution_to_bar_type(instrument_id, resolution) {
1935 Ok(bar_type) => {
1936 let price_precision = instrument.price_precision();
1937 let size_precision = instrument.size_precision();
1938 let use_cost_for_volume =
1939 use_cost_for_bar_volume(instrument);
1940
1941 match parse_chart_msg(
1942 &chart_msg,
1943 bar_type,
1944 price_precision,
1945 size_precision,
1946 use_cost_for_volume,
1947 self.bars_timestamp_on_close,
1948 ts_init,
1949 ) {
1950 Ok(new_bar) => {
1951 // Check if we have a pending bar for this channel
1952 let channel_key = channel.clone();
1953
1954 if let Some(pending_bar) =
1955 self.pending_bars.get(&channel_key)
1956 {
1957 // If new bar has different timestamp, the pending bar is closed
1958 if new_bar.ts_event
1959 != pending_bar.ts_event
1960 {
1961 let closed_bar = *pending_bar;
1962 self.pending_bars
1963 .insert(channel_key, new_bar);
1964 log::debug!(
1965 "Emitting closed bar: {closed_bar:?}"
1966 );
1967 return Some(
1968 NautilusWsMessage::Data(vec![
1969 Data::Bar(closed_bar),
1970 ]),
1971 );
1972 }
1973 // Same timestamp - update pending bar with latest values
1974 self.pending_bars
1975 .insert(channel_key, new_bar);
1976 } else {
1977 // First bar for this channel - store as pending
1978 self.pending_bars
1979 .insert(channel_key, new_bar);
1980 }
1981 }
1982 Err(e) => {
1983 log::warn!(
1984 "Failed to parse chart message to bar: {e}"
1985 );
1986 }
1987 }
1988 }
1989 Err(e) => {
1990 log::warn!(
1991 "Failed to create BarType from resolution {resolution}: {e}"
1992 );
1993 }
1994 }
1995 } else {
1996 log::warn!(
1997 "Instrument {instrument_name} not found in cache for chart data"
1998 );
1999 }
2000 }
2001 }
2002 }
2003 DeribitWsChannel::UserOrders => {
2004 // Handle both array and single object responses
2005 let orders_result =
2006 serde_json::from_value::<Vec<DeribitOrderMsg>>(data.clone())
2007 .or_else(|_| {
2008 serde_json::from_value::<DeribitOrderMsg>(data.clone())
2009 .map(|order| vec![order])
2010 });
2011
2012 match orders_result {
2013 Ok(orders) => {
2014 log::debug!("Received {} user order updates", orders.len());
2015
2016 // Require account_id for parsing
2017 let Some(account_id) = self.account_id else {
2018 log::warn!("Cannot parse user orders: account_id not set");
2019 return Some(NautilusWsMessage::Raw(data.clone()));
2020 };
2021
2022 let mut outgoing = Vec::new();
2023
2024 // Process each order and emit appropriate events
2025 for order in &orders {
2026 let venue_order_id_str = &order.order_id;
2027 let venue_order_id =
2028 VenueOrderId::new(venue_order_id_str.as_str());
2029 let instrument_name = order.instrument_name;
2030
2031 let Some(instrument) =
2032 self.instruments_cache.get(&instrument_name)
2033 else {
2034 log::warn!(
2035 "Instrument {instrument_name} not found in cache"
2036 );
2037 continue;
2038 };
2039
2040 // Look up OrderContext for this order
2041 // First check order_contexts (for orders whose response has been processed)
2042 // Then check pending_requests (for orders whose response hasn't arrived yet)
2043 // If neither found, this is a true external order
2044 let context =
2045 self.order_contexts.get(&venue_order_id).cloned();
2046
2047 // Extract client_order_id from order label for pending check
2048 let label_client_order_id = order
2049 .label
2050 .as_ref()
2051 .filter(|l| !l.is_empty())
2052 .map(ClientOrderId::new);
2053
2054 // Check for pending request if not in order_contexts
2055 let pending_context = if context.is_none() {
2056 if let Some(client_id) = &label_client_order_id {
2057 self.get_pending_order_context(client_id)
2058 } else {
2059 None
2060 }
2061 } else {
2062 None
2063 };
2064
2065 // Check if order has a pending request for context resolution
2066 let has_pending_request =
2067 if let Some(client_id) = &label_client_order_id {
2068 self.is_pending_order(client_id)
2069 } else {
2070 false
2071 };
2072
2073 let effective_context = context.or(pending_context);
2074 let is_known_order =
2075 effective_context.is_some() || has_pending_request;
2076
2077 // Determine event type based on order state
2078 let event_type = determine_order_event_type(
2079 &order.order_state,
2080 !is_known_order, // is_new if we don't know about it
2081 false, // not from edit response
2082 );
2083
2084 let (trader_id, strategy_id, client_order_id) =
2085 if let Some(ctx) = effective_context {
2086 (
2087 ctx.trader_id,
2088 ctx.strategy_id,
2089 ctx.client_order_id,
2090 )
2091 } else {
2092 // External order - use default values
2093 // Note: These won't match any strategy, which is correct
2094 (
2095 TraderId::new("EXTERNAL-000"),
2096 StrategyId::new("EXTERNAL"),
2097 ClientOrderId::new(venue_order_id_str),
2098 )
2099 };
2100
2101 match event_type {
2102 OrderEventType::Accepted => {
2103 // Skip if order already reached terminal state (race condition)
2104 if self.terminal_orders.contains(&client_order_id) {
2105 log::debug!(
2106 "Skipping OrderAccepted for terminal order: client_order_id={client_order_id}"
2107 );
2108 continue;
2109 }
2110
2111 // Check if we already emitted OrderAccepted for this order
2112 // This prevents duplicates from both response and subscription paths
2113 if self.emitted_accepted.contains(&venue_order_id) {
2114 log::trace!(
2115 "Skipping duplicate OrderAccepted: venue_order_id={venue_order_id}"
2116 );
2117 continue;
2118 }
2119
2120 let event = parse_order_accepted(
2121 order,
2122 instrument,
2123 account_id,
2124 trader_id,
2125 strategy_id,
2126 ts_init,
2127 );
2128
2129 // Mark OrderAccepted as emitted
2130 self.emitted_accepted.add(venue_order_id);
2131
2132 log::debug!(
2133 "Emitting OrderAccepted: venue_order_id={venue_order_id}, is_known={is_known_order}"
2134 );
2135 outgoing
2136 .push(NautilusWsMessage::OrderAccepted(event));
2137 }
2138 OrderEventType::Canceled => {
2139 // Skip if already emitted from the cancel
2140 // response path
2141 if self.terminal_orders.contains(&client_order_id) {
2142 log::trace!(
2143 "Skipping duplicate OrderCanceled: client_order_id={client_order_id}"
2144 );
2145 continue;
2146 }
2147
2148 let event = parse_order_canceled(
2149 order,
2150 instrument,
2151 account_id,
2152 trader_id,
2153 strategy_id,
2154 ts_init,
2155 );
2156 log::debug!(
2157 "Emitting OrderCanceled: venue_order_id={venue_order_id}"
2158 );
2159 self.terminal_orders.add(client_order_id);
2160 self.order_contexts.remove(&venue_order_id);
2161 self.emitted_accepted.remove(&venue_order_id);
2162 outgoing
2163 .push(NautilusWsMessage::OrderCanceled(event));
2164 }
2165 OrderEventType::Expired => {
2166 let event = parse_order_expired(
2167 order,
2168 instrument,
2169 account_id,
2170 trader_id,
2171 strategy_id,
2172 ts_init,
2173 );
2174 log::debug!(
2175 "Emitting OrderExpired: venue_order_id={venue_order_id}"
2176 );
2177 self.terminal_orders.add(client_order_id);
2178 self.order_contexts.remove(&venue_order_id);
2179 self.emitted_accepted.remove(&venue_order_id);
2180 outgoing
2181 .push(NautilusWsMessage::OrderExpired(event));
2182 }
2183 OrderEventType::Updated => {
2184 // Emit OrderStatusReport for updates
2185 // This includes quantity/price changes from modify
2186 match parse_user_order_msg(
2187 order, instrument, account_id, ts_init,
2188 ) {
2189 Ok(report) => {
2190 log::debug!(
2191 "Emitting OrderStatusReport (updated): venue_order_id={venue_order_id}"
2192 );
2193 outgoing.push(
2194 NautilusWsMessage::OrderStatusReports(
2195 vec![report],
2196 ),
2197 );
2198 }
2199 Err(e) => {
2200 log::warn!(
2201 "Failed to parse order update: {e}"
2202 );
2203 }
2204 }
2205 }
2206 OrderEventType::None => {
2207 // Fills handled via user.trades, track terminal state
2208 // for race condition prevention
2209 if matches!(
2210 order.order_state.as_str(),
2211 "filled" | "rejected"
2212 ) {
2213 log::debug!(
2214 "Recording terminal order: venue_order_id={venue_order_id}, state={}",
2215 order.order_state
2216 );
2217 self.terminal_orders.add(client_order_id);
2218 self.order_contexts.remove(&venue_order_id);
2219 self.emitted_accepted.remove(&venue_order_id);
2220 } else {
2221 log::trace!(
2222 "No event to emit for order {}, state={}",
2223 venue_order_id,
2224 order.order_state
2225 );
2226 }
2227 }
2228 }
2229 }
2230
2231 if !outgoing.is_empty() {
2232 self.pending_outgoing.extend(outgoing);
2233 }
2234 }
2235 Err(e) => {
2236 log::warn!("Failed to deserialize user orders: {e}");
2237 }
2238 }
2239 }
2240 DeribitWsChannel::UserTrades => {
2241 // Handle both array and single object responses
2242 let trades_result =
2243 serde_json::from_value::<Vec<DeribitUserTradeMsg>>(data.clone())
2244 .or_else(|_| {
2245 serde_json::from_value::<DeribitUserTradeMsg>(data.clone())
2246 .map(|trade| vec![trade])
2247 });
2248
2249 match trades_result {
2250 Ok(trades) => {
2251 log::debug!("Received {} user trade updates", trades.len());
2252
2253 let Some(account_id) = self.account_id else {
2254 log::warn!("Cannot parse user trades: account_id not set");
2255 return Some(NautilusWsMessage::Raw(data.clone()));
2256 };
2257
2258 let mut reports = Vec::with_capacity(trades.len());
2259 for trade in &trades {
2260 let instrument_name = trade.instrument_name;
2261
2262 if let Some(instrument) =
2263 self.instruments_cache.get(&instrument_name)
2264 {
2265 match parse_user_trade_msg(
2266 trade, instrument, account_id, ts_init,
2267 ) {
2268 Ok(report) => {
2269 log::debug!(
2270 "Parsed fill report: {} @ {}",
2271 report.trade_id,
2272 report.last_px
2273 );
2274 reports.push(report);
2275 }
2276 Err(e) => {
2277 log::warn!(
2278 "Failed to parse trade {}: {e}",
2279 trade.trade_id
2280 );
2281 }
2282 }
2283 } else {
2284 log::warn!(
2285 "Instrument {instrument_name} not found in cache"
2286 );
2287 }
2288 }
2289
2290 if !reports.is_empty() {
2291 return Some(NautilusWsMessage::FillReports(reports));
2292 }
2293 }
2294 Err(e) => {
2295 log::warn!("Failed to deserialize user trades: {e}");
2296 }
2297 }
2298 }
2299 DeribitWsChannel::UserPortfolio => {
2300 match serde_json::from_value::<DeribitPortfolioMsg>(data.clone()) {
2301 Ok(portfolio) => {
2302 // Skip zero-balance currencies (common with cross-collateral)
2303 // Only check equity and balance - initial_margin can be non-zero
2304 // for all currencies when cross-collateral is enabled
2305 if portfolio.equity.is_zero() && portfolio.balance.is_zero() {
2306 log::trace!(
2307 "Skipping zero-balance portfolio for {}",
2308 portfolio.currency
2309 );
2310 return None;
2311 }
2312
2313 // Require account_id for parsing
2314 let Some(account_id) = self.account_id else {
2315 log::warn!("Cannot parse portfolio: account_id not set");
2316 return None;
2317 };
2318
2319 match parse_portfolio_to_account_state(
2320 &portfolio, account_id, ts_init,
2321 ) {
2322 Ok(account_state) => {
2323 // Check for duplicate per currency
2324 let currency_key = portfolio.currency.clone();
2325
2326 if let Some(last) =
2327 self.last_account_states.get(¤cy_key)
2328 && account_state.has_same_balances_and_margins(last)
2329 {
2330 log::trace!(
2331 "Skipping duplicate portfolio update for {}",
2332 portfolio.currency
2333 );
2334 return None;
2335 }
2336
2337 self.last_account_states
2338 .insert(currency_key, account_state.clone());
2339 return Some(NautilusWsMessage::AccountState(
2340 account_state,
2341 ));
2342 }
2343 Err(e) => {
2344 log::warn!(
2345 "Failed to parse portfolio to AccountState: {e}"
2346 );
2347 }
2348 }
2349 }
2350 Err(e) => {
2351 log::warn!("Failed to deserialize portfolio: {e}");
2352 }
2353 }
2354 }
2355 _ => {
2356 // Unhandled channel - return raw
2357 log::trace!("Unhandled channel: {channel}");
2358 return Some(NautilusWsMessage::Raw(data.clone()));
2359 }
2360 }
2361 } else {
2362 log::trace!("Unknown channel: {channel}");
2363 return Some(NautilusWsMessage::Raw(data.clone()));
2364 }
2365 None
2366 }
2367 DeribitWsMessage::Heartbeat(heartbeat) => {
2368 match heartbeat.heartbeat_type {
2369 DeribitHeartbeatType::TestRequest => {
2370 log::trace!(
2371 "Received heartbeat test_request - responding with public/test"
2372 );
2373
2374 if let Err(e) = self.handle_heartbeat_test_request().await {
2375 log::error!("Failed to respond to heartbeat test_request: {e}");
2376
2377 // Return error to signal connection may be unhealthy
2378 return Some(NautilusWsMessage::Error(DeribitWsError::Send(format!(
2379 "Heartbeat response failed: {e}"
2380 ))));
2381 }
2382 }
2383 DeribitHeartbeatType::Heartbeat => {
2384 log::trace!("Received heartbeat acknowledgment");
2385 }
2386 }
2387 None
2388 }
2389 DeribitWsMessage::Error(err) => {
2390 log::error!("Deribit error {}: {}", err.code, err.message);
2391 Some(NautilusWsMessage::Error(DeribitWsError::DeribitError {
2392 code: err.code,
2393 message: err.message,
2394 }))
2395 }
2396 DeribitWsMessage::Reconnected => Some(NautilusWsMessage::Reconnected),
2397 }
2398 }
2399
2400 /// Main message processing loop.
2401 ///
2402 /// Returns `None` when the handler should stop.
2403 /// Messages that need client-side handling (e.g., Reconnected) are returned.
2404 /// Data messages are sent directly to `out_tx` for the user stream.
2405 pub async fn next(&mut self) -> Option<NautilusWsMessage> {
2406 loop {
2407 if let Some(msg) = self.pending_outgoing.pop_front() {
2408 match msg {
2409 NautilusWsMessage::Reconnected
2410 | NautilusWsMessage::Authenticated(_)
2411 | NautilusWsMessage::AuthenticationFailed(_) => {
2412 return Some(msg);
2413 }
2414 _ => {
2415 let _ = self.out_tx.send(msg);
2416 continue;
2417 }
2418 }
2419 }
2420
2421 tokio::select! {
2422 // Process commands from client
2423 Some(cmd) = self.cmd_rx.recv() => {
2424 self.process_command(cmd).await;
2425 }
2426 // Process raw WebSocket messages
2427 Some(msg) = self.raw_rx.recv() => {
2428 match msg {
2429 Message::Text(text) => {
2430 if let Some(nautilus_msg) = self.process_raw_message(&text).await {
2431 // Send data messages to user stream
2432 match &nautilus_msg {
2433 NautilusWsMessage::Data(_)
2434 | NautilusWsMessage::Deltas(_)
2435 | NautilusWsMessage::Instrument(_)
2436 | NautilusWsMessage::InstrumentStatus(_)
2437 | NautilusWsMessage::OptionGreeks(_)
2438 | NautilusWsMessage::Raw(_)
2439 | NautilusWsMessage::Error(_) => {
2440 let _ = self.out_tx.send(nautilus_msg);
2441 }
2442 NautilusWsMessage::FundingRates(rates) => {
2443 let msg_to_send =
2444 NautilusWsMessage::FundingRates(rates.clone());
2445
2446 if let Err(e) = self.out_tx.send(msg_to_send) {
2447 log::error!("Failed to send funding rates: {e}");
2448 }
2449 }
2450 NautilusWsMessage::OrderStatusReports(_)
2451 | NautilusWsMessage::FillReports(_)
2452 | NautilusWsMessage::OrderAccepted(_)
2453 | NautilusWsMessage::OrderCanceled(_)
2454 | NautilusWsMessage::OrderExpired(_)
2455 | NautilusWsMessage::OrderUpdated(_)
2456 | NautilusWsMessage::OrderRejected(_)
2457 | NautilusWsMessage::OrderCancelRejected(_)
2458 | NautilusWsMessage::OrderModifyRejected(_)
2459 | NautilusWsMessage::AccountState(_) => {
2460 let _ = self.out_tx.send(nautilus_msg);
2461 }
2462 // Return messages that need client-side handling
2463 NautilusWsMessage::Reconnected
2464 | NautilusWsMessage::Authenticated(_)
2465 | NautilusWsMessage::AuthenticationFailed(_) => {
2466 return Some(nautilus_msg);
2467 }
2468 }
2469 }
2470 }
2471 Message::Ping(data) => {
2472 // Respond to ping with pong
2473 if let Some(client) = &self.inner {
2474 let _ = client.send_pong(data.to_vec()).await;
2475 }
2476 }
2477 Message::Close(_) => {
2478 log::debug!("Received close frame");
2479 }
2480 _ => {}
2481 }
2482 }
2483 // Check for stop signal
2484 () = tokio::time::sleep(tokio::time::Duration::from_millis(100)) => {
2485 if self.signal.load(Ordering::Relaxed) {
2486 log::debug!("Stop signal received");
2487 return None;
2488 }
2489 }
2490 }
2491 }
2492 }
2493}