1use std::{
23 collections::VecDeque,
24 sync::{
25 Arc,
26 atomic::{AtomicBool, AtomicU64, Ordering},
27 },
28};
29
30use ahash::AHashMap;
31use nautilus_common::cache::fifo::{FifoCache, FifoCacheMap};
32use nautilus_core::{
33 AtomicSet, AtomicTime, UUID4, UnixNanos,
34 string::secret::{SecretString, zeroize_json_value},
35 time::get_atomic_clock_realtime,
36};
37use nautilus_model::{
38 data::{Bar, CustomData, Data, DataType, InstrumentStatus},
39 enums::{MarketStatusAction, OrderSide, OrderType},
40 events::{
41 AccountState, OrderAccepted, OrderCancelRejected, OrderFilled, OrderModifyRejected,
42 OrderRejected,
43 },
44 identifiers::{
45 AccountId, ClientOrderId, InstrumentId, StrategyId, Symbol, TradeId, TraderId, VenueOrderId,
46 },
47 instruments::{Instrument, InstrumentAny},
48};
49use nautilus_network::{
50 RECONNECTED,
51 retry::{RetryManager, create_websocket_retry_manager},
52 websocket::{AuthTracker, SubscriptionState, WebSocketClient},
53};
54use parking_lot::Mutex;
55use rust_decimal::Decimal;
56use tokio_tungstenite::tungstenite::Message;
57use ustr::Ustr;
58
59use super::{
60 enums::{DeribitBookMsgType, DeribitHeartbeatType, DeribitWsChannel, DeribitWsMethod},
61 error::DeribitWsError,
62 messages::{
63 DeribitAuthResult, DeribitBookMsg, DeribitCancelAllByInstrumentParams, DeribitCancelParams,
64 DeribitChartMsg, DeribitEditParams, DeribitHeartbeatParams, DeribitInstrumentStateMsg,
65 DeribitJsonRpcRequest, DeribitOrderMsg, DeribitOrderParams, DeribitOrderResponse,
66 DeribitPerpetualMsg, DeribitPortfolioMsg, DeribitQuoteMsg, DeribitSubscribeParams,
67 DeribitTickerMsg, DeribitTradeMsg, DeribitUserTradeMsg, DeribitVolatilityIndexMsg,
68 DeribitWsMessage, NautilusWsMessage, parse_raw_message,
69 },
70 parse::{
71 OrderEventType, determine_order_event_type, parse_book_msg, parse_chart_msg,
72 parse_deribit_order_type, parse_order_accepted_with_client_order_id,
73 parse_order_canceled_with_client_order_id, parse_order_expired_with_client_order_id,
74 parse_order_updated_with_client_order_id, parse_perpetual_to_funding_rate, parse_quote_msg,
75 parse_ticker_to_index_price, parse_ticker_to_mark_price, parse_ticker_to_option_greeks,
76 parse_trades_data, parse_user_order_msg, parse_user_trade_msg, resolution_to_bar_type,
77 },
78};
79use crate::{
80 common::{
81 consts::{DERIBIT_POST_ONLY_ERROR_CODE, DERIBIT_RATE_LIMIT_KEY_ORDER, DERIBIT_VENUE},
82 enums::DeribitInstrumentState,
83 parse::{parse_portfolio_to_account_state, use_cost_for_bar_volume},
84 },
85 data_types::DeribitVolatilityIndex,
86};
87
88#[derive(Debug, Clone)]
90pub enum PendingRequestType {
91 Authenticate,
93 Subscribe { channels: Vec<String> },
95 Unsubscribe { channels: Vec<String> },
97 SetHeartbeat,
99 Test,
101 Buy {
103 client_order_id: ClientOrderId,
104 trader_id: TraderId,
105 strategy_id: StrategyId,
106 instrument_id: InstrumentId,
107 order_side: OrderSide,
108 order_type: OrderType,
109 },
110 Sell {
112 client_order_id: ClientOrderId,
113 trader_id: TraderId,
114 strategy_id: StrategyId,
115 instrument_id: InstrumentId,
116 order_side: OrderSide,
117 order_type: OrderType,
118 },
119 Edit {
121 client_order_id: ClientOrderId,
122 trader_id: TraderId,
123 strategy_id: StrategyId,
124 instrument_id: InstrumentId,
125 },
126 Cancel {
128 client_order_id: ClientOrderId,
129 trader_id: TraderId,
130 strategy_id: StrategyId,
131 instrument_id: InstrumentId,
132 },
133 CancelAllByInstrument { instrument_id: InstrumentId },
135 GetOrderState {
137 client_order_id: ClientOrderId,
138 trader_id: TraderId,
139 strategy_id: StrategyId,
140 instrument_id: InstrumentId,
141 },
142}
143
144#[allow(missing_debug_implementations)]
146pub enum HandlerCommand {
147 SetClient(WebSocketClient),
149 Disconnect,
151 Authenticate {
153 auth_params: SecretString,
155 },
156 SetHeartbeat { interval: u64 },
158 InitializeInstruments(Vec<InstrumentAny>),
160 UpdateInstrument(Box<InstrumentAny>),
162 Subscribe { channels: Vec<String> },
164 Unsubscribe { channels: Vec<String> },
166 Buy {
168 params: DeribitOrderParams,
169 client_order_id: ClientOrderId,
170 trader_id: TraderId,
171 strategy_id: StrategyId,
172 instrument_id: InstrumentId,
173 },
174 Sell {
176 params: DeribitOrderParams,
177 client_order_id: ClientOrderId,
178 trader_id: TraderId,
179 strategy_id: StrategyId,
180 instrument_id: InstrumentId,
181 },
182 Edit {
184 params: DeribitEditParams,
185 client_order_id: ClientOrderId,
186 trader_id: TraderId,
187 strategy_id: StrategyId,
188 instrument_id: InstrumentId,
189 },
190 Cancel {
192 params: DeribitCancelParams,
193 client_order_id: ClientOrderId,
194 trader_id: TraderId,
195 strategy_id: StrategyId,
196 instrument_id: InstrumentId,
197 },
198 CancelAllByInstrument {
200 params: DeribitCancelAllByInstrumentParams,
201 instrument_id: InstrumentId,
202 },
203 GetOrderState {
205 order_id: String,
206 client_order_id: ClientOrderId,
207 trader_id: TraderId,
208 strategy_id: StrategyId,
209 instrument_id: InstrumentId,
210 },
211}
212
213#[derive(Debug, Clone)]
217pub struct OrderContext {
218 pub client_order_id: ClientOrderId,
219 pub trader_id: TraderId,
220 pub strategy_id: StrategyId,
221 pub instrument_id: InstrumentId,
222 pub order_side: OrderSide,
223 pub order_type: OrderType,
224 pub accepted: bool,
225 pub last_order_signature: Option<OrderSignature>,
226}
227
228pub type OrderSignature = (Decimal, Option<Decimal>, Option<Decimal>);
230
231#[allow(missing_debug_implementations)]
235pub struct DeribitWsFeedHandler {
236 clock: &'static AtomicTime,
237 signal: Arc<AtomicBool>,
238 inner: Option<WebSocketClient>,
239 cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
240 raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
241 out_tx: tokio::sync::mpsc::UnboundedSender<NautilusWsMessage>,
242 auth_tracker: AuthTracker,
243 subscriptions_state: SubscriptionState,
244 retry_manager: RetryManager<DeribitWsError>,
245 instruments_cache: AHashMap<Ustr, InstrumentAny>,
246 option_greeks_subs: Arc<AtomicSet<InstrumentId>>,
247 mark_price_subs: Arc<AtomicSet<InstrumentId>>,
248 index_price_subs: Arc<AtomicSet<InstrumentId>>,
249 request_id_counter: AtomicU64,
250 pending_requests: AHashMap<u64, PendingRequestType>,
251 account_id: Option<AccountId>,
252 order_contexts: AHashMap<VenueOrderId, OrderContext>,
253 submitted_order_contexts: FifoCacheMap<ClientOrderId, OrderContext, 10_000>,
254
255 terminal_order_contexts: FifoCacheMap<VenueOrderId, OrderContext, 10_000>,
257
258 emitted_trades: FifoCache<(VenueOrderId, TradeId), 10_000>,
259 pending_bars: AHashMap<String, Bar>,
260 bars_timestamp_on_close: bool,
261 last_account_states: AHashMap<String, AccountState>,
262 book_sequence: AHashMap<Ustr, u64>,
263 pending_book_resync: Vec<String>,
264 pending_outgoing: VecDeque<NautilusWsMessage>,
265 subscribe_errors: Arc<Mutex<Vec<String>>>,
266}
267
268impl DeribitWsFeedHandler {
269 #[expect(clippy::too_many_arguments)]
271 #[must_use]
272 pub fn new(
273 signal: Arc<AtomicBool>,
274 cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
275 raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
276 out_tx: tokio::sync::mpsc::UnboundedSender<NautilusWsMessage>,
277 auth_tracker: AuthTracker,
278 subscriptions_state: SubscriptionState,
279 option_greeks_subs: Arc<AtomicSet<InstrumentId>>,
280 mark_price_subs: Arc<AtomicSet<InstrumentId>>,
281 index_price_subs: Arc<AtomicSet<InstrumentId>>,
282 account_id: Option<AccountId>,
283 bars_timestamp_on_close: bool,
284 subscribe_errors: Arc<Mutex<Vec<String>>>,
285 ) -> Self {
286 Self {
287 clock: get_atomic_clock_realtime(),
288 signal,
289 inner: None,
290 cmd_rx,
291 raw_rx,
292 out_tx,
293 auth_tracker,
294 subscriptions_state,
295 retry_manager: create_websocket_retry_manager(),
296 instruments_cache: AHashMap::new(),
297 option_greeks_subs,
298 mark_price_subs,
299 index_price_subs,
300 request_id_counter: AtomicU64::new(1),
301 pending_requests: AHashMap::new(),
302 account_id,
303 order_contexts: AHashMap::new(),
304 submitted_order_contexts: FifoCacheMap::new(),
305 terminal_order_contexts: FifoCacheMap::new(),
306 emitted_trades: FifoCache::new(),
307 pending_bars: AHashMap::new(),
308 bars_timestamp_on_close,
309 last_account_states: AHashMap::new(),
310 book_sequence: AHashMap::new(),
311 pending_book_resync: Vec::new(),
312 pending_outgoing: VecDeque::new(),
313 subscribe_errors,
314 }
315 }
316
317 pub fn set_account_id(&mut self, account_id: AccountId) {
319 self.account_id = Some(account_id);
320 }
321
322 #[must_use]
324 pub fn account_id(&self) -> Option<AccountId> {
325 self.account_id
326 }
327
328 fn clear_state(&mut self) {
329 let pending_count = self.pending_requests.len();
330 let bars_count = self.pending_bars.len();
331 let account_count = self.last_account_states.len();
332 let book_count = self.book_sequence.len();
333 let outgoing_count = self.pending_outgoing.len();
334
335 self.pending_requests.clear();
336 self.pending_bars.clear();
337 self.last_account_states.clear();
338 self.book_sequence.clear();
339 self.pending_book_resync.clear();
340 self.pending_outgoing.clear();
341
342 log::debug!(
343 "Reset state: pending_requests={pending_count}, pending_bars={bars_count}, \
344 account_states={account_count}, book_sequence={book_count}, \
345 pending_outgoing={outgoing_count}"
346 );
347 }
348
349 fn next_request_id(&self) -> u64 {
351 self.request_id_counter.fetch_add(1, Ordering::Relaxed)
352 }
353
354 fn ts_init(&self) -> UnixNanos {
356 self.clock.get_time_ns()
357 }
358
359 async fn send_tracked_request(
360 &mut self,
361 request_id: u64,
362 payload: Result<String, DeribitWsError>,
363 rate_limit_keys: Option<&[Ustr]>,
364 ) -> Result<(), DeribitWsError> {
365 let payload = match payload {
366 Ok(p) => p,
367 Err(e) => {
368 self.pending_requests.remove(&request_id);
369 return Err(e);
370 }
371 };
372 self.send_with_retry(payload, rate_limit_keys).await
373 }
374
375 async fn send_with_retry(
377 &self,
378 payload: String,
379 rate_limit_keys: Option<&[Ustr]>,
380 ) -> Result<(), DeribitWsError> {
381 self.send_secret_with_retry(payload.into(), rate_limit_keys)
382 .await
383 }
384
385 async fn send_secret_with_retry(
386 &self,
387 payload: SecretString,
388 rate_limit_keys: Option<&[Ustr]>,
389 ) -> Result<(), DeribitWsError> {
390 if let Some(client) = &self.inner {
391 let keys_owned: Option<Vec<Ustr>> = rate_limit_keys.map(<[Ustr]>::to_vec);
392 self.retry_manager
393 .invocation(
394 "websocket_send",
395 || {
396 let payload = payload.clone();
397 let keys = keys_owned.clone();
398 async move {
399 client
400 .send_text(payload.expose_secret().to_owned(), keys.as_deref())
401 .await
402 .map_err(|e| DeribitWsError::Send(e.to_string()))
403 }
404 },
405 |e| matches!(e, DeribitWsError::Send(_)),
406 |e| DeribitWsError::Timeout(e.to_string()),
407 )
408 .execute()
409 .await
410 } else {
411 Err(DeribitWsError::NotConnected)
412 }
413 }
414
415 async fn handle_authenticate(&mut self, auth_params: SecretString) {
416 let request_id = self.next_request_id();
417 log::debug!("Authenticating: request_id={request_id}");
418
419 let auth_params: serde_json::Value = match serde_json::from_str(auth_params.expose_secret())
420 {
421 Ok(auth_params) => auth_params,
422 Err(e) => {
423 log::error!("Failed to deserialize auth params: {e}");
424 self.auth_tracker.fail(format!("Serialization failed: {e}"));
425 return;
426 }
427 };
428 self.pending_requests
429 .insert(request_id, PendingRequestType::Authenticate);
430
431 let mut request = DeribitJsonRpcRequest::new(
432 request_id,
433 DeribitWsMethod::PublicAuth.as_method_str(),
434 auth_params,
435 );
436 let serialized = serde_json::to_string(&request).map(SecretString::from);
437 zeroize_json_value(&mut request.params);
438 drop(request);
439
440 match serialized {
441 Ok(payload) => {
442 if let Err(e) = self.send_secret_with_retry(payload, None).await {
443 self.pending_requests.remove(&request_id);
444 log::error!("Authentication send failed: {e}");
445 self.auth_tracker.fail(format!("Send failed: {e}"));
446 }
447 }
448 Err(e) => {
449 self.pending_requests.remove(&request_id);
450 log::error!("Failed to serialize auth request: {e}");
451 self.auth_tracker.fail(format!("Serialization failed: {e}"));
452 }
453 }
454 }
455
456 async fn handle_subscribe(&mut self, channels: Vec<String>) -> Result<(), DeribitWsError> {
460 let request_id = self.next_request_id();
461
462 self.pending_requests.insert(
464 request_id,
465 PendingRequestType::Subscribe {
466 channels: channels.clone(),
467 },
468 );
469
470 let method = if channels
472 .iter()
473 .any(|ch| DeribitWsChannel::requires_auth(ch))
474 {
475 DeribitWsMethod::PrivateSubscribe
476 } else {
477 DeribitWsMethod::PublicSubscribe
478 };
479
480 let request = DeribitJsonRpcRequest::new(
481 request_id,
482 method.as_method_str(),
483 DeribitSubscribeParams {
484 channels: channels.clone(),
485 },
486 );
487
488 let payload =
489 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
490
491 log::debug!("Subscribing to channels: request_id={request_id}, channels={channels:?}");
492 self.send_tracked_request(request_id, payload, None).await
493 }
494
495 async fn handle_unsubscribe(&mut self, channels: Vec<String>) -> Result<(), DeribitWsError> {
497 let request_id = self.next_request_id();
498
499 self.pending_requests.insert(
501 request_id,
502 PendingRequestType::Unsubscribe {
503 channels: channels.clone(),
504 },
505 );
506
507 let method = if channels
508 .iter()
509 .any(|ch| DeribitWsChannel::requires_auth(ch))
510 {
511 DeribitWsMethod::PrivateUnsubscribe
512 } else {
513 DeribitWsMethod::PublicUnsubscribe
514 };
515
516 let request = DeribitJsonRpcRequest::new(
517 request_id,
518 method.as_method_str(),
519 DeribitSubscribeParams {
520 channels: channels.clone(),
521 },
522 );
523
524 let payload =
525 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
526
527 log::debug!("Unsubscribing from channels: request_id={request_id}, channels={channels:?}");
528 self.send_tracked_request(request_id, payload, None).await
529 }
530
531 async fn handle_set_heartbeat(&mut self, interval: u64) -> Result<(), DeribitWsError> {
533 let request_id = self.next_request_id();
534
535 self.pending_requests
537 .insert(request_id, PendingRequestType::SetHeartbeat);
538
539 let request = DeribitJsonRpcRequest::new(
540 request_id,
541 DeribitWsMethod::SetHeartbeat.as_method_str(),
542 DeribitHeartbeatParams { interval },
543 );
544
545 let payload =
546 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
547
548 log::debug!(
549 "Enabling heartbeat with interval: request_id={request_id}, interval={interval} seconds"
550 );
551 self.send_tracked_request(request_id, payload, None).await
552 }
553
554 async fn handle_heartbeat_test_request(&mut self) -> Result<(), DeribitWsError> {
556 let request_id = self.next_request_id();
557
558 self.pending_requests
560 .insert(request_id, PendingRequestType::Test);
561
562 let request = DeribitJsonRpcRequest::new(
563 request_id,
564 DeribitWsMethod::Test.as_method_str(),
565 serde_json::json!({}),
566 );
567
568 let payload =
569 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
570
571 log::trace!("Responding to heartbeat test_request: request_id={request_id}");
572 self.send_tracked_request(request_id, payload, None).await
573 }
574
575 async fn handle_buy(
577 &mut self,
578 params: DeribitOrderParams,
579 client_order_id: ClientOrderId,
580 trader_id: TraderId,
581 strategy_id: StrategyId,
582 instrument_id: InstrumentId,
583 ) -> Result<(), DeribitWsError> {
584 let request_id = self.next_request_id();
585 let order_type = parse_deribit_order_type(¶ms.order_type);
586 let order_signature = (params.amount, params.price, params.trigger_price);
587
588 let request =
589 DeribitJsonRpcRequest::new(request_id, DeribitWsMethod::Buy.as_method_str(), params);
590
591 let payload =
593 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()))?;
594
595 self.submitted_order_contexts.insert(
596 client_order_id,
597 OrderContext {
598 client_order_id,
599 trader_id,
600 strategy_id,
601 instrument_id,
602 order_side: OrderSide::Buy,
603 order_type,
604 accepted: false,
605 last_order_signature: Some(order_signature),
606 },
607 );
608
609 self.pending_requests.insert(
610 request_id,
611 PendingRequestType::Buy {
612 client_order_id,
613 trader_id,
614 strategy_id,
615 instrument_id,
616 order_side: OrderSide::Buy,
617 order_type,
618 },
619 );
620
621 log::debug!("Sending buy order: request_id={request_id}");
622 self.send_with_retry(payload, Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()))
623 .await
624 }
625
626 async fn handle_sell(
628 &mut self,
629 params: DeribitOrderParams,
630 client_order_id: ClientOrderId,
631 trader_id: TraderId,
632 strategy_id: StrategyId,
633 instrument_id: InstrumentId,
634 ) -> Result<(), DeribitWsError> {
635 let request_id = self.next_request_id();
636 let order_type = parse_deribit_order_type(¶ms.order_type);
637 let order_signature = (params.amount, params.price, params.trigger_price);
638
639 let request =
640 DeribitJsonRpcRequest::new(request_id, DeribitWsMethod::Sell.as_method_str(), params);
641
642 let payload =
644 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()))?;
645
646 self.submitted_order_contexts.insert(
647 client_order_id,
648 OrderContext {
649 client_order_id,
650 trader_id,
651 strategy_id,
652 instrument_id,
653 order_side: OrderSide::Sell,
654 order_type,
655 accepted: false,
656 last_order_signature: Some(order_signature),
657 },
658 );
659
660 self.pending_requests.insert(
661 request_id,
662 PendingRequestType::Sell {
663 client_order_id,
664 trader_id,
665 strategy_id,
666 instrument_id,
667 order_side: OrderSide::Sell,
668 order_type,
669 },
670 );
671
672 log::debug!("Sending sell order: request_id={request_id}");
673 self.send_with_retry(payload, Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()))
674 .await
675 }
676
677 async fn handle_edit(
679 &mut self,
680 params: DeribitEditParams,
681 client_order_id: ClientOrderId,
682 trader_id: TraderId,
683 strategy_id: StrategyId,
684 instrument_id: InstrumentId,
685 ) -> Result<(), DeribitWsError> {
686 let request_id = self.next_request_id();
687 let order_id = params.order_id.clone();
688
689 self.pending_requests.insert(
690 request_id,
691 PendingRequestType::Edit {
692 client_order_id,
693 trader_id,
694 strategy_id,
695 instrument_id,
696 },
697 );
698
699 let request =
700 DeribitJsonRpcRequest::new(request_id, DeribitWsMethod::Edit.as_method_str(), params);
701
702 let payload =
703 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
704
705 log::debug!("Sending edit order: request_id={request_id}, order_id={order_id}");
706 self.send_tracked_request(
707 request_id,
708 payload,
709 Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()),
710 )
711 .await
712 }
713
714 async fn handle_cancel(
716 &mut self,
717 params: DeribitCancelParams,
718 client_order_id: ClientOrderId,
719 trader_id: TraderId,
720 strategy_id: StrategyId,
721 instrument_id: InstrumentId,
722 ) -> Result<(), DeribitWsError> {
723 let request_id = self.next_request_id();
724 let order_id = params.order_id.clone();
725
726 self.pending_requests.insert(
727 request_id,
728 PendingRequestType::Cancel {
729 client_order_id,
730 trader_id,
731 strategy_id,
732 instrument_id,
733 },
734 );
735
736 let request =
737 DeribitJsonRpcRequest::new(request_id, DeribitWsMethod::Cancel.as_method_str(), params);
738
739 let payload =
740 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
741
742 log::debug!("Sending cancel order: request_id={request_id}, order_id={order_id}");
743 self.send_tracked_request(
744 request_id,
745 payload,
746 Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()),
747 )
748 .await
749 }
750
751 async fn handle_cancel_all_by_instrument(
753 &mut self,
754 params: DeribitCancelAllByInstrumentParams,
755 instrument_id: InstrumentId,
756 ) -> Result<(), DeribitWsError> {
757 let request_id = self.next_request_id();
758 let instrument_name = params.instrument_name.clone();
759
760 self.pending_requests.insert(
762 request_id,
763 PendingRequestType::CancelAllByInstrument { instrument_id },
764 );
765
766 let request = DeribitJsonRpcRequest::new(
767 request_id,
768 DeribitWsMethod::CancelAllByInstrument.as_method_str(),
769 params,
770 );
771
772 let payload =
773 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
774
775 log::debug!(
776 "Sending cancel_all_by_instrument: request_id={request_id}, instrument={instrument_name}"
777 );
778 self.send_tracked_request(
779 request_id,
780 payload,
781 Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()),
782 )
783 .await
784 }
785
786 async fn handle_get_order_state(
788 &mut self,
789 order_id: String,
790 client_order_id: ClientOrderId,
791 trader_id: TraderId,
792 strategy_id: StrategyId,
793 instrument_id: InstrumentId,
794 ) -> Result<(), DeribitWsError> {
795 let request_id = self.next_request_id();
796
797 self.pending_requests.insert(
799 request_id,
800 PendingRequestType::GetOrderState {
801 client_order_id,
802 trader_id,
803 strategy_id,
804 instrument_id,
805 },
806 );
807
808 let params = serde_json::json!({
809 "order_id": order_id
810 });
811
812 let request = DeribitJsonRpcRequest::new(
813 request_id,
814 DeribitWsMethod::GetOrderState.as_method_str(),
815 params,
816 );
817
818 let payload =
819 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
820
821 log::debug!("Sending get_order_state: request_id={request_id}, order_id={order_id}");
822 self.send_tracked_request(
823 request_id,
824 payload,
825 Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()),
826 )
827 .await
828 }
829
830 async fn process_command(&mut self, cmd: HandlerCommand) {
832 match cmd {
833 HandlerCommand::SetClient(client) => {
834 log::debug!("Setting WebSocket client");
835 self.inner = Some(client);
836 }
837 HandlerCommand::Disconnect => {
838 log::debug!("Disconnecting WebSocket");
839
840 if let Some(client) = self.inner.take() {
841 client.disconnect().await;
842 }
843 }
844 HandlerCommand::Authenticate { auth_params } => {
845 self.handle_authenticate(auth_params).await;
846 }
847 HandlerCommand::SetHeartbeat { interval } => {
848 if let Err(e) = self.handle_set_heartbeat(interval).await {
849 log::error!("Set heartbeat failed: {e}");
850 }
851 }
852 HandlerCommand::InitializeInstruments(instruments) => {
853 log::debug!("Handler received {} instruments", instruments.len());
854 self.instruments_cache.clear();
855 for inst in instruments {
856 self.instruments_cache
857 .insert(inst.raw_symbol().inner(), inst);
858 }
859 }
860 HandlerCommand::UpdateInstrument(instrument) => {
861 log::trace!("Updating instrument: {}", instrument.raw_symbol());
862 self.instruments_cache
863 .insert(instrument.raw_symbol().inner(), *instrument);
864 }
865 HandlerCommand::Subscribe { channels } => {
866 if let Err(e) = self.handle_subscribe(channels).await {
867 log::error!("Subscribe failed: {e}");
868 }
869 }
870 HandlerCommand::Unsubscribe { channels } => {
871 self.pending_book_resync.retain(|ch| !channels.contains(ch));
874
875 if let Err(e) = self.handle_unsubscribe(channels).await {
876 log::error!("Unsubscribe failed: {e}");
877 }
878 }
879 HandlerCommand::Buy {
880 params,
881 client_order_id,
882 trader_id,
883 strategy_id,
884 instrument_id,
885 } => {
886 if let Err(e) = self
887 .handle_buy(
888 params,
889 client_order_id,
890 trader_id,
891 strategy_id,
892 instrument_id,
893 )
894 .await
895 {
896 log::error!("Buy order failed: {e}");
897 }
898 }
899 HandlerCommand::Sell {
900 params,
901 client_order_id,
902 trader_id,
903 strategy_id,
904 instrument_id,
905 } => {
906 if let Err(e) = self
907 .handle_sell(
908 params,
909 client_order_id,
910 trader_id,
911 strategy_id,
912 instrument_id,
913 )
914 .await
915 {
916 log::error!("Sell order failed: {e}");
917 }
918 }
919 HandlerCommand::Edit {
920 params,
921 client_order_id,
922 trader_id,
923 strategy_id,
924 instrument_id,
925 } => {
926 if let Err(e) = self
927 .handle_edit(
928 params,
929 client_order_id,
930 trader_id,
931 strategy_id,
932 instrument_id,
933 )
934 .await
935 {
936 log::error!("Edit order failed: {e}");
937 }
938 }
939 HandlerCommand::Cancel {
940 params,
941 client_order_id,
942 trader_id,
943 strategy_id,
944 instrument_id,
945 } => {
946 if let Err(e) = self
947 .handle_cancel(
948 params,
949 client_order_id,
950 trader_id,
951 strategy_id,
952 instrument_id,
953 )
954 .await
955 {
956 log::error!("Cancel order failed: {e}");
957 }
958 }
959 HandlerCommand::CancelAllByInstrument {
960 params,
961 instrument_id,
962 } => {
963 if let Err(e) = self
964 .handle_cancel_all_by_instrument(params, instrument_id)
965 .await
966 {
967 log::error!("Cancel all by instrument failed: {e}");
968 }
969 }
970 HandlerCommand::GetOrderState {
971 order_id,
972 client_order_id,
973 trader_id,
974 strategy_id,
975 instrument_id,
976 } => {
977 if let Err(e) = self
978 .handle_get_order_state(
979 order_id,
980 client_order_id,
981 trader_id,
982 strategy_id,
983 instrument_id,
984 )
985 .await
986 {
987 log::error!("Get order state failed: {e}");
988 }
989 }
990 }
991 }
992
993 async fn process_raw_message(&mut self, text: &str) -> Option<NautilusWsMessage> {
995 if text == RECONNECTED {
996 log::info!("Received reconnection signal");
997
998 self.auth_tracker.invalidate();
999 self.clear_state();
1000
1001 return Some(NautilusWsMessage::Reconnected);
1002 }
1003
1004 let ws_msg = match parse_raw_message(text) {
1006 Ok(msg) => msg,
1007 Err(e) => {
1008 log::warn!("Failed to parse message: {e}");
1009 return None;
1010 }
1011 };
1012
1013 let ts_init = self.ts_init();
1014
1015 match ws_msg {
1016 DeribitWsMessage::Response(response) => {
1017 if let Some(request_id) = response.id
1019 && let Some(request_type) = self.pending_requests.remove(&request_id)
1020 {
1021 match request_type {
1022 PendingRequestType::Authenticate => {
1023 if let Some(error) = &response.error {
1024 let reason = format!(
1025 "Authentication error code={}: {}",
1026 error.code, error.message
1027 );
1028 log::error!(
1029 "Authentication failed: code={}, message={}, request_id={}",
1030 error.code,
1031 error.message,
1032 request_id
1033 );
1034 self.auth_tracker.fail(reason.clone());
1035 return Some(NautilusWsMessage::AuthenticationFailed(reason));
1036 } else if let Some(result) = &response.result {
1037 match serde_json::from_str::<DeribitAuthResult>(result.get()) {
1038 Ok(auth_result) => {
1039 self.auth_tracker.succeed();
1040 log::debug!(
1041 "WebSocket authenticated successfully (request_id={}, scope={}, expires_in={}s)",
1042 request_id,
1043 auth_result.scope,
1044 auth_result.expires_in
1045 );
1046 return Some(NautilusWsMessage::Authenticated(Box::new(
1047 auth_result,
1048 )));
1049 }
1050 Err(e) => {
1051 let reason = format!("Failed to parse auth result: {e}");
1052 log::error!("{reason}: request_id={request_id}");
1053 self.auth_tracker.fail(reason.clone());
1054 return Some(NautilusWsMessage::AuthenticationFailed(
1055 reason,
1056 ));
1057 }
1058 }
1059 }
1060 }
1061 PendingRequestType::Subscribe { channels } => {
1062 if let Some(error) = &response.error {
1063 log::error!(
1064 "Subscribe failed: code={}, message={}, channels={:?}, request_id={}",
1065 error.code,
1066 error.message,
1067 channels,
1068 request_id
1069 );
1070
1071 self.subscribe_errors.lock().push(format!(
1072 "Subscribe rejected: code={}, message={}",
1073 error.code, error.message,
1074 ));
1075 } else {
1076 for ch in &channels {
1078 self.subscriptions_state.confirm_subscribe(ch);
1079 log::debug!("Subscription confirmed: {ch}");
1080 }
1081 }
1082 }
1083 PendingRequestType::Unsubscribe { channels } => {
1084 if let Some(error) = &response.error {
1085 log::error!(
1086 "Unsubscribe failed: code={}, message={}, channels={:?}, request_id={}",
1087 error.code,
1088 error.message,
1089 channels,
1090 request_id
1091 );
1092 } else {
1093 for ch in &channels {
1094 self.subscriptions_state.confirm_unsubscribe(ch);
1095 log::debug!("Unsubscription confirmed: {ch}");
1096 }
1097 }
1098
1099 if !self.pending_book_resync.is_empty() {
1102 let resync: Vec<String> = channels
1103 .iter()
1104 .filter(|ch| self.pending_book_resync.contains(ch))
1105 .cloned()
1106 .collect();
1107
1108 if !resync.is_empty() {
1109 let _ = self.handle_subscribe(resync).await;
1110 }
1111 }
1112 }
1113 PendingRequestType::SetHeartbeat => {
1114 if let Some(error) = &response.error {
1115 log::error!(
1116 "Set heartbeat failed: code={}, message={}, request_id={}",
1117 error.code,
1118 error.message,
1119 request_id
1120 );
1121 } else {
1122 log::debug!("Heartbeat enabled (request_id={request_id})");
1123 }
1124 }
1125 PendingRequestType::Test => {
1126 if let Some(error) = &response.error {
1127 log::warn!(
1128 "Heartbeat test failed: code={}, message={}, request_id={}",
1129 error.code,
1130 error.message,
1131 request_id
1132 );
1133 } else {
1134 log::trace!(
1135 "Heartbeat test acknowledged (request_id={request_id})"
1136 );
1137 }
1138 }
1139 PendingRequestType::Cancel {
1140 client_order_id,
1141 trader_id,
1142 strategy_id,
1143 instrument_id,
1144 } => {
1145 if let Some(result) = &response.result {
1146 match serde_json::from_str::<DeribitOrderMsg>(result.get()) {
1147 Ok(order_msg) => {
1148 let venue_order_id =
1149 VenueOrderId::new(order_msg.order_id.as_str());
1150 log::debug!(
1151 "Cancel confirmed: venue_order_id={venue_order_id}, \
1152 client_order_id={client_order_id}, state={}",
1153 order_msg.order_state
1154 );
1155
1156 if order_msg.order_state == "cancelled"
1161 && !self
1162 .terminal_order_contexts
1163 .contains_key(&venue_order_id)
1164 {
1165 let instrument_name_ustr = order_msg.instrument_name;
1166
1167 if let Some(instrument) =
1168 self.instruments_cache.get(&instrument_name_ustr)
1169 && let Some(account_id) = self.account_id
1170 {
1171 let event =
1172 parse_order_canceled_with_client_order_id(
1173 &order_msg,
1174 instrument,
1175 account_id,
1176 trader_id,
1177 strategy_id,
1178 client_order_id,
1179 ts_init,
1180 );
1181 let context = self
1182 .find_order_context(
1183 venue_order_id,
1184 Some(client_order_id),
1185 )
1186 .or_else(|| {
1187 let order_side =
1188 match order_msg.direction.as_str() {
1189 "buy" => OrderSide::Buy,
1190 "sell" => OrderSide::Sell,
1191 _ => return None,
1192 };
1193 Some(OrderContext {
1194 client_order_id,
1195 trader_id,
1196 strategy_id,
1197 instrument_id,
1198 order_side,
1199 order_type: parse_deribit_order_type(
1200 &order_msg.order_type,
1201 ),
1202 accepted: true,
1203 last_order_signature: Some(
1204 Self::order_signature(&order_msg),
1205 ),
1206 })
1207 });
1208
1209 if let Some(context) = context {
1210 self.finish_order_context(
1211 venue_order_id,
1212 &context,
1213 );
1214 }
1215 return Some(NautilusWsMessage::OrderCanceled(
1216 event,
1217 ));
1218 }
1219 }
1220 }
1221 Err(e) => {
1222 log::error!(
1223 "Failed to parse cancel response: request_id={request_id}, error={e}"
1224 );
1225 }
1226 }
1227 } else if let Some(error) = &response.error {
1228 log::warn!(
1229 "Cancel rejected: code={}, message={}, client_order_id={}",
1230 error.code,
1231 error.message,
1232 client_order_id
1233 );
1234 return Some(NautilusWsMessage::OrderCancelRejected(
1235 OrderCancelRejected::new(
1236 trader_id,
1237 strategy_id,
1238 instrument_id,
1239 client_order_id,
1240 ustr::ustr(&format!(
1241 "code={}: {}",
1242 error.code, error.message
1243 )),
1244 UUID4::new(),
1245 ts_init,
1246 ts_init,
1247 false,
1248 None, self.account_id,
1250 ),
1251 ));
1252 }
1253 }
1254 PendingRequestType::CancelAllByInstrument { instrument_id } => {
1255 if let Some(result) = &response.result {
1256 match serde_json::from_str::<u64>(result.get()) {
1257 Ok(count) => {
1258 log::debug!(
1259 "Cancelled {count} orders for instrument {instrument_id}"
1260 );
1261 }
1263 Err(e) => {
1264 log::warn!("Failed to parse cancel_all response: {e}");
1265 }
1266 }
1267 } else if let Some(error) = &response.error {
1268 log::error!(
1269 "Cancel all by instrument rejected: code={}, message={}, instrument_id={}",
1270 error.code,
1271 error.message,
1272 instrument_id
1273 );
1274 }
1275 }
1276 PendingRequestType::Buy {
1277 client_order_id,
1278 trader_id,
1279 strategy_id,
1280 instrument_id,
1281 order_side,
1282 order_type,
1283 }
1284 | PendingRequestType::Sell {
1285 client_order_id,
1286 trader_id,
1287 strategy_id,
1288 instrument_id,
1289 order_side,
1290 order_type,
1291 } => {
1292 if let Some(result) = &response.result {
1293 match serde_json::from_str::<DeribitOrderResponse>(result.get()) {
1294 Ok(order_response) => {
1295 let venue_order_id_str = &order_response.order.order_id;
1296 let venue_order_id =
1297 VenueOrderId::new(venue_order_id_str.as_str());
1298 let order_state = &order_response.order.order_state;
1299 log::debug!(
1300 "Order response: venue_order_id={venue_order_id}, client_order_id={client_order_id}, state={order_state}"
1301 );
1302
1303 let mut context = self
1304 .find_order_context(
1305 venue_order_id,
1306 Some(client_order_id),
1307 )
1308 .unwrap_or(OrderContext {
1309 client_order_id,
1310 trader_id,
1311 strategy_id,
1312 instrument_id,
1313 order_side,
1314 order_type,
1315 accepted: false,
1316 last_order_signature: Some(Self::order_signature(
1317 &order_response.order,
1318 )),
1319 });
1320 context.last_order_signature =
1321 Some(Self::order_signature(&order_response.order));
1322 self.bind_order_context(venue_order_id, context.clone());
1323
1324 if !order_response.trades.is_empty() {
1325 let outgoing = self
1326 .route_user_trades(&order_response.trades, ts_init);
1327 self.pending_outgoing.extend(outgoing);
1328 } else if order_state == "filled" {
1329 log::debug!(
1330 "Deferring acceptance for fast-filled order until its trade arrives: venue_order_id={venue_order_id}, client_order_id={client_order_id}"
1331 );
1332 self.finish_order_context(venue_order_id, &context);
1333 } else if context.accepted {
1334 log::trace!(
1335 "Skipping duplicate OrderAccepted response: venue_order_id={venue_order_id}"
1336 );
1337 } else {
1338 let instrument_name_ustr = Ustr::from(
1339 order_response.order.instrument_name.as_str(),
1340 );
1341
1342 if let Some(instrument) =
1343 self.instruments_cache.get(&instrument_name_ustr)
1344 {
1345 if let Some(account_id) = self.account_id {
1346 let event =
1347 parse_order_accepted_with_client_order_id(
1348 &order_response.order,
1349 instrument,
1350 account_id,
1351 trader_id,
1352 strategy_id,
1353 client_order_id,
1354 ts_init,
1355 );
1356 context.accepted = true;
1357 self.bind_order_context(
1358 venue_order_id,
1359 context,
1360 );
1361 return Some(NautilusWsMessage::OrderAccepted(
1362 event,
1363 ));
1364 } else {
1365 log::warn!(
1366 "Cannot create OrderAccepted: account_id not set"
1367 );
1368 }
1369 } else {
1370 log::warn!(
1371 "Instrument {instrument_name_ustr} not found in cache for order response"
1372 );
1373 }
1374 }
1375 }
1376 Err(e) => {
1377 log::error!(
1378 "Failed to parse order response: request_id={request_id}, error={e}"
1379 );
1380 }
1381 }
1382 } else if let Some(error) = &response.error {
1383 let due_post_only = error.code == DERIBIT_POST_ONLY_ERROR_CODE;
1384 let reason = if let Some(data) = &error.data {
1385 format!(
1386 "code={}: {} (data: {})",
1387 error.code, error.message, data
1388 )
1389 } else {
1390 format!("code={}: {}", error.code, error.message)
1391 };
1392
1393 log::debug!(
1394 "Order rejected: {reason}, client_order_id={client_order_id}"
1395 );
1396 self.submitted_order_contexts.remove(&client_order_id);
1397 return Some(NautilusWsMessage::OrderRejected(OrderRejected::new(
1398 trader_id,
1399 strategy_id,
1400 instrument_id,
1401 client_order_id,
1402 self.account_id.unwrap_or(AccountId::new("DERIBIT-UNKNOWN")),
1403 ustr::ustr(&reason),
1404 UUID4::new(),
1405 ts_init,
1406 ts_init,
1407 false,
1408 due_post_only,
1409 )));
1410 }
1411 }
1412 PendingRequestType::Edit {
1413 client_order_id,
1414 trader_id,
1415 strategy_id,
1416 instrument_id,
1417 } => {
1418 if let Some(result) = &response.result {
1419 match serde_json::from_str::<DeribitOrderResponse>(result.get()) {
1420 Ok(order_response) => {
1421 let venue_order_id =
1422 VenueOrderId::new(&order_response.order.order_id);
1423 log::debug!(
1424 "Order updated: venue_order_id={}, client_order_id={}, state={}",
1425 venue_order_id,
1426 client_order_id,
1427 order_response.order.order_state
1428 );
1429
1430 let instrument_name_ustr = Ustr::from(
1431 order_response.order.instrument_name.as_str(),
1432 );
1433
1434 if let Some(instrument) =
1435 self.instruments_cache.get(&instrument_name_ustr)
1436 {
1437 if let Some(account_id) = self.account_id {
1438 let Some(mut context) = self.find_order_context(
1439 venue_order_id,
1440 Some(client_order_id),
1441 ) else {
1442 let report = parse_user_order_msg(
1443 &order_response.order,
1444 instrument,
1445 account_id,
1446 ts_init,
1447 );
1448 let outgoing = self.route_user_trades(
1449 &order_response.trades,
1450 ts_init,
1451 );
1452 self.pending_outgoing.extend(outgoing);
1453 return report
1454 .map(|report| {
1455 NautilusWsMessage::OrderStatusReports(vec![
1456 report,
1457 ])
1458 })
1459 .map_err(|e| {
1460 log::warn!(
1461 "Failed to parse external edit response: {e}"
1462 );
1463 })
1464 .ok();
1465 };
1466 let was_terminal = self
1467 .terminal_order_contexts
1468 .contains_key(&venue_order_id);
1469 let signature =
1470 Self::order_signature(&order_response.order);
1471 let duplicate_update = was_terminal
1472 || context.last_order_signature
1473 == Some(signature);
1474 let event = (!duplicate_update).then(|| {
1475 parse_order_updated_with_client_order_id(
1476 &order_response.order,
1477 instrument,
1478 account_id,
1479 context.trader_id,
1480 context.strategy_id,
1481 context.client_order_id,
1482 ts_init,
1483 )
1484 });
1485 context.accepted = true;
1486 context.last_order_signature = Some(signature);
1487
1488 if was_terminal {
1489 self.finish_order_context(
1490 venue_order_id,
1491 &context,
1492 );
1493 } else {
1494 self.bind_order_context(
1495 venue_order_id,
1496 context,
1497 );
1498 }
1499 let outgoing = self.route_user_trades(
1500 &order_response.trades,
1501 ts_init,
1502 );
1503 self.pending_outgoing.extend(outgoing);
1504 return event.map(NautilusWsMessage::OrderUpdated);
1505 } else {
1506 log::warn!(
1507 "Cannot create OrderUpdated: account_id not set"
1508 );
1509 }
1510 } else {
1511 log::warn!(
1512 "Instrument {instrument_name_ustr} not found in cache for edit response"
1513 );
1514 }
1515 }
1516 Err(e) => {
1517 log::error!(
1518 "Failed to parse edit response: request_id={request_id}, error={e}"
1519 );
1520 }
1521 }
1522 } else if let Some(error) = &response.error {
1523 log::warn!(
1524 "Order modify rejected: code={}, message={}, client_order_id={}",
1525 error.code,
1526 error.message,
1527 client_order_id
1528 );
1529 return Some(NautilusWsMessage::OrderModifyRejected(
1530 OrderModifyRejected::new(
1531 trader_id,
1532 strategy_id,
1533 instrument_id,
1534 client_order_id,
1535 ustr::ustr(&format!(
1536 "code={}: {}",
1537 error.code, error.message
1538 )),
1539 UUID4::new(),
1540 ts_init,
1541 ts_init,
1542 false,
1543 None, self.account_id,
1545 ),
1546 ));
1547 }
1548 }
1549 PendingRequestType::GetOrderState {
1550 client_order_id,
1551 trader_id: _,
1552 strategy_id: _,
1553 instrument_id: _,
1554 } => {
1555 if let Some(result) = &response.result {
1556 match serde_json::from_str::<DeribitOrderMsg>(result.get()) {
1557 Ok(order_msg) => {
1558 log::debug!(
1559 "Order state received: venue_order_id={}, client_order_id={}, state={}",
1560 order_msg.order_id,
1561 client_order_id,
1562 order_msg.order_state
1563 );
1564
1565 let instrument_name_ustr = order_msg.instrument_name;
1567
1568 if let Some(instrument) =
1569 self.instruments_cache.get(&instrument_name_ustr)
1570 {
1571 if let Some(account_id) = self.account_id {
1572 match parse_user_order_msg(
1573 &order_msg, instrument, account_id, ts_init,
1574 ) {
1575 Ok(report) => {
1576 return Some(
1577 NautilusWsMessage::OrderStatusReports(
1578 vec![report],
1579 ),
1580 );
1581 }
1582 Err(e) => {
1583 log::warn!(
1584 "Failed to parse get_order_state response to report: {e}"
1585 );
1586 }
1587 }
1588 } else {
1589 log::warn!(
1590 "Cannot create OrderStatusReport: account_id not set"
1591 );
1592 }
1593 } else {
1594 log::warn!(
1595 "Instrument {instrument_name_ustr} not found in cache for get_order_state response"
1596 );
1597 }
1598 }
1599 Err(e) => {
1600 log::error!(
1601 "Failed to parse get_order_state response: request_id={request_id}, error={e}"
1602 );
1603 }
1604 }
1605 } else if let Some(error) = &response.error {
1606 log::error!(
1607 "Get order state failed: code={}, message={}, client_order_id={}",
1608 error.code,
1609 error.message,
1610 client_order_id
1611 );
1612 }
1613 }
1614 }
1615 } else if let Some(request_id) = response.id {
1616 if let Some(error) = &response.error {
1618 log::error!(
1620 "Deribit error for unknown request: code={}, message={}, request_id={}, data={:?}",
1621 error.code,
1622 error.message,
1623 request_id,
1624 error.data
1625 );
1626 return Some(NautilusWsMessage::Error(DeribitWsError::DeribitError {
1627 code: error.code,
1628 message: error.message.clone(),
1629 }));
1630 } else {
1631 log::debug!(
1633 "Received response for unknown request_id={}, result present: {}",
1634 request_id,
1635 response.result.is_some()
1636 );
1637 }
1638 } else if let Some(error) = &response.error {
1639 log::error!(
1641 "Deribit error with no request_id: code={}, message={}, data={:?}",
1642 error.code,
1643 error.message,
1644 error.data
1645 );
1646 return Some(NautilusWsMessage::Error(DeribitWsError::DeribitError {
1647 code: error.code,
1648 message: error.message.clone(),
1649 }));
1650 }
1651 None
1652 }
1653 DeribitWsMessage::Notification(notification) => {
1654 let channel = ¬ification.params.channel;
1655 let data = ¬ification.params.data;
1656
1657 if let Some(channel_type) = DeribitWsChannel::from_channel_string(channel) {
1659 match channel_type {
1660 DeribitWsChannel::Trades => {
1661 match serde_json::from_str::<Vec<DeribitTradeMsg>>(data.get()) {
1663 Ok(trades) => {
1664 log::debug!("Received {} trades", trades.len());
1665 let data_vec = parse_trades_data(
1666 &trades,
1667 &self.instruments_cache,
1668 ts_init,
1669 );
1670
1671 if data_vec.is_empty() && !trades.is_empty() {
1672 let missing: Vec<&Ustr> = trades
1673 .iter()
1674 .map(|t| &t.instrument_name)
1675 .filter(|name| {
1676 !self.instruments_cache.contains_key(name)
1677 })
1678 .collect();
1679
1680 if missing.is_empty() {
1681 log::warn!(
1682 "Received {} trades but parsed 0 (parse failures); cache size: {}",
1683 trades.len(),
1684 self.instruments_cache.len()
1685 );
1686 } else {
1687 log::warn!(
1688 "Trade message received but instrument(s) not found in cache: {:?} (cache size: {})",
1689 missing,
1690 self.instruments_cache.len()
1691 );
1692 }
1693 } else if !data_vec.is_empty() {
1694 log::debug!("Parsed {} trade ticks", data_vec.len());
1695 return Some(NautilusWsMessage::Data(data_vec));
1696 }
1697 }
1698 Err(e) => {
1699 log::warn!("Failed to deserialize trades: {e}");
1700 }
1701 }
1702 }
1703 DeribitWsChannel::Book => {
1704 match serde_json::from_str::<DeribitBookMsg>(data.get()) {
1706 Ok(book_msg) => {
1707 if let Some(instrument) =
1708 self.instruments_cache.get(&book_msg.instrument_name)
1709 {
1710 let inst_name = book_msg.instrument_name.as_str();
1711 let awaiting_resync = self
1712 .pending_book_resync
1713 .iter()
1714 .any(|ch| is_instrument_book_channel(ch, inst_name));
1715
1716 if awaiting_resync
1717 && book_msg.msg_type == DeribitBookMsgType::Change
1718 {
1719 } else if book_msg.msg_type == DeribitBookMsgType::Change
1721 && let Some(prev_id) = book_msg.prev_change_id
1722 && let Some(&last_id) =
1723 self.book_sequence.get(&book_msg.instrument_name)
1724 && prev_id != last_id
1725 {
1726 log::warn!(
1727 "Book sequence gap for {}: expected prev_change_id={}, was {} \
1728 - dropping delta, forcing resync",
1729 book_msg.instrument_name,
1730 last_id,
1731 prev_id
1732 );
1733 self.resync_book(book_msg.instrument_name).await;
1734 } else {
1735 match parse_book_msg(&book_msg, instrument, ts_init) {
1738 Ok(deltas) => {
1739 if awaiting_resync {
1743 self.pending_book_resync
1744 .retain(|ch| ch != channel);
1745 }
1746
1747 self.book_sequence.insert(
1748 book_msg.instrument_name,
1749 book_msg.change_id,
1750 );
1751 return Some(NautilusWsMessage::Deltas(deltas));
1752 }
1753 Err(e) if is_grouped_book_channel(channel) => {
1757 log::error!(
1758 "Failed to parse book snapshot for {} on {channel}: {e}, waiting for the next snapshot",
1759 book_msg.instrument_name
1760 );
1761 }
1762 Err(e) => {
1763 log::error!(
1764 "Failed to parse book message for {}: {e}, forcing resync",
1765 book_msg.instrument_name
1766 );
1767 self.resync_book(book_msg.instrument_name)
1768 .await;
1769 }
1770 }
1771 }
1772 } else {
1773 log::warn!(
1774 "Book message received but instrument '{}' not found in cache (cache size: {})",
1775 book_msg.instrument_name,
1776 self.instruments_cache.len()
1777 );
1778 }
1779 }
1780 Err(e) => {
1781 log::warn!(
1782 "Failed to deserialize book message: {e}, channel: {channel}"
1783 );
1784 }
1785 }
1786 }
1787 DeribitWsChannel::Ticker => {
1788 match serde_json::from_str::<DeribitTickerMsg>(data.get()) {
1789 Ok(ticker_msg) => {
1790 if let Some(instrument) =
1791 self.instruments_cache.get(&ticker_msg.instrument_name)
1792 {
1793 if self.option_greeks_subs.contains(&instrument.id())
1795 && let Some(option_greeks) =
1796 parse_ticker_to_option_greeks(
1797 &ticker_msg,
1798 instrument,
1799 ts_init,
1800 )
1801 {
1802 let _ = self.out_tx.send(
1803 NautilusWsMessage::OptionGreeks(option_greeks),
1804 );
1805 }
1806
1807 let instrument_id = instrument.id();
1808 let mut data_vec = Vec::new();
1809
1810 if self.mark_price_subs.contains(&instrument_id) {
1812 match parse_ticker_to_mark_price(
1813 &ticker_msg,
1814 instrument,
1815 ts_init,
1816 ) {
1817 Ok(mark_price) => {
1818 data_vec.push(Data::MarkPrice(mark_price));
1819 }
1820 Err(e) => {
1821 log::warn!("Failed to parse mark price: {e}");
1822 }
1823 }
1824 }
1825
1826 if self.index_price_subs.contains(&instrument_id) {
1828 match parse_ticker_to_index_price(
1829 &ticker_msg,
1830 instrument,
1831 ts_init,
1832 ) {
1833 Ok(index_price) => {
1834 data_vec.push(Data::IndexPrice(index_price));
1835 }
1836 Err(e) => {
1837 log::warn!("Failed to parse index price: {e}");
1838 }
1839 }
1840 }
1841
1842 if !data_vec.is_empty() {
1843 return Some(NautilusWsMessage::Data(data_vec));
1844 }
1845 } else {
1846 log::warn!(
1847 "Ticker message received but instrument '{}' not found in cache (cache size: {})",
1848 ticker_msg.instrument_name,
1849 self.instruments_cache.len()
1850 );
1851 }
1852 }
1853 Err(e) => {
1854 log::warn!(
1855 "Failed to deserialize ticker message: {e}, channel: {channel}"
1856 );
1857 }
1858 }
1859 }
1860 DeribitWsChannel::Perpetual => {
1861 match serde_json::from_str::<DeribitPerpetualMsg>(data.get()) {
1865 Ok(perpetual_msg) => {
1866 let parts: Vec<&str> = channel.split('.').collect();
1868 if parts.len() >= 2 {
1869 let instrument_name = Ustr::from(parts[1]);
1870
1871 if let Some(instrument) =
1872 self.instruments_cache.get(&instrument_name)
1873 {
1874 let funding_rate = parse_perpetual_to_funding_rate(
1875 &perpetual_msg,
1876 instrument,
1877 ts_init,
1878 );
1879 return Some(NautilusWsMessage::FundingRates(vec![
1880 funding_rate,
1881 ]));
1882 } else {
1883 log::warn!(
1884 "Instrument {} not found in cache (cache size: {})",
1885 instrument_name,
1886 self.instruments_cache.len()
1887 );
1888 }
1889 }
1890 }
1891 Err(e) => {
1892 log::warn!(
1893 "Failed to deserialize perpetual message: {e}, data: {data}"
1894 );
1895 }
1896 }
1897 }
1898 DeribitWsChannel::Quote => {
1899 match serde_json::from_str::<DeribitQuoteMsg>(data.get()) {
1901 Ok(quote_msg) => {
1902 if let Some(instrument) =
1903 self.instruments_cache.get("e_msg.instrument_name)
1904 {
1905 match parse_quote_msg("e_msg, instrument, ts_init) {
1906 Ok(quote) => {
1907 return Some(NautilusWsMessage::Data(vec![
1908 Data::Quote(quote),
1909 ]));
1910 }
1911 Err(e) => {
1912 log::warn!("Failed to parse quote message: {e}");
1913 }
1914 }
1915 } else {
1916 log::warn!(
1917 "Quote message received but instrument '{}' not found in cache (cache size: {})",
1918 quote_msg.instrument_name,
1919 self.instruments_cache.len()
1920 );
1921 }
1922 }
1923 Err(e) => {
1924 log::warn!(
1925 "Failed to deserialize quote message: {e}, channel: {channel}"
1926 );
1927 }
1928 }
1929 }
1930 DeribitWsChannel::VolatilityIndex => {
1931 match serde_json::from_str::<DeribitVolatilityIndexMsg>(data.get()) {
1932 Ok(msg) => {
1933 let ts_event = UnixNanos::from(msg.timestamp * 1_000_000);
1934 let mut metadata = nautilus_core::Params::new();
1935 metadata.insert(
1936 "index_name".to_string(),
1937 serde_json::Value::String(msg.index_name.clone()),
1938 );
1939 let data_type = DataType::new(
1940 "DeribitVolatilityIndex",
1941 Some(metadata),
1942 None,
1943 );
1944
1945 let dvol = DeribitVolatilityIndex::new(
1946 msg.index_name,
1947 msg.volatility,
1948 ts_event,
1949 ts_init,
1950 );
1951
1952 return Some(NautilusWsMessage::Data(vec![Data::Custom(
1953 CustomData::new(Arc::new(dvol), data_type),
1954 )]));
1955 }
1956 Err(e) => {
1957 log::warn!("Failed to deserialize volatility index: {e}");
1958 }
1959 }
1960 }
1961 DeribitWsChannel::InstrumentState => {
1962 match serde_json::from_str::<DeribitInstrumentStateMsg>(data.get()) {
1963 Ok(state_msg) => {
1964 log::debug!(
1965 "Instrument state change: {} -> {} (timestamp: {})",
1966 state_msg.instrument_name,
1967 state_msg.state,
1968 state_msg.timestamp
1969 );
1970
1971 let instrument_id = if let Some(instrument) =
1972 self.instruments_cache.get(&state_msg.instrument_name)
1973 {
1974 instrument.id()
1975 } else {
1976 log::debug!(
1977 "Instrument '{}' not in cache, constructing ID",
1978 state_msg.instrument_name
1979 );
1980 InstrumentId::new(
1981 Symbol::new(state_msg.instrument_name),
1982 *DERIBIT_VENUE,
1983 )
1984 };
1985
1986 let action = MarketStatusAction::from(state_msg.state);
1987 let is_trading =
1988 Some(state_msg.state == DeribitInstrumentState::Started);
1989 let ts_event = UnixNanos::from(state_msg.timestamp * 1_000_000);
1990 let status = InstrumentStatus::new(
1991 instrument_id,
1992 action,
1993 ts_event,
1994 ts_init,
1995 None,
1996 None,
1997 is_trading,
1998 None,
1999 None,
2000 );
2001 return Some(NautilusWsMessage::InstrumentStatus(status));
2002 }
2003 Err(e) => {
2004 log::warn!("Failed to parse instrument status message: {e}");
2005 }
2006 }
2007 }
2008 DeribitWsChannel::ChartTrades => {
2009 if let Ok(chart_msg) =
2014 serde_json::from_str::<DeribitChartMsg>(data.get())
2015 {
2016 let parts: Vec<&str> = channel.split('.').collect();
2019 if parts.len() >= 4 {
2020 let instrument_name = Ustr::from(parts[2]);
2021 let resolution = parts[3];
2022
2023 if let Some(instrument) =
2024 self.instruments_cache.get(&instrument_name)
2025 {
2026 let instrument_id = instrument.id();
2027
2028 match resolution_to_bar_type(instrument_id, resolution) {
2029 Ok(bar_type) => {
2030 let price_precision = instrument.price_precision();
2031 let size_precision = instrument.size_precision();
2032 let use_cost_for_volume =
2033 use_cost_for_bar_volume(instrument);
2034
2035 match parse_chart_msg(
2036 &chart_msg,
2037 bar_type,
2038 price_precision,
2039 size_precision,
2040 use_cost_for_volume,
2041 self.bars_timestamp_on_close,
2042 ts_init,
2043 ) {
2044 Ok(new_bar) => {
2045 let channel_key = channel.clone();
2047
2048 if let Some(pending_bar) =
2049 self.pending_bars.get(&channel_key)
2050 {
2051 if new_bar.ts_event
2053 != pending_bar.ts_event
2054 {
2055 let closed_bar = *pending_bar;
2056 self.pending_bars
2057 .insert(channel_key, new_bar);
2058 log::debug!(
2059 "Emitting closed bar: {closed_bar:?}"
2060 );
2061 return Some(
2062 NautilusWsMessage::Data(vec![
2063 Data::Bar(closed_bar),
2064 ]),
2065 );
2066 }
2067 self.pending_bars
2069 .insert(channel_key, new_bar);
2070 } else {
2071 self.pending_bars
2073 .insert(channel_key, new_bar);
2074 }
2075 }
2076 Err(e) => {
2077 log::warn!(
2078 "Failed to parse chart message to bar: {e}"
2079 );
2080 }
2081 }
2082 }
2083 Err(e) => {
2084 log::warn!(
2085 "Failed to create BarType from resolution {resolution}: {e}"
2086 );
2087 }
2088 }
2089 } else {
2090 log::warn!(
2091 "Instrument {instrument_name} not found in cache for chart data"
2092 );
2093 }
2094 }
2095 }
2096 }
2097 DeribitWsChannel::UserOrders => {
2098 let orders_result = serde_json::from_str::<Vec<DeribitOrderMsg>>(
2100 data.get(),
2101 )
2102 .or_else(|_| {
2103 serde_json::from_str::<DeribitOrderMsg>(data.get())
2104 .map(|order| vec![order])
2105 });
2106
2107 match orders_result {
2108 Ok(orders) => {
2109 log::debug!("Received {} user order updates", orders.len());
2110
2111 let Some(account_id) = self.account_id else {
2113 log::warn!("Cannot parse user orders: account_id not set");
2114 return Some(NautilusWsMessage::Raw(data.clone()));
2115 };
2116
2117 let mut outgoing = Vec::new();
2118
2119 for order in &orders {
2121 let venue_order_id_str = &order.order_id;
2122 let venue_order_id =
2123 VenueOrderId::new(venue_order_id_str.as_str());
2124 let instrument_name = order.instrument_name;
2125
2126 let Some(instrument) =
2127 self.instruments_cache.get(&instrument_name)
2128 else {
2129 log::warn!(
2130 "Instrument {instrument_name} not found in cache"
2131 );
2132 continue;
2133 };
2134
2135 let label_client_order_id = order
2136 .label
2137 .as_ref()
2138 .filter(|l| !l.is_empty())
2139 .map(ClientOrderId::new);
2140 let was_terminal = self
2141 .terminal_order_contexts
2142 .contains_key(&venue_order_id);
2143 let Some(mut context) = self.find_order_context(
2144 venue_order_id,
2145 label_client_order_id,
2146 ) else {
2147 match parse_user_order_msg(
2148 order, instrument, account_id, ts_init,
2149 ) {
2150 Ok(report) => outgoing.push(
2151 NautilusWsMessage::OrderStatusReports(vec![
2152 report,
2153 ]),
2154 ),
2155 Err(e) => log::warn!(
2156 "Failed to parse external order update: {e}"
2157 ),
2158 }
2159 continue;
2160 };
2161
2162 let signature = Self::order_signature(order);
2163
2164 let event_type = determine_order_event_type(
2166 &order.order_state,
2167 !context.accepted,
2168 order.replaced
2169 && context.last_order_signature != Some(signature),
2170 );
2171 context.last_order_signature = Some(signature);
2172
2173 let trader_id = context.trader_id;
2174 let strategy_id = context.strategy_id;
2175 let client_order_id = context.client_order_id;
2176
2177 match event_type {
2178 OrderEventType::Accepted => {
2179 if self
2181 .terminal_order_contexts
2182 .contains_key(&venue_order_id)
2183 {
2184 log::debug!(
2185 "Skipping OrderAccepted for terminal order: client_order_id={client_order_id}"
2186 );
2187 continue;
2188 }
2189
2190 let event =
2191 parse_order_accepted_with_client_order_id(
2192 order,
2193 instrument,
2194 account_id,
2195 trader_id,
2196 strategy_id,
2197 client_order_id,
2198 ts_init,
2199 );
2200 context.accepted = true;
2201 self.bind_order_context(venue_order_id, context);
2202
2203 log::debug!(
2204 "Emitting OrderAccepted: venue_order_id={venue_order_id}"
2205 );
2206 outgoing
2207 .push(NautilusWsMessage::OrderAccepted(event));
2208 }
2209 OrderEventType::Canceled => {
2210 if self
2213 .terminal_order_contexts
2214 .contains_key(&venue_order_id)
2215 {
2216 log::trace!(
2217 "Skipping duplicate OrderCanceled: client_order_id={client_order_id}"
2218 );
2219 continue;
2220 }
2221
2222 if !context.accepted {
2223 outgoing
2224 .push(NautilusWsMessage::OrderAccepted(
2225 parse_order_accepted_with_client_order_id(
2226 order,
2227 instrument,
2228 account_id,
2229 trader_id,
2230 strategy_id,
2231 client_order_id,
2232 ts_init,
2233 ),
2234 ));
2235 context.accepted = true;
2236 }
2237
2238 let event =
2239 parse_order_canceled_with_client_order_id(
2240 order,
2241 instrument,
2242 account_id,
2243 trader_id,
2244 strategy_id,
2245 client_order_id,
2246 ts_init,
2247 );
2248 log::debug!(
2249 "Emitting OrderCanceled: venue_order_id={venue_order_id}"
2250 );
2251 self.finish_order_context(venue_order_id, &context);
2252 outgoing
2253 .push(NautilusWsMessage::OrderCanceled(event));
2254 }
2255 OrderEventType::Expired => {
2256 if self
2257 .terminal_order_contexts
2258 .contains_key(&venue_order_id)
2259 {
2260 log::trace!(
2261 "Skipping duplicate OrderExpired: client_order_id={client_order_id}"
2262 );
2263 continue;
2264 }
2265
2266 if !context.accepted {
2267 outgoing
2268 .push(NautilusWsMessage::OrderAccepted(
2269 parse_order_accepted_with_client_order_id(
2270 order,
2271 instrument,
2272 account_id,
2273 trader_id,
2274 strategy_id,
2275 client_order_id,
2276 ts_init,
2277 ),
2278 ));
2279 context.accepted = true;
2280 }
2281
2282 let event =
2283 parse_order_expired_with_client_order_id(
2284 order,
2285 instrument,
2286 account_id,
2287 trader_id,
2288 strategy_id,
2289 client_order_id,
2290 ts_init,
2291 );
2292 log::debug!(
2293 "Emitting OrderExpired: venue_order_id={venue_order_id}"
2294 );
2295 self.finish_order_context(venue_order_id, &context);
2296 outgoing
2297 .push(NautilusWsMessage::OrderExpired(event));
2298 }
2299 OrderEventType::Updated => {
2300 if was_terminal {
2301 log::trace!(
2302 "Skipping amendment for terminal order: venue_order_id={venue_order_id}"
2303 );
2304 continue;
2305 }
2306
2307 if !context.accepted {
2308 outgoing
2309 .push(NautilusWsMessage::OrderAccepted(
2310 parse_order_accepted_with_client_order_id(
2311 order,
2312 instrument,
2313 account_id,
2314 trader_id,
2315 strategy_id,
2316 client_order_id,
2317 ts_init,
2318 ),
2319 ));
2320 context.accepted = true;
2321 }
2322
2323 let event =
2324 parse_order_updated_with_client_order_id(
2325 order,
2326 instrument,
2327 account_id,
2328 trader_id,
2329 strategy_id,
2330 client_order_id,
2331 ts_init,
2332 );
2333 self.bind_order_context(venue_order_id, context);
2334 log::debug!(
2335 "Emitting OrderUpdated: venue_order_id={venue_order_id}"
2336 );
2337 outgoing
2338 .push(NautilusWsMessage::OrderUpdated(event));
2339 }
2340 OrderEventType::None => {
2341 if order.order_state == "filled" {
2344 log::debug!(
2345 "Recording terminal order: venue_order_id={venue_order_id}, state={}",
2346 order.order_state
2347 );
2348 self.finish_order_context(
2349 venue_order_id,
2350 &context,
2351 );
2352 } else if order.order_state == "rejected" {
2353 log::debug!(
2354 "Recording rejected order: venue_order_id={venue_order_id}"
2355 );
2356 self.finish_order_context(
2357 venue_order_id,
2358 &context,
2359 );
2360 } else if was_terminal {
2361 self.finish_order_context(
2362 venue_order_id,
2363 &context,
2364 );
2365 } else {
2366 log::trace!(
2367 "No event to emit for order {}, state={}",
2368 venue_order_id,
2369 order.order_state
2370 );
2371 self.bind_order_context(
2372 venue_order_id,
2373 context,
2374 );
2375 }
2376 }
2377 }
2378 }
2379
2380 if !outgoing.is_empty() {
2381 self.pending_outgoing.extend(outgoing);
2382 }
2383 }
2384 Err(e) => {
2385 log::warn!("Failed to deserialize user orders: {e}");
2386 }
2387 }
2388 }
2389 DeribitWsChannel::UserTrades => {
2390 let trades_result =
2392 serde_json::from_str::<Vec<DeribitUserTradeMsg>>(data.get())
2393 .or_else(|_| {
2394 serde_json::from_str::<DeribitUserTradeMsg>(data.get())
2395 .map(|trade| vec![trade])
2396 });
2397
2398 match trades_result {
2399 Ok(trades) => {
2400 log::debug!("Received {} user trade updates", trades.len());
2401 if self.account_id.is_none() {
2402 log::warn!("Cannot parse user trades: account_id not set");
2403 return Some(NautilusWsMessage::Raw(data.clone()));
2404 }
2405 let outgoing = self.route_user_trades(&trades, ts_init);
2406 if !outgoing.is_empty() {
2407 self.pending_outgoing.extend(outgoing);
2408 }
2409 }
2410 Err(e) => {
2411 log::warn!("Failed to deserialize user trades: {e}");
2412 }
2413 }
2414 }
2415 DeribitWsChannel::UserPortfolio => {
2416 match serde_json::from_str::<DeribitPortfolioMsg>(data.get()) {
2417 Ok(portfolio) => {
2418 if portfolio.equity.is_zero() && portfolio.balance.is_zero() {
2422 log::trace!(
2423 "Skipping zero-balance portfolio for {}",
2424 portfolio.currency
2425 );
2426 return None;
2427 }
2428
2429 let Some(account_id) = self.account_id else {
2431 log::warn!("Cannot parse portfolio: account_id not set");
2432 return None;
2433 };
2434
2435 match parse_portfolio_to_account_state(
2436 &portfolio, account_id, ts_init,
2437 ) {
2438 Ok(account_state) => {
2439 let currency_key = portfolio.currency.clone();
2441
2442 if let Some(last) =
2443 self.last_account_states.get(¤cy_key)
2444 && account_state.has_same_balances_and_margins(last)
2445 {
2446 log::trace!(
2447 "Skipping duplicate portfolio update for {}",
2448 portfolio.currency
2449 );
2450 return None;
2451 }
2452
2453 self.last_account_states
2454 .insert(currency_key, account_state.clone());
2455 return Some(NautilusWsMessage::AccountState(
2456 account_state,
2457 ));
2458 }
2459 Err(e) => {
2460 log::warn!(
2461 "Failed to parse portfolio to AccountState: {e}"
2462 );
2463 }
2464 }
2465 }
2466 Err(e) => {
2467 log::warn!("Failed to deserialize portfolio: {e}");
2468 }
2469 }
2470 }
2471 _ => {
2472 log::trace!("Unhandled channel: {channel}");
2474 return Some(NautilusWsMessage::Raw(data.clone()));
2475 }
2476 }
2477 } else {
2478 log::trace!("Unknown channel: {channel}");
2479 return Some(NautilusWsMessage::Raw(data.clone()));
2480 }
2481 None
2482 }
2483 DeribitWsMessage::Heartbeat(heartbeat) => {
2484 match heartbeat.heartbeat_type {
2485 DeribitHeartbeatType::TestRequest => {
2486 log::trace!(
2487 "Received heartbeat test_request - responding with public/test"
2488 );
2489
2490 if let Err(e) = self.handle_heartbeat_test_request().await {
2491 log::error!("Failed to respond to heartbeat test_request: {e}");
2492
2493 return Some(NautilusWsMessage::Error(DeribitWsError::Send(format!(
2495 "Heartbeat response failed: {e}"
2496 ))));
2497 }
2498 }
2499 DeribitHeartbeatType::Heartbeat => {
2500 log::trace!("Received heartbeat acknowledgment");
2501 }
2502 }
2503 None
2504 }
2505 DeribitWsMessage::Error(err) => {
2506 log::error!("Deribit error {}: {}", err.code, err.message);
2507 Some(NautilusWsMessage::Error(DeribitWsError::DeribitError {
2508 code: err.code,
2509 message: err.message,
2510 }))
2511 }
2512 DeribitWsMessage::Reconnected => Some(NautilusWsMessage::Reconnected),
2513 }
2514 }
2515
2516 async fn resync_book(&mut self, instrument_name: Ustr) {
2520 self.book_sequence.remove(&instrument_name);
2521
2522 let book_channels: Vec<String> = self
2523 .subscriptions_state
2524 .all_topics()
2525 .into_iter()
2526 .filter(|t| {
2527 is_instrument_book_channel(t, &instrument_name) && !is_grouped_book_channel(t)
2528 })
2529 .collect();
2530
2531 if book_channels.is_empty() {
2532 return;
2533 }
2534
2535 for ch in &book_channels {
2536 self.subscriptions_state.mark_failure(ch);
2537
2538 if !self.pending_book_resync.contains(ch) {
2539 self.pending_book_resync.push(ch.clone());
2540 }
2541 }
2542
2543 let _ = self.handle_unsubscribe(book_channels).await;
2545 }
2546
2547 fn order_signature(order: &DeribitOrderMsg) -> OrderSignature {
2548 (order.amount, order.price, order.trigger_price)
2549 }
2550
2551 fn find_order_context(
2552 &self,
2553 venue_order_id: VenueOrderId,
2554 client_order_id: Option<ClientOrderId>,
2555 ) -> Option<OrderContext> {
2556 self.order_contexts
2557 .get(&venue_order_id)
2558 .or_else(|| self.terminal_order_contexts.get(&venue_order_id))
2559 .cloned()
2560 .or_else(|| {
2561 client_order_id.and_then(|client_order_id| {
2562 self.submitted_order_contexts.get(&client_order_id).cloned()
2563 })
2564 })
2565 }
2566
2567 fn bind_order_context(&mut self, venue_order_id: VenueOrderId, context: OrderContext) {
2568 self.submitted_order_contexts
2569 .remove(&context.client_order_id);
2570 self.terminal_order_contexts.remove(&venue_order_id);
2571 self.order_contexts.insert(venue_order_id, context);
2572 }
2573
2574 fn finish_order_context(&mut self, venue_order_id: VenueOrderId, context: &OrderContext) {
2575 self.order_contexts.remove(&venue_order_id);
2576 self.submitted_order_contexts
2577 .remove(&context.client_order_id);
2578 self.terminal_order_contexts
2579 .insert(venue_order_id, context.clone());
2580 }
2581
2582 fn route_user_trades(
2583 &mut self,
2584 trades: &[DeribitUserTradeMsg],
2585 ts_init: UnixNanos,
2586 ) -> Vec<NautilusWsMessage> {
2587 let Some(account_id) = self.account_id else {
2588 log::warn!("Cannot parse user trades: account_id not set");
2589 return Vec::new();
2590 };
2591
2592 let mut outgoing = Vec::with_capacity(trades.len() + 1);
2593 let mut reports = Vec::new();
2594
2595 for trade in trades {
2596 let instrument_name = trade.instrument_name;
2597 let Some((report, quote_currency)) =
2598 self.instruments_cache
2599 .get(&instrument_name)
2600 .map(|instrument| {
2601 (
2602 parse_user_trade_msg(trade, instrument, account_id, ts_init),
2603 instrument.quote_currency(),
2604 )
2605 })
2606 else {
2607 log::warn!("Instrument {instrument_name} not found in cache");
2608 continue;
2609 };
2610
2611 let report = match report {
2612 Ok(report) => report,
2613 Err(e) => {
2614 log::warn!("Failed to parse trade {}: {e}", trade.trade_id);
2615 continue;
2616 }
2617 };
2618 let venue_order_id = report.venue_order_id;
2619 let was_terminal = self.terminal_order_contexts.contains_key(&venue_order_id);
2620 let Some(mut context) = self.find_order_context(venue_order_id, report.client_order_id)
2621 else {
2622 log::debug!(
2623 "Parsed external fill report: {} @ {}",
2624 report.trade_id,
2625 report.last_px
2626 );
2627 reports.push(report);
2628 continue;
2629 };
2630
2631 if !context.accepted {
2632 outgoing.push(NautilusWsMessage::OrderAccepted(OrderAccepted::new(
2633 context.trader_id,
2634 context.strategy_id,
2635 context.instrument_id,
2636 context.client_order_id,
2637 venue_order_id,
2638 account_id,
2639 UUID4::new(),
2640 report.ts_event,
2641 report.ts_init,
2642 false,
2643 )));
2644 context.accepted = true;
2645 }
2646
2647 log::debug!(
2648 "Parsed tracked fill event: {} @ {}",
2649 report.trade_id,
2650 report.last_px
2651 );
2652 outgoing.push(NautilusWsMessage::OrderFilled(OrderFilled::new(
2653 context.trader_id,
2654 context.strategy_id,
2655 context.instrument_id,
2656 context.client_order_id,
2657 venue_order_id,
2658 account_id,
2659 report.trade_id,
2660 context.order_side,
2661 context.order_type,
2662 report.last_qty,
2663 report.last_px,
2664 quote_currency,
2665 report.liquidity_side,
2666 UUID4::new(),
2667 report.ts_event,
2668 report.ts_init,
2669 false,
2670 report.venue_position_id,
2671 Some(report.commission),
2672 None,
2673 )));
2674
2675 if was_terminal || trade.state == "filled" {
2676 self.finish_order_context(venue_order_id, &context);
2677 } else {
2678 self.bind_order_context(venue_order_id, context);
2679 }
2680 }
2681
2682 outgoing.retain(|message| match message {
2684 NautilusWsMessage::OrderFilled(fill) => self
2685 .emitted_trades
2686 .insert((fill.venue_order_id, fill.trade_id)),
2687 _ => true,
2688 });
2689
2690 reports.retain(|report| {
2691 self.emitted_trades
2692 .insert((report.venue_order_id, report.trade_id))
2693 });
2694
2695 if !reports.is_empty() {
2696 outgoing.push(NautilusWsMessage::FillReports(reports));
2697 }
2698 outgoing
2699 }
2700
2701 pub async fn next(&mut self) -> Option<NautilusWsMessage> {
2707 loop {
2708 if let Some(msg) = self.pending_outgoing.pop_front() {
2709 match msg {
2710 NautilusWsMessage::Reconnected
2711 | NautilusWsMessage::Authenticated(_)
2712 | NautilusWsMessage::AuthenticationFailed(_) => {
2713 return Some(msg);
2714 }
2715 _ => {
2716 let _ = self.out_tx.send(msg);
2717 continue;
2718 }
2719 }
2720 }
2721
2722 tokio::select! {
2723 Some(cmd) = self.cmd_rx.recv() => {
2725 self.process_command(cmd).await;
2726 }
2727 Some(msg) = self.raw_rx.recv() => {
2729 match msg {
2730 Message::Text(text) => {
2731 if let Some(nautilus_msg) = self.process_raw_message(&text).await {
2732 match &nautilus_msg {
2734 NautilusWsMessage::Data(_)
2735 | NautilusWsMessage::Deltas(_)
2736 | NautilusWsMessage::Instrument(_)
2737 | NautilusWsMessage::InstrumentStatus(_)
2738 | NautilusWsMessage::OptionGreeks(_)
2739 | NautilusWsMessage::Raw(_)
2740 | NautilusWsMessage::Error(_) => {
2741 let _ = self.out_tx.send(nautilus_msg);
2742 }
2743 NautilusWsMessage::FundingRates(rates) => {
2744 let msg_to_send =
2745 NautilusWsMessage::FundingRates(rates.clone());
2746
2747 if let Err(e) = self.out_tx.send(msg_to_send) {
2748 log::error!("Failed to send funding rates: {e}");
2749 }
2750 }
2751 NautilusWsMessage::OrderStatusReports(_)
2752 | NautilusWsMessage::FillReports(_)
2753 | NautilusWsMessage::OrderFilled(_)
2754 | NautilusWsMessage::OrderAccepted(_)
2755 | NautilusWsMessage::OrderCanceled(_)
2756 | NautilusWsMessage::OrderExpired(_)
2757 | NautilusWsMessage::OrderUpdated(_)
2758 | NautilusWsMessage::OrderRejected(_)
2759 | NautilusWsMessage::OrderCancelRejected(_)
2760 | NautilusWsMessage::OrderModifyRejected(_)
2761 | NautilusWsMessage::AccountState(_) => {
2762 let _ = self.out_tx.send(nautilus_msg);
2763 }
2764 NautilusWsMessage::Reconnected
2766 | NautilusWsMessage::Authenticated(_)
2767 | NautilusWsMessage::AuthenticationFailed(_) => {
2768 return Some(nautilus_msg);
2769 }
2770 }
2771 }
2772 }
2773 Message::Ping(data) => {
2774 if let Some(client) = &self.inner {
2776 let _ = client.send_pong(data.to_vec()).await;
2777 }
2778 }
2779 Message::Close(_) => {
2780 log::debug!("Received close frame");
2781 }
2782 _ => {}
2783 }
2784 }
2785 () = tokio::time::sleep(tokio::time::Duration::from_millis(100)) => {
2787 if self.signal.load(Ordering::Relaxed) {
2788 log::debug!("Stop signal received");
2789 return None;
2790 }
2791 }
2792 }
2793 }
2794 }
2795}
2796
2797fn is_instrument_book_channel(channel: &str, instrument_name: &str) -> bool {
2798 channel.starts_with("book.")
2799 && channel
2800 .split('.')
2801 .nth(1)
2802 .is_some_and(|s| s == instrument_name)
2803}
2804
2805fn is_grouped_book_channel(channel: &str) -> bool {
2807 channel.starts_with("book.") && channel.split('.').count() == 5
2808}
2809
2810#[cfg(test)]
2811mod tests {
2812 use nautilus_model::{
2813 enums::LiquiditySide,
2814 instruments::Instrument,
2815 types::{Money, Price, Quantity},
2816 };
2817 use rstest::rstest;
2818
2819 use super::*;
2820 use crate::{
2821 common::{parse::parse_deribit_instrument_any, testing::load_test_json},
2822 http::models::{DeribitInstrument, DeribitJsonRpcResponse},
2823 };
2824
2825 fn routing_test_handler() -> DeribitWsFeedHandler {
2826 let signal = Arc::new(AtomicBool::new(false));
2827 let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
2828 let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
2829 let (out_tx, _out_rx) = tokio::sync::mpsc::unbounded_channel();
2830 let mut handler = DeribitWsFeedHandler::new(
2831 signal,
2832 cmd_rx,
2833 raw_rx,
2834 out_tx,
2835 AuthTracker::new(),
2836 SubscriptionState::new('.'),
2837 Arc::new(AtomicSet::new()),
2838 Arc::new(AtomicSet::new()),
2839 Arc::new(AtomicSet::new()),
2840 Some(AccountId::from("DERIBIT-001")),
2841 true,
2842 Arc::new(Mutex::new(Vec::new())),
2843 );
2844 let json = load_test_json("http_get_instruments.json");
2845 let response: DeribitJsonRpcResponse<Vec<DeribitInstrument>> =
2846 serde_json::from_str(&json).unwrap();
2847 let instrument = parse_deribit_instrument_any(
2848 &response.result.unwrap()[0],
2849 UnixNanos::default(),
2850 UnixNanos::default(),
2851 )
2852 .unwrap()
2853 .unwrap();
2854 handler
2855 .instruments_cache
2856 .insert(instrument.raw_symbol().inner(), instrument);
2857 handler
2858 }
2859
2860 fn order_data(order_state: &str, replaced: bool) -> serde_json::Value {
2861 serde_json::json!({
2862 "order_id": "ETH-584830574",
2863 "label": "O-19700101-000000-001-001-1",
2864 "instrument_name": "BTC-PERPETUAL",
2865 "direction": "buy",
2866 "order_type": "market",
2867 "order_state": order_state,
2868 "replaced": replaced,
2869 "price": 203.8,
2870 "amount": 2.0,
2871 "filled_amount": if order_state == "filled" { 2.0 } else { 0.0 },
2872 "average_price": 203.8,
2873 "creation_timestamp": 1_590_480_712_700_u64,
2874 "last_update_timestamp": 1_590_480_712_800_u64,
2875 "time_in_force": "good_til_cancelled",
2876 "commission": 0.00073602,
2877 "post_only": false,
2878 "reduce_only": false,
2879 "trigger_price": null,
2880 "trigger": null,
2881 "max_show": null,
2882 "api": true,
2883 "reject_reason": null,
2884 "cancel_reason": null
2885 })
2886 }
2887
2888 fn trade_data() -> serde_json::Value {
2889 serde_json::json!({
2890 "trade_id": "ETH-2696068",
2891 "order_id": "ETH-584830574",
2892 "instrument_name": "BTC-PERPETUAL",
2893 "direction": "buy",
2894 "price": 203.8,
2895 "amount": 2.0,
2896 "fee": 0.00073602,
2897 "fee_currency": "USDT",
2898 "timestamp": 1_590_480_712_800_u64,
2899 "trade_seq": 1_966_042_u64,
2900 "liquidity": "T",
2901 "order_type": "market",
2902 "index_price": 203.89,
2903 "mark_price": 203.78,
2904 "tick_direction": 3,
2905 "state": "filled",
2906 "label": "O-19700101-000000-001-001-1",
2907 "reduce_only": false,
2908 "post_only": false,
2909 "liquidation": null,
2910 "profit_loss": null
2911 })
2912 }
2913
2914 fn subscription(channel: &str, data: &serde_json::Value) -> String {
2915 serde_json::json!({
2916 "jsonrpc": "2.0",
2917 "method": "subscription",
2918 "params": { "channel": channel, "data": data }
2919 })
2920 .to_string()
2921 }
2922
2923 #[rstest]
2924 #[tokio::test]
2925 async fn tracked_fast_fill_synthesizes_accepted_before_filled() {
2926 let mut handler = routing_test_handler();
2927 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
2928 let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
2929 handler.submitted_order_contexts.insert(
2930 client_order_id,
2931 OrderContext {
2932 client_order_id,
2933 trader_id: TraderId::from("TRADER-001"),
2934 strategy_id: StrategyId::from("S-001"),
2935 instrument_id,
2936 order_side: OrderSide::Buy,
2937 order_type: OrderType::Market,
2938 accepted: false,
2939 last_order_signature: None,
2940 },
2941 );
2942 handler.pending_requests.insert(
2943 42,
2944 PendingRequestType::Buy {
2945 client_order_id,
2946 trader_id: TraderId::from("TRADER-001"),
2947 strategy_id: StrategyId::from("S-001"),
2948 instrument_id,
2949 order_side: OrderSide::Buy,
2950 order_type: OrderType::Market,
2951 },
2952 );
2953 let response = serde_json::json!({
2954 "jsonrpc": "2.0",
2955 "id": 42,
2956 "result": { "order": order_data("filled", false), "trades": [] }
2957 });
2958
2959 handler.process_raw_message(&response.to_string()).await;
2960 assert!(handler.pending_outgoing.is_empty());
2961 assert!(
2962 !handler
2963 .terminal_order_contexts
2964 .get(&VenueOrderId::from("ETH-584830574"))
2965 .unwrap()
2966 .accepted
2967 );
2968
2969 handler
2970 .process_raw_message(&subscription(
2971 "user.trades.any.any.raw",
2972 &serde_json::json!([trade_data()]),
2973 ))
2974 .await;
2975
2976 assert!(matches!(
2977 handler.pending_outgoing.pop_front().unwrap(),
2978 NautilusWsMessage::OrderAccepted(event)
2979 if event.client_order_id == client_order_id
2980 ));
2981 assert!(matches!(
2982 handler.pending_outgoing.pop_front().unwrap(),
2983 NautilusWsMessage::OrderFilled(event)
2984 if event.client_order_id == client_order_id
2985 && event.trade_id.to_string() == "ETH-2696068"
2986 && event.order_side == OrderSide::Buy
2987 && event.order_type == OrderType::Market
2988 && event.last_qty.to_string() == "2"
2989 && event.last_px.to_string() == "203.8"
2990 && event.liquidity_side == LiquiditySide::Taker
2991 && event.commission == Some(Money::from("0.00073602 USDT"))
2992 ));
2993 assert!(handler.pending_outgoing.is_empty());
2994
2995 handler
2996 .process_raw_message(&subscription(
2997 "user.trades.any.any.raw",
2998 &serde_json::json!([trade_data()]),
2999 ))
3000 .await;
3001
3002 assert!(handler.pending_outgoing.is_empty());
3003 }
3004
3005 #[rstest]
3006 #[case::response_first(false)]
3007 #[case::subscription_first(true)]
3008 #[tokio::test]
3009 async fn submit_response_trades_use_tracked_event_path(
3010 #[case] subscription_first: bool,
3011 #[values("filled", "open")] trade_state: &str,
3012 ) {
3013 let mut handler = routing_test_handler();
3014 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3015 let context = OrderContext {
3016 client_order_id,
3017 trader_id: TraderId::from("TRADER-001"),
3018 strategy_id: StrategyId::from("S-001"),
3019 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3020 order_side: OrderSide::Buy,
3021 order_type: OrderType::Market,
3022 accepted: false,
3023 last_order_signature: None,
3024 };
3025 handler
3026 .submitted_order_contexts
3027 .insert(client_order_id, context.clone());
3028 handler.pending_requests.insert(
3029 42,
3030 PendingRequestType::Buy {
3031 client_order_id,
3032 trader_id: context.trader_id,
3033 strategy_id: context.strategy_id,
3034 instrument_id: context.instrument_id,
3035 order_side: context.order_side,
3036 order_type: context.order_type,
3037 },
3038 );
3039
3040 let mut trade = trade_data();
3041 trade["state"] = serde_json::json!(trade_state);
3042 trade["amount"] = serde_json::json!(if trade_state == "filled" { 2 } else { 1 });
3043 let mut order = order_data(trade_state, false);
3044 order["filled_amount"] = trade["amount"].clone();
3045 let response = serde_json::json!({
3046 "jsonrpc": "2.0",
3047 "id": 42,
3048 "result": { "order": order, "trades": [trade.clone()] }
3049 });
3050 let trade_update = subscription("user.trades.any.any.raw", &serde_json::json!([trade]));
3051
3052 let (raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
3053 let (out_tx, mut out_rx) = tokio::sync::mpsc::unbounded_channel();
3054 handler.raw_rx = raw_rx;
3055 handler.out_tx = out_tx;
3056 let response = response.to_string();
3057 let messages = if subscription_first {
3058 [trade_update, response]
3059 } else {
3060 [response, trade_update]
3061 };
3062
3063 for message in messages {
3064 raw_tx.send(Message::Text(message.into())).unwrap();
3065 }
3066
3067 raw_tx.send(Message::Text(RECONNECTED.into())).unwrap();
3068 let reconnected = tokio::time::timeout(std::time::Duration::from_secs(2), handler.next())
3069 .await
3070 .unwrap();
3071
3072 assert!(matches!(reconnected, Some(NautilusWsMessage::Reconnected)));
3073 assert!(matches!(
3074 out_rx.try_recv().unwrap(),
3075 NautilusWsMessage::OrderAccepted(event)
3076 if event.client_order_id == client_order_id
3077 ));
3078 assert!(matches!(
3079 out_rx.try_recv().unwrap(),
3080 NautilusWsMessage::OrderFilled(event)
3081 if event.client_order_id == client_order_id
3082 && event.trade_id.to_string() == "ETH-2696068"
3083 && event.last_qty.to_string() == if trade_state == "filled" { "2" } else { "1" }
3084 ));
3085 assert!(out_rx.try_recv().is_err());
3086 assert!(handler.pending_outgoing.is_empty());
3087 let venue_order_id = VenueOrderId::from("ETH-584830574");
3088 assert_eq!(
3089 handler
3090 .terminal_order_contexts
3091 .contains_key(&venue_order_id),
3092 trade_state == "filled",
3093 );
3094 assert_eq!(
3095 handler.order_contexts.contains_key(&venue_order_id),
3096 trade_state == "open",
3097 );
3098 }
3099
3100 #[rstest]
3101 #[case::tracked(true)]
3102 #[case::external(false)]
3103 #[tokio::test]
3104 async fn trade_replays_after_reconnect_preserve_new_fills(#[case] tracked: bool) {
3105 let mut handler = routing_test_handler();
3106 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3107 let venue_order_id = VenueOrderId::from("ETH-584830574");
3108
3109 if tracked {
3110 handler.order_contexts.insert(
3111 venue_order_id,
3112 OrderContext {
3113 client_order_id,
3114 trader_id: TraderId::from("TRADER-001"),
3115 strategy_id: StrategyId::from("S-001"),
3116 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3117 order_side: OrderSide::Buy,
3118 order_type: OrderType::Market,
3119 accepted: true,
3120 last_order_signature: None,
3121 },
3122 );
3123 }
3124
3125 let mut first_trade = trade_data();
3126 first_trade["amount"] = serde_json::json!(1);
3127 let mut second_trade = first_trade.clone();
3128 second_trade["trade_id"] = serde_json::json!("ETH-2696069");
3129 second_trade["price"] = serde_json::json!(204.1);
3130 second_trade["fee"] = serde_json::json!(0.00083602);
3131 let first_update = subscription(
3132 "user.trades.any.any.raw",
3133 &serde_json::json!([first_trade.clone(), first_trade.clone()]),
3134 );
3135 let replay_update = subscription(
3136 "user.trades.any.any.raw",
3137 &serde_json::json!([first_trade, second_trade.clone(), second_trade]),
3138 );
3139
3140 handler.process_raw_message(&first_update).await;
3141 let first = handler.pending_outgoing.pop_front().unwrap();
3142 assert!(handler.pending_outgoing.is_empty());
3143 let reconnected = handler.process_raw_message(RECONNECTED).await;
3144 handler.process_raw_message(&replay_update).await;
3145 let second = handler.pending_outgoing.pop_front().unwrap();
3146
3147 let fills = [first, second].map(|message| match message {
3148 NautilusWsMessage::OrderFilled(fill) => {
3149 assert!(tracked);
3150 assert_eq!(fill.client_order_id, client_order_id);
3151 (
3152 fill.trade_id,
3153 fill.last_qty,
3154 fill.last_px,
3155 fill.commission.unwrap(),
3156 )
3157 }
3158 NautilusWsMessage::FillReports(mut reports) => {
3159 assert!(!tracked);
3160 assert_eq!(reports.len(), 1);
3161 let report = reports.pop().unwrap();
3162 (
3163 report.trade_id,
3164 report.last_qty,
3165 report.last_px,
3166 report.commission,
3167 )
3168 }
3169 message => panic!("unexpected fill message: {message:?}"),
3170 });
3171
3172 assert!(matches!(reconnected, Some(NautilusWsMessage::Reconnected)));
3173 assert!(handler.pending_outgoing.is_empty());
3174 assert_eq!(
3175 fills,
3176 [
3177 (
3178 TradeId::from("ETH-2696068"),
3179 Quantity::from(1),
3180 Price::from("203.8"),
3181 Money::from("0.00073602 USDT")
3182 ),
3183 (
3184 TradeId::from("ETH-2696069"),
3185 Quantity::from(1),
3186 Price::from("204.1"),
3187 Money::from("0.00083602 USDT")
3188 ),
3189 ],
3190 );
3191 assert_eq!(
3192 handler
3193 .terminal_order_contexts
3194 .contains_key(&venue_order_id),
3195 tracked
3196 );
3197 assert!(!handler.order_contexts.contains_key(&venue_order_id));
3198 }
3199
3200 #[rstest]
3201 fn trade_dedup_is_scoped_to_venue_order() {
3202 let mut handler = routing_test_handler();
3203 let first_trade = trade_data();
3204 let mut second_trade = first_trade.clone();
3205 second_trade["order_id"] = serde_json::json!("ETH-584830575");
3206 let trades: Vec<DeribitUserTradeMsg> = serde_json::from_str(
3207 &serde_json::json!([first_trade.clone(), first_trade, second_trade]).to_string(),
3208 )
3209 .unwrap();
3210
3211 let messages = handler.route_user_trades(&trades, UnixNanos::default());
3212
3213 assert_eq!(messages.len(), 1);
3214
3215 let NautilusWsMessage::FillReports(reports) = &messages[0] else {
3216 panic!("expected external fill reports");
3217 };
3218
3219 assert_eq!(
3220 reports
3221 .iter()
3222 .map(|report| (report.venue_order_id, report.trade_id))
3223 .collect::<Vec<_>>(),
3224 vec![
3225 (
3226 VenueOrderId::from("ETH-584830574"),
3227 TradeId::from("ETH-2696068")
3228 ),
3229 (
3230 VenueOrderId::from("ETH-584830575"),
3231 TradeId::from("ETH-2696068")
3232 ),
3233 ],
3234 );
3235 }
3236
3237 #[rstest]
3238 #[tokio::test]
3239 async fn reconnect_preserves_submit_identity_before_response() {
3240 let mut handler = routing_test_handler();
3241 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3242 handler.submitted_order_contexts.insert(
3243 client_order_id,
3244 OrderContext {
3245 client_order_id,
3246 trader_id: TraderId::from("TRADER-001"),
3247 strategy_id: StrategyId::from("S-001"),
3248 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3249 order_side: OrderSide::Buy,
3250 order_type: OrderType::Market,
3251 accepted: false,
3252 last_order_signature: None,
3253 },
3254 );
3255 handler
3256 .pending_requests
3257 .insert(42, PendingRequestType::Test);
3258
3259 handler.clear_state();
3260 handler
3261 .process_raw_message(&subscription(
3262 "user.orders.any.any.raw",
3263 &serde_json::json!([order_data("open", false)]),
3264 ))
3265 .await;
3266
3267 assert!(handler.pending_requests.is_empty());
3268 assert!(matches!(
3269 handler.pending_outgoing.pop_front().unwrap(),
3270 NautilusWsMessage::OrderAccepted(event)
3271 if event.client_order_id == client_order_id
3272 ));
3273 assert!(
3274 handler
3275 .order_contexts
3276 .get(&VenueOrderId::from("ETH-584830574"))
3277 .unwrap()
3278 .accepted
3279 );
3280 }
3281
3282 #[rstest]
3283 #[case("cancelled")]
3284 #[case("expired")]
3285 #[tokio::test]
3286 async fn tracked_terminal_order_synthesizes_accepted_first(#[case] order_state: &str) {
3287 let mut handler = routing_test_handler();
3288 let venue_order_id = VenueOrderId::from("ETH-584830574");
3289 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3290 handler.order_contexts.insert(
3291 venue_order_id,
3292 OrderContext {
3293 client_order_id,
3294 trader_id: TraderId::from("TRADER-001"),
3295 strategy_id: StrategyId::from("S-001"),
3296 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3297 order_side: OrderSide::Buy,
3298 order_type: OrderType::Market,
3299 accepted: false,
3300 last_order_signature: None,
3301 },
3302 );
3303
3304 handler
3305 .process_raw_message(&subscription(
3306 "user.orders.any.any.raw",
3307 &serde_json::json!([order_data(order_state, false)]),
3308 ))
3309 .await;
3310
3311 let accepted = handler.pending_outgoing.pop_front().unwrap();
3312 let terminal = handler.pending_outgoing.pop_front().unwrap();
3313 assert!(matches!(
3314 accepted,
3315 NautilusWsMessage::OrderAccepted(event)
3316 if event.client_order_id == client_order_id
3317 ));
3318 assert!(
3319 matches!(
3320 (order_state, terminal),
3321 ("cancelled", NautilusWsMessage::OrderCanceled(_))
3322 | ("expired", NautilusWsMessage::OrderExpired(_))
3323 ),
3324 "unexpected terminal message for {order_state}",
3325 );
3326 assert!(handler.pending_outgoing.is_empty());
3327 assert!(!handler.order_contexts.contains_key(&venue_order_id));
3328 }
3329
3330 #[rstest]
3331 #[case("filled")]
3332 #[case("open")]
3333 #[tokio::test]
3334 async fn late_fill_after_cancel_stays_on_tracked_event_path(#[case] trade_state: &str) {
3335 let mut handler = routing_test_handler();
3336 let venue_order_id = VenueOrderId::from("ETH-584830574");
3337 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3338 handler.order_contexts.insert(
3339 venue_order_id,
3340 OrderContext {
3341 client_order_id,
3342 trader_id: TraderId::from("TRADER-001"),
3343 strategy_id: StrategyId::from("S-001"),
3344 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3345 order_side: OrderSide::Buy,
3346 order_type: OrderType::Market,
3347 accepted: true,
3348 last_order_signature: None,
3349 },
3350 );
3351
3352 handler
3353 .process_raw_message(&subscription(
3354 "user.orders.any.any.raw",
3355 &serde_json::json!([order_data("cancelled", false)]),
3356 ))
3357 .await;
3358 let mut trade = trade_data();
3359 trade["state"] = serde_json::Value::String(trade_state.to_string());
3360 handler
3361 .process_raw_message(&subscription(
3362 "user.trades.any.any.raw",
3363 &serde_json::json!([trade]),
3364 ))
3365 .await;
3366
3367 assert!(matches!(
3368 handler.pending_outgoing.pop_front().unwrap(),
3369 NautilusWsMessage::OrderCanceled(event)
3370 if event.client_order_id == client_order_id
3371 ));
3372 assert!(matches!(
3373 handler.pending_outgoing.pop_front().unwrap(),
3374 NautilusWsMessage::OrderFilled(event)
3375 if event.client_order_id == client_order_id
3376 ));
3377 assert!(handler.pending_outgoing.is_empty());
3378 assert!(!handler.order_contexts.contains_key(&venue_order_id));
3379 assert!(
3380 handler
3381 .terminal_order_contexts
3382 .contains_key(&venue_order_id)
3383 );
3384 }
3385
3386 #[rstest]
3387 #[tokio::test]
3388 async fn tracked_order_uses_stored_client_id_when_label_changes() {
3389 let mut handler = routing_test_handler();
3390 let venue_order_id = VenueOrderId::from("ETH-584830574");
3391 let client_order_id = ClientOrderId::from("ORIGINAL-CLIENT-ID");
3392 handler.order_contexts.insert(
3393 venue_order_id,
3394 OrderContext {
3395 client_order_id,
3396 trader_id: TraderId::from("TRADER-001"),
3397 strategy_id: StrategyId::from("S-001"),
3398 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3399 order_side: OrderSide::Buy,
3400 order_type: OrderType::Market,
3401 accepted: false,
3402 last_order_signature: None,
3403 },
3404 );
3405 let mut order = order_data("open", false);
3406 order["label"] = serde_json::Value::Null;
3407
3408 handler
3409 .process_raw_message(&subscription(
3410 "user.orders.any.any.raw",
3411 &serde_json::json!([order]),
3412 ))
3413 .await;
3414
3415 assert!(matches!(
3416 handler.pending_outgoing.pop_front().unwrap(),
3417 NautilusWsMessage::OrderAccepted(event)
3418 if event.client_order_id == client_order_id
3419 ));
3420 assert!(handler.pending_outgoing.is_empty());
3421 }
3422
3423 #[rstest]
3424 #[tokio::test]
3425 async fn subscription_accept_before_submit_response_is_not_duplicated() {
3426 let mut handler = routing_test_handler();
3427 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3428 handler.submitted_order_contexts.insert(
3429 client_order_id,
3430 OrderContext {
3431 client_order_id,
3432 trader_id: TraderId::from("TRADER-001"),
3433 strategy_id: StrategyId::from("S-001"),
3434 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3435 order_side: OrderSide::Buy,
3436 order_type: OrderType::Market,
3437 accepted: false,
3438 last_order_signature: None,
3439 },
3440 );
3441 handler.pending_requests.insert(
3442 42,
3443 PendingRequestType::Buy {
3444 client_order_id,
3445 trader_id: TraderId::from("TRADER-001"),
3446 strategy_id: StrategyId::from("S-001"),
3447 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3448 order_side: OrderSide::Buy,
3449 order_type: OrderType::Market,
3450 },
3451 );
3452
3453 handler
3454 .process_raw_message(&subscription(
3455 "user.orders.any.any.raw",
3456 &serde_json::json!([order_data("open", false)]),
3457 ))
3458 .await;
3459 let accepted = handler.pending_outgoing.pop_front().unwrap();
3460 let response = serde_json::json!({
3461 "jsonrpc": "2.0",
3462 "id": 42,
3463 "result": { "order": order_data("open", false), "trades": [] }
3464 });
3465 let duplicate = handler.process_raw_message(&response.to_string()).await;
3466 handler.clear_state();
3467 handler
3468 .process_raw_message(&subscription(
3469 "user.orders.any.any.raw",
3470 &serde_json::json!([order_data("open", false)]),
3471 ))
3472 .await;
3473
3474 assert!(matches!(accepted, NautilusWsMessage::OrderAccepted(_)));
3475 assert!(duplicate.is_none());
3476 assert!(handler.pending_outgoing.is_empty());
3477 assert!(
3478 handler
3479 .order_contexts
3480 .get(&VenueOrderId::from("ETH-584830574"))
3481 .unwrap()
3482 .accepted
3483 );
3484 }
3485
3486 #[rstest]
3487 #[tokio::test]
3488 async fn untracked_live_order_and_fill_use_report_paths() {
3489 let mut handler = routing_test_handler();
3490
3491 handler
3492 .process_raw_message(&subscription(
3493 "user.orders.any.any.raw",
3494 &serde_json::json!([order_data("open", false)]),
3495 ))
3496 .await;
3497 handler
3498 .process_raw_message(&subscription(
3499 "user.trades.any.any.raw",
3500 &serde_json::json!([trade_data()]),
3501 ))
3502 .await;
3503
3504 assert!(matches!(
3505 handler.pending_outgoing.pop_front().unwrap(),
3506 NautilusWsMessage::OrderStatusReports(reports) if reports.len() == 1
3507 ));
3508 assert!(matches!(
3509 handler.pending_outgoing.pop_front().unwrap(),
3510 NautilusWsMessage::FillReports(reports) if reports.len() == 1
3511 ));
3512 assert!(handler.pending_outgoing.is_empty());
3513 }
3514
3515 #[rstest]
3516 #[tokio::test]
3517 async fn edit_response_without_tracked_context_uses_report_path() {
3518 let mut handler = routing_test_handler();
3519 handler.pending_requests.insert(
3520 42,
3521 PendingRequestType::Edit {
3522 client_order_id: ClientOrderId::from("UNKNOWN-CLIENT-ID"),
3523 trader_id: TraderId::from("TRADER-001"),
3524 strategy_id: StrategyId::from("S-001"),
3525 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3526 },
3527 );
3528 let response = serde_json::json!({
3529 "jsonrpc": "2.0",
3530 "id": 42,
3531 "result": { "order": order_data("open", true), "trades": [trade_data()] }
3532 });
3533
3534 let message = handler.process_raw_message(&response.to_string()).await;
3535 let fill = handler.pending_outgoing.pop_front().unwrap();
3536
3537 assert!(matches!(
3538 message,
3539 Some(NautilusWsMessage::OrderStatusReports(reports)) if reports.len() == 1
3540 ));
3541 assert!(matches!(
3542 fill,
3543 NautilusWsMessage::FillReports(reports) if reports.len() == 1
3544 ));
3545 assert!(handler.order_contexts.is_empty());
3546 assert!(handler.pending_outgoing.is_empty());
3547 }
3548
3549 #[rstest]
3550 #[tokio::test]
3551 async fn tracked_edit_response_routes_fill_and_deduplicates_subscription_echo() {
3552 let mut handler = routing_test_handler();
3553 let venue_order_id = VenueOrderId::from("ETH-584830574");
3554 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3555 handler.order_contexts.insert(
3556 venue_order_id,
3557 OrderContext {
3558 client_order_id,
3559 trader_id: TraderId::from("TRADER-001"),
3560 strategy_id: StrategyId::from("S-001"),
3561 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3562 order_side: OrderSide::Buy,
3563 order_type: OrderType::Market,
3564 accepted: true,
3565 last_order_signature: None,
3566 },
3567 );
3568 handler.pending_requests.insert(
3569 42,
3570 PendingRequestType::Edit {
3571 client_order_id,
3572 trader_id: TraderId::from("TRADER-001"),
3573 strategy_id: StrategyId::from("S-001"),
3574 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3575 },
3576 );
3577 let response = serde_json::json!({
3578 "jsonrpc": "2.0",
3579 "id": 42,
3580 "result": { "order": order_data("open", true), "trades": [trade_data()] }
3581 });
3582
3583 let updated = handler.process_raw_message(&response.to_string()).await;
3584 let filled = handler.pending_outgoing.pop_front().unwrap();
3585 handler
3586 .process_raw_message(&subscription(
3587 "user.orders.any.any.raw",
3588 &serde_json::json!([order_data("open", true)]),
3589 ))
3590 .await;
3591
3592 assert!(matches!(
3593 updated,
3594 Some(NautilusWsMessage::OrderUpdated(event))
3595 if event.client_order_id == client_order_id
3596 ));
3597 assert!(matches!(
3598 filled,
3599 NautilusWsMessage::OrderFilled(event)
3600 if event.client_order_id == client_order_id
3601 ));
3602 assert!(handler.pending_outgoing.is_empty());
3603 assert!(!handler.order_contexts.contains_key(&venue_order_id));
3604 assert!(
3605 handler
3606 .terminal_order_contexts
3607 .get(&venue_order_id)
3608 .unwrap()
3609 .last_order_signature
3610 .is_some(),
3611 );
3612 }
3613
3614 #[rstest]
3615 #[tokio::test]
3616 async fn subscription_edit_echo_before_response_is_not_duplicated() {
3617 let mut handler = routing_test_handler();
3618 let venue_order_id = VenueOrderId::from("ETH-584830574");
3619 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3620 handler.order_contexts.insert(
3621 venue_order_id,
3622 OrderContext {
3623 client_order_id,
3624 trader_id: TraderId::from("TRADER-001"),
3625 strategy_id: StrategyId::from("S-001"),
3626 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3627 order_side: OrderSide::Buy,
3628 order_type: OrderType::Market,
3629 accepted: true,
3630 last_order_signature: None,
3631 },
3632 );
3633 handler.pending_requests.insert(
3634 42,
3635 PendingRequestType::Edit {
3636 client_order_id,
3637 trader_id: TraderId::from("TRADER-001"),
3638 strategy_id: StrategyId::from("S-001"),
3639 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3640 },
3641 );
3642 handler
3643 .process_raw_message(&subscription(
3644 "user.orders.any.any.raw",
3645 &serde_json::json!([order_data("open", true)]),
3646 ))
3647 .await;
3648 let updated = handler.pending_outgoing.pop_front().unwrap();
3649 let response = serde_json::json!({
3650 "jsonrpc": "2.0",
3651 "id": 42,
3652 "result": { "order": order_data("open", true), "trades": [trade_data()] }
3653 });
3654
3655 let duplicate = handler.process_raw_message(&response.to_string()).await;
3656 let filled = handler.pending_outgoing.pop_front().unwrap();
3657
3658 assert!(matches!(
3659 updated,
3660 NautilusWsMessage::OrderUpdated(event)
3661 if event.client_order_id == client_order_id
3662 ));
3663 assert!(duplicate.is_none());
3664 assert!(matches!(
3665 filled,
3666 NautilusWsMessage::OrderFilled(event)
3667 if event.client_order_id == client_order_id
3668 ));
3669 assert!(handler.pending_outgoing.is_empty());
3670 }
3671
3672 #[rstest]
3673 #[case::new_trade(true)]
3674 #[case::repeated_trade(false)]
3675 #[tokio::test]
3676 async fn delayed_edit_response_keeps_partial_fill_terminal(#[case] new_trade: bool) {
3677 let mut handler = routing_test_handler();
3678 let venue_order_id = VenueOrderId::from("ETH-584830574");
3679 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3680 handler.order_contexts.insert(
3681 venue_order_id,
3682 OrderContext {
3683 client_order_id,
3684 trader_id: TraderId::from("TRADER-001"),
3685 strategy_id: StrategyId::from("S-001"),
3686 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3687 order_side: OrderSide::Buy,
3688 order_type: OrderType::Market,
3689 accepted: true,
3690 last_order_signature: None,
3691 },
3692 );
3693 handler
3694 .process_raw_message(&subscription(
3695 "user.trades.any.any.raw",
3696 &serde_json::json!([trade_data()]),
3697 ))
3698 .await;
3699 handler.pending_outgoing.clear();
3700 handler.pending_requests.insert(
3701 42,
3702 PendingRequestType::Edit {
3703 client_order_id,
3704 trader_id: TraderId::from("TRADER-001"),
3705 strategy_id: StrategyId::from("S-001"),
3706 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3707 },
3708 );
3709 let mut partial_trade = trade_data();
3710 partial_trade["state"] = serde_json::Value::String("open".to_string());
3711 if new_trade {
3712 partial_trade["trade_id"] = serde_json::json!("ETH-2696069");
3713 }
3714
3715 let response = serde_json::json!({
3716 "jsonrpc": "2.0",
3717 "id": 42,
3718 "result": { "order": order_data("open", true), "trades": [partial_trade] }
3719 });
3720
3721 let updated = handler.process_raw_message(&response.to_string()).await;
3722 let filled = handler.pending_outgoing.pop_front();
3723
3724 assert!(updated.is_none());
3725
3726 match filled {
3727 Some(NautilusWsMessage::OrderFilled(event)) => {
3728 assert!(new_trade);
3729 assert_eq!(event.client_order_id, client_order_id);
3730 assert_eq!(event.trade_id, TradeId::from("ETH-2696069"));
3731 }
3732 None => assert!(!new_trade),
3733 message => panic!("unexpected fill message: {message:?}"),
3734 }
3735
3736 assert!(!handler.order_contexts.contains_key(&venue_order_id));
3737 assert!(
3738 handler
3739 .terminal_order_contexts
3740 .contains_key(&venue_order_id)
3741 );
3742 assert!(handler.pending_outgoing.is_empty());
3743 }
3744
3745 #[rstest]
3746 #[tokio::test]
3747 async fn tracked_replaced_order_emits_updated_event() {
3748 let mut handler = routing_test_handler();
3749 let venue_order_id = VenueOrderId::from("ETH-584830574");
3750 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3751 handler.order_contexts.insert(
3752 venue_order_id,
3753 OrderContext {
3754 client_order_id,
3755 trader_id: TraderId::from("TRADER-001"),
3756 strategy_id: StrategyId::from("S-001"),
3757 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3758 order_side: OrderSide::Buy,
3759 order_type: OrderType::Market,
3760 accepted: true,
3761 last_order_signature: None,
3762 },
3763 );
3764
3765 handler
3766 .process_raw_message(&subscription(
3767 "user.orders.any.any.raw",
3768 &serde_json::json!([order_data("open", true)]),
3769 ))
3770 .await;
3771
3772 assert!(matches!(
3773 handler.pending_outgoing.pop_front().unwrap(),
3774 NautilusWsMessage::OrderUpdated(event)
3775 if event.client_order_id == client_order_id
3776 ));
3777
3778 let mut partial_fill = order_data("open", true);
3779 partial_fill["filled_amount"] = serde_json::json!(1.0);
3780 partial_fill["last_update_timestamp"] = serde_json::json!(1_590_480_712_900_u64);
3781 handler
3782 .process_raw_message(&subscription(
3783 "user.orders.any.any.raw",
3784 &serde_json::json!([partial_fill]),
3785 ))
3786 .await;
3787
3788 assert!(handler.pending_outgoing.is_empty());
3789 }
3790
3791 const BOOK_CHANNEL: &str = "book.BTC-PERPETUAL.100ms";
3792 const GROUPED_CHANNEL: &str = "book.BTC-PERPETUAL.none.10.100ms";
3793
3794 const UNCONVERTIBLE_PRICE: f64 = 1e20;
3796
3797 fn book_message(
3798 msg_type: &str,
3799 change_id: u64,
3800 prev_change_id: Option<u64>,
3801 bid_price: f64,
3802 ) -> String {
3803 subscription(
3804 BOOK_CHANNEL,
3805 &serde_json::json!({
3806 "type": msg_type,
3807 "instrument_name": "BTC-PERPETUAL",
3808 "timestamp": 1_700_000_000_000_u64,
3809 "change_id": change_id,
3810 "prev_change_id": prev_change_id,
3811 "bids": [["new", bid_price, 10.0]],
3812 "asks": [],
3813 }),
3814 )
3815 }
3816
3817 fn grouped_book_message(change_id: u64, bid_price: f64) -> String {
3819 subscription(
3820 GROUPED_CHANNEL,
3821 &serde_json::json!({
3822 "instrument_name": "BTC-PERPETUAL",
3823 "timestamp": 1_700_000_000_000_u64,
3824 "change_id": change_id,
3825 "bids": [[bid_price, 10.0]],
3826 "asks": [],
3827 }),
3828 )
3829 }
3830
3831 fn book_test_handler() -> DeribitWsFeedHandler {
3832 let handler = routing_test_handler();
3833
3834 for channel in [
3835 BOOK_CHANNEL,
3836 "book.ETH-PERPETUAL.100ms",
3837 "trades.BTC-PERPETUAL.100ms",
3838 ] {
3839 handler.subscriptions_state.mark_subscribe(channel);
3840 handler.subscriptions_state.confirm_subscribe(channel);
3841 }
3842
3843 handler
3844 }
3845
3846 fn pending_unsubscribe_channels(handler: &DeribitWsFeedHandler) -> Vec<Vec<String>> {
3847 handler
3848 .pending_requests
3849 .values()
3850 .filter_map(|request| match request {
3851 PendingRequestType::Unsubscribe { channels } => Some(channels.clone()),
3852 _ => None,
3853 })
3854 .collect()
3855 }
3856
3857 #[rstest]
3858 #[tokio::test]
3859 async fn book_sequence_gap_forces_resync_after_contiguous_delta() {
3860 let mut handler = book_test_handler();
3861
3862 let snapshot = handler
3863 .process_raw_message(&book_message("snapshot", 100, None, 42500.0))
3864 .await;
3865 let contiguous = handler
3866 .process_raw_message(&book_message("change", 101, Some(100), 42499.5))
3867 .await;
3868
3869 assert!(matches!(
3870 snapshot,
3871 Some(NautilusWsMessage::Deltas(deltas)) if deltas.sequence == 100
3872 ));
3873 assert!(matches!(
3874 contiguous,
3875 Some(NautilusWsMessage::Deltas(deltas)) if deltas.sequence == 101
3876 ));
3877 assert_eq!(
3878 handler.book_sequence.get(&Ustr::from("BTC-PERPETUAL")),
3879 Some(&101)
3880 );
3881 assert!(handler.pending_book_resync.is_empty());
3882
3883 let gap = handler
3884 .process_raw_message(&book_message("change", 103, Some(102), 42499.0))
3885 .await;
3886
3887 assert!(gap.is_none());
3888 assert!(handler.book_sequence.is_empty());
3889 assert_eq!(handler.pending_book_resync, vec![BOOK_CHANNEL.to_string()]);
3890 assert_eq!(
3891 pending_unsubscribe_channels(&handler),
3892 vec![vec![BOOK_CHANNEL.to_string()]]
3893 );
3894 }
3895
3896 #[rstest]
3897 #[tokio::test]
3898 async fn rejected_grouped_book_snapshot_waits_for_next_snapshot() {
3899 let mut handler = book_test_handler();
3900 handler.subscriptions_state.mark_subscribe(GROUPED_CHANNEL);
3901 handler
3902 .subscriptions_state
3903 .confirm_subscribe(GROUPED_CHANNEL);
3904
3905 let rejected = handler
3906 .process_raw_message(&grouped_book_message(100, UNCONVERTIBLE_PRICE))
3907 .await;
3908
3909 assert!(rejected.is_none());
3910 assert!(handler.book_sequence.is_empty());
3911 assert!(handler.pending_book_resync.is_empty());
3912 assert!(pending_unsubscribe_channels(&handler).is_empty());
3913
3914 let next = handler
3915 .process_raw_message(&grouped_book_message(101, 42500.0))
3916 .await;
3917
3918 assert!(matches!(
3919 next,
3920 Some(NautilusWsMessage::Deltas(deltas)) if deltas.sequence == 101
3921 ));
3922 }
3923
3924 #[rstest]
3925 #[tokio::test]
3926 async fn delta_channel_resync_survives_grouped_snapshot_before_unsubscribe_ack() {
3927 let mut handler = book_test_handler();
3928 handler.subscriptions_state.mark_subscribe(GROUPED_CHANNEL);
3929 handler
3930 .subscriptions_state
3931 .confirm_subscribe(GROUPED_CHANNEL);
3932
3933 handler
3934 .process_raw_message(&book_message("snapshot", 100, None, 42500.0))
3935 .await;
3936 handler
3937 .process_raw_message(&book_message("change", 101, Some(100), UNCONVERTIBLE_PRICE))
3938 .await;
3939 let grouped = handler
3940 .process_raw_message(&grouped_book_message(102, 42500.0))
3941 .await;
3942
3943 assert!(matches!(
3944 grouped,
3945 Some(NautilusWsMessage::Deltas(deltas)) if deltas.sequence == 102
3946 ));
3947 assert_eq!(handler.pending_book_resync, vec![BOOK_CHANNEL.to_string()]);
3948 assert_eq!(
3949 pending_unsubscribe_channels(&handler),
3950 vec![vec![BOOK_CHANNEL.to_string()]]
3951 );
3952
3953 let unsubscribe_id = handler
3954 .pending_requests
3955 .iter()
3956 .find_map(|(id, request)| {
3957 matches!(request, PendingRequestType::Unsubscribe { .. }).then_some(*id)
3958 })
3959 .unwrap();
3960
3961 let ack = serde_json::json!({
3962 "jsonrpc": "2.0",
3963 "id": unsubscribe_id,
3964 "result": [BOOK_CHANNEL],
3965 });
3966 handler.process_raw_message(&ack.to_string()).await;
3967
3968 let resubscribed: Vec<Vec<String>> = handler
3969 .pending_requests
3970 .values()
3971 .filter_map(|request| match request {
3972 PendingRequestType::Subscribe { channels } => Some(channels.clone()),
3973 _ => None,
3974 })
3975 .collect();
3976
3977 assert_eq!(resubscribed, vec![vec![BOOK_CHANNEL.to_string()]]);
3978 }
3979
3980 #[rstest]
3981 #[tokio::test]
3982 async fn rejected_book_delta_forces_resync_and_drops_successor() {
3983 let mut handler = book_test_handler();
3984
3985 let snapshot = handler
3986 .process_raw_message(&book_message("snapshot", 100, None, 42500.0))
3987 .await;
3988 let rejected = handler
3989 .process_raw_message(&book_message("change", 101, Some(100), UNCONVERTIBLE_PRICE))
3990 .await;
3991 let successor = handler
3992 .process_raw_message(&book_message("change", 102, Some(101), 42499.5))
3993 .await;
3994
3995 assert!(matches!(
3996 snapshot,
3997 Some(NautilusWsMessage::Deltas(deltas)) if deltas.sequence == 100
3998 ));
3999 assert!(rejected.is_none());
4000 assert!(successor.is_none());
4001 assert!(handler.book_sequence.is_empty());
4002 assert_eq!(handler.pending_book_resync, vec![BOOK_CHANNEL.to_string()]);
4003 assert_eq!(
4004 pending_unsubscribe_channels(&handler),
4005 vec![vec![BOOK_CHANNEL.to_string()]]
4006 );
4007 assert_eq!(
4008 handler.subscriptions_state.pending_subscribe_topics(),
4009 vec![BOOK_CHANNEL.to_string()]
4010 );
4011
4012 let recovered = handler
4013 .process_raw_message(&book_message("snapshot", 200, None, 42500.0))
4014 .await;
4015
4016 assert!(matches!(
4017 recovered,
4018 Some(NautilusWsMessage::Deltas(deltas)) if deltas.sequence == 200
4019 ));
4020 assert!(handler.pending_book_resync.is_empty());
4021 assert_eq!(
4022 handler.book_sequence.get(&Ustr::from("BTC-PERPETUAL")),
4023 Some(&200)
4024 );
4025 }
4026
4027 #[rstest]
4028 #[tokio::test]
4029 async fn rejected_resync_book_snapshot_keeps_resync_pending() {
4030 let mut handler = book_test_handler();
4031 handler.pending_book_resync.push(BOOK_CHANNEL.to_string());
4032
4033 let rejected = handler
4034 .process_raw_message(&book_message("snapshot", 200, None, UNCONVERTIBLE_PRICE))
4035 .await;
4036 let delta = handler
4037 .process_raw_message(&book_message("change", 201, Some(200), 42499.5))
4038 .await;
4039
4040 assert!(rejected.is_none());
4041 assert!(delta.is_none());
4042 assert!(handler.book_sequence.is_empty());
4043 assert_eq!(handler.pending_book_resync, vec![BOOK_CHANNEL.to_string()]);
4044 assert_eq!(
4045 pending_unsubscribe_channels(&handler),
4046 vec![vec![BOOK_CHANNEL.to_string()]]
4047 );
4048 }
4049
4050 #[rstest]
4051 #[case(OrderSide::Buy)]
4052 #[case(OrderSide::Sell)]
4053 #[tokio::test]
4054 async fn order_serialization_failure_leaves_no_order_context(#[case] side: OrderSide) {
4055 let mut handler = routing_test_handler();
4056 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
4057 let trader_id = TraderId::from("TRADER-001");
4058 let strategy_id = StrategyId::from("S-001");
4059 let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
4060
4061 let params = DeribitOrderParams {
4062 instrument_name: "BTC-PERPETUAL".to_string(),
4063 amount: Decimal::from_str_exact("100000000.005000001").unwrap(),
4064 order_type: "limit".to_string(),
4065 label: Some(client_order_id.to_string()),
4066 price: Some(Decimal::from_str_exact("65000.5").unwrap()),
4067 time_in_force: Some("good_til_cancelled".to_string()),
4068 post_only: None,
4069 reject_post_only: None,
4070 reduce_only: None,
4071 trigger_price: None,
4072 trigger: None,
4073 max_show: None,
4074 valid_until: None,
4075 };
4076
4077 let result = match side {
4078 OrderSide::Buy => {
4079 handler
4080 .handle_buy(
4081 params,
4082 client_order_id,
4083 trader_id,
4084 strategy_id,
4085 instrument_id,
4086 )
4087 .await
4088 }
4089 OrderSide::Sell => {
4090 handler
4091 .handle_sell(
4092 params,
4093 client_order_id,
4094 trader_id,
4095 strategy_id,
4096 instrument_id,
4097 )
4098 .await
4099 }
4100 };
4101
4102 if "0.10".parse::<serde_json::Number>().unwrap().to_string() == "0.10" {
4104 assert!(matches!(result, Err(DeribitWsError::NotConnected)));
4105 assert!(
4106 handler
4107 .submitted_order_contexts
4108 .get(&client_order_id)
4109 .is_some()
4110 );
4111 assert_eq!(handler.pending_requests.len(), 1);
4112 } else {
4113 assert!(matches!(
4114 result,
4115 Err(DeribitWsError::Json(message))
4116 if message == "exact decimal number serialization requires serde_json/arbitrary_precision"
4117 ));
4118 assert!(
4119 handler
4120 .submitted_order_contexts
4121 .get(&client_order_id)
4122 .is_none()
4123 );
4124 assert!(handler.pending_requests.is_empty());
4125 }
4126 }
4127}