1use std::{
23 collections::VecDeque,
24 sync::{
25 Arc,
26 atomic::{AtomicBool, AtomicU64, Ordering},
27 },
28};
29
30use ahash::AHashMap;
31use futures_util::future::BoxFuture;
32use nautilus_common::cache::fifo::{FifoCache, FifoCacheMap};
33use nautilus_core::{
34 AtomicSet, AtomicTime, UUID4, UnixNanos,
35 string::secret::{SecretString, zeroize_json_value},
36 time::get_atomic_clock_realtime,
37};
38use nautilus_model::{
39 data::{Bar, CustomData, Data, DataType, InstrumentStatus},
40 enums::{MarketStatusAction, OrderSide, OrderType},
41 events::{
42 AccountState, OrderAccepted, OrderCancelRejected, OrderFilled, OrderModifyRejected,
43 OrderRejected,
44 },
45 identifiers::{
46 AccountId, ClientOrderId, InstrumentId, StrategyId, Symbol, TradeId, TraderId, VenueOrderId,
47 },
48 instruments::{Instrument, InstrumentAny},
49};
50use nautilus_network::{
51 RECONNECTED,
52 retry::{RetryManager, create_websocket_retry_manager},
53 websocket::{AuthTracker, SubscriptionState, WebSocketClient},
54};
55use parking_lot::Mutex;
56use rust_decimal::Decimal;
57use tokio_tungstenite::tungstenite::Message;
58use ustr::Ustr;
59
60use super::{
61 enums::{DeribitBookMsgType, DeribitHeartbeatType, DeribitWsChannel, DeribitWsMethod},
62 error::DeribitWsError,
63 messages::{
64 DeribitAuthResult, DeribitBookMsg, DeribitCancelAllByInstrumentParams, DeribitCancelParams,
65 DeribitChartMsg, DeribitEditParams, DeribitHeartbeatParams, DeribitInstrumentStateMsg,
66 DeribitJsonRpcRequest, DeribitOrderMsg, DeribitOrderParams, DeribitOrderResponse,
67 DeribitPerpetualMsg, DeribitPortfolioMsg, DeribitQuoteMsg, DeribitSubscribeParams,
68 DeribitTickerMsg, DeribitTradeMsg, DeribitUserTradeMsg, DeribitVolatilityIndexMsg,
69 DeribitWsMessage, NautilusWsMessage, parse_raw_message,
70 },
71 parse::{
72 OrderEventType, determine_order_event_type, parse_book_depth, parse_book_msg,
73 parse_chart_msg, parse_deribit_order_type, parse_order_accepted_with_client_order_id,
74 parse_order_canceled_with_client_order_id, parse_order_expired_with_client_order_id,
75 parse_order_updated_with_client_order_id, parse_perpetual_to_funding_rate, parse_quote_msg,
76 parse_ticker_to_index_price, parse_ticker_to_mark_price, parse_ticker_to_option_greeks,
77 parse_trades_data, parse_user_order_msg, parse_user_trade_msg, resolution_to_bar_type,
78 },
79};
80use crate::{
81 common::{
82 consts::{
83 DERIBIT_POST_ONLY_ERROR_CODE, DERIBIT_RATE_LIMIT_KEY_ORDER, DERIBIT_VENUE,
84 DERIBIT_WS_SUBSCRIPTION_KEY,
85 },
86 enums::DeribitInstrumentState,
87 parse::{parse_portfolio_to_account_state, use_cost_for_bar_volume},
88 },
89 data_types::DeribitVolatilityIndex,
90};
91
92const SUBSCRIPTION_REQUEST_MAX_BYTES: usize = 32_000;
94
95#[derive(Debug, Clone)]
97pub enum PendingRequestType {
98 Authenticate,
100 Subscribe { channels: Vec<String> },
102 Unsubscribe { channels: Vec<String> },
104 SetHeartbeat,
106 Test,
108 Buy {
110 client_order_id: ClientOrderId,
111 trader_id: TraderId,
112 strategy_id: StrategyId,
113 instrument_id: InstrumentId,
114 order_side: OrderSide,
115 order_type: OrderType,
116 },
117 Sell {
119 client_order_id: ClientOrderId,
120 trader_id: TraderId,
121 strategy_id: StrategyId,
122 instrument_id: InstrumentId,
123 order_side: OrderSide,
124 order_type: OrderType,
125 },
126 Edit {
128 client_order_id: ClientOrderId,
129 trader_id: TraderId,
130 strategy_id: StrategyId,
131 instrument_id: InstrumentId,
132 },
133 Cancel {
135 client_order_id: ClientOrderId,
136 trader_id: TraderId,
137 strategy_id: StrategyId,
138 instrument_id: InstrumentId,
139 },
140 CancelAllByInstrument { instrument_id: InstrumentId },
142 GetOrderState {
144 client_order_id: ClientOrderId,
145 trader_id: TraderId,
146 strategy_id: StrategyId,
147 instrument_id: InstrumentId,
148 },
149}
150
151#[allow(missing_debug_implementations)]
153pub enum HandlerCommand {
154 SetClient(WebSocketClient),
156 Disconnect,
158 Authenticate {
160 auth_params: SecretString,
162 },
163 SetHeartbeat { interval: u64 },
165 InitializeInstruments(Vec<InstrumentAny>),
167 UpdateInstrument(Box<InstrumentAny>),
169 Subscribe { channels: Vec<String> },
171 Unsubscribe { channels: Vec<String> },
173 Buy {
175 params: DeribitOrderParams,
176 client_order_id: ClientOrderId,
177 trader_id: TraderId,
178 strategy_id: StrategyId,
179 instrument_id: InstrumentId,
180 },
181 Sell {
183 params: DeribitOrderParams,
184 client_order_id: ClientOrderId,
185 trader_id: TraderId,
186 strategy_id: StrategyId,
187 instrument_id: InstrumentId,
188 },
189 Edit {
191 params: DeribitEditParams,
192 client_order_id: ClientOrderId,
193 trader_id: TraderId,
194 strategy_id: StrategyId,
195 instrument_id: InstrumentId,
196 },
197 Cancel {
199 params: DeribitCancelParams,
200 client_order_id: ClientOrderId,
201 trader_id: TraderId,
202 strategy_id: StrategyId,
203 instrument_id: InstrumentId,
204 },
205 CancelAllByInstrument {
207 params: DeribitCancelAllByInstrumentParams,
208 instrument_id: InstrumentId,
209 },
210 GetOrderState {
212 order_id: String,
213 client_order_id: ClientOrderId,
214 trader_id: TraderId,
215 strategy_id: StrategyId,
216 instrument_id: InstrumentId,
217 },
218}
219
220#[derive(Debug, Clone)]
224pub struct OrderContext {
225 pub client_order_id: ClientOrderId,
226 pub trader_id: TraderId,
227 pub strategy_id: StrategyId,
228 pub instrument_id: InstrumentId,
229 pub order_side: OrderSide,
230 pub order_type: OrderType,
231 pub accepted: bool,
232 pub last_order_signature: Option<OrderSignature>,
233}
234
235pub type OrderSignature = (Decimal, Option<Decimal>, Option<Decimal>);
237
238#[allow(missing_debug_implementations)]
242pub struct DeribitWsFeedHandler {
243 clock: &'static AtomicTime,
244 signal: Arc<AtomicBool>,
245 inner: Option<Arc<WebSocketClient>>,
246 cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
247 pending_command: Option<HandlerCommand>,
248 subscription_requests: VecDeque<(u64, String)>,
249 subscription_send: Option<SubscriptionSend>,
250 raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
251 out_tx: tokio::sync::mpsc::UnboundedSender<NautilusWsMessage>,
252 auth_tracker: AuthTracker,
253 subscriptions_state: SubscriptionState,
254 retry_manager: RetryManager<DeribitWsError>,
255 instruments_cache: AHashMap<Ustr, InstrumentAny>,
256 option_greeks_subs: Arc<AtomicSet<InstrumentId>>,
257 mark_price_subs: Arc<AtomicSet<InstrumentId>>,
258 index_price_subs: Arc<AtomicSet<InstrumentId>>,
259 request_id_counter: AtomicU64,
260 pending_requests: AHashMap<u64, PendingRequestType>,
261 account_id: Option<AccountId>,
262 order_contexts: AHashMap<VenueOrderId, OrderContext>,
263 submitted_order_contexts: FifoCacheMap<ClientOrderId, OrderContext, 10_000>,
264
265 terminal_order_contexts: FifoCacheMap<VenueOrderId, OrderContext, 10_000>,
267
268 emitted_trades: FifoCache<(VenueOrderId, TradeId), 10_000>,
269 pending_bars: AHashMap<String, Bar>,
270 bars_timestamp_on_close: bool,
271 last_account_states: AHashMap<String, AccountState>,
272 book_depth_subs: Arc<tokio::sync::Mutex<SubscriptionState>>,
273 book_sequence: AHashMap<Ustr, u64>,
274 pending_book_resync: Vec<String>,
275 pending_outgoing: VecDeque<NautilusWsMessage>,
276 subscribe_errors: Arc<Mutex<Vec<String>>>,
277}
278
279impl DeribitWsFeedHandler {
280 #[expect(clippy::too_many_arguments)]
282 #[must_use]
283 pub fn new(
284 signal: Arc<AtomicBool>,
285 cmd_rx: tokio::sync::mpsc::UnboundedReceiver<HandlerCommand>,
286 raw_rx: tokio::sync::mpsc::UnboundedReceiver<Message>,
287 out_tx: tokio::sync::mpsc::UnboundedSender<NautilusWsMessage>,
288 auth_tracker: AuthTracker,
289 subscriptions_state: SubscriptionState,
290 option_greeks_subs: Arc<AtomicSet<InstrumentId>>,
291 mark_price_subs: Arc<AtomicSet<InstrumentId>>,
292 index_price_subs: Arc<AtomicSet<InstrumentId>>,
293 account_id: Option<AccountId>,
294 bars_timestamp_on_close: bool,
295 subscribe_errors: Arc<Mutex<Vec<String>>>,
296 ) -> Self {
297 Self {
298 clock: get_atomic_clock_realtime(),
299 signal,
300 inner: None,
301 cmd_rx,
302 pending_command: None,
303 subscription_requests: VecDeque::new(),
304 subscription_send: None,
305 raw_rx,
306 out_tx,
307 auth_tracker,
308 subscriptions_state,
309 retry_manager: create_websocket_retry_manager(),
310 instruments_cache: AHashMap::new(),
311 option_greeks_subs,
312 mark_price_subs,
313 index_price_subs,
314 request_id_counter: AtomicU64::new(1),
315 pending_requests: AHashMap::new(),
316 account_id,
317 order_contexts: AHashMap::new(),
318 submitted_order_contexts: FifoCacheMap::new(),
319 terminal_order_contexts: FifoCacheMap::new(),
320 emitted_trades: FifoCache::new(),
321 pending_bars: AHashMap::new(),
322 bars_timestamp_on_close,
323 last_account_states: AHashMap::new(),
324 book_depth_subs: Arc::new(tokio::sync::Mutex::new(SubscriptionState::new('.'))),
325 book_sequence: AHashMap::new(),
326 pending_book_resync: Vec::new(),
327 pending_outgoing: VecDeque::new(),
328 subscribe_errors,
329 }
330 }
331
332 pub fn set_account_id(&mut self, account_id: AccountId) {
334 self.account_id = Some(account_id);
335 }
336
337 #[must_use]
339 pub fn account_id(&self) -> Option<AccountId> {
340 self.account_id
341 }
342
343 fn clear_state(&mut self) {
344 let pending_count = self.pending_requests.len();
345 let bars_count = self.pending_bars.len();
346 let account_count = self.last_account_states.len();
347 let book_count = self.book_sequence.len();
348 let outgoing_count = self.pending_outgoing.len();
349
350 self.pending_requests.clear();
351 self.subscription_requests.clear();
352 self.subscription_send = None;
353 self.pending_bars.clear();
354 self.last_account_states.clear();
355 self.book_sequence.clear();
356 self.pending_book_resync.clear();
357 self.pending_outgoing.clear();
358
359 log::debug!(
360 "Reset state: pending_requests={pending_count}, pending_bars={bars_count}, \
361 account_states={account_count}, book_sequence={book_count}, \
362 pending_outgoing={outgoing_count}"
363 );
364 }
365
366 pub(crate) fn set_book_depth_subs(&mut self, subs: Arc<tokio::sync::Mutex<SubscriptionState>>) {
367 self.book_depth_subs = subs;
368 }
369
370 fn next_request_id(&self) -> u64 {
372 self.request_id_counter.fetch_add(1, Ordering::Relaxed)
373 }
374
375 fn ts_init(&self) -> UnixNanos {
377 self.clock.get_time_ns()
378 }
379
380 async fn send_tracked_request(
381 &mut self,
382 request_id: u64,
383 payload: Result<String, DeribitWsError>,
384 rate_limit_keys: Option<&[Ustr]>,
385 ) -> Result<(), DeribitWsError> {
386 let payload = match payload {
387 Ok(p) => p,
388 Err(e) => {
389 self.pending_requests.remove(&request_id);
390 return Err(e);
391 }
392 };
393 self.send_with_retry(payload, rate_limit_keys).await
394 }
395
396 fn send_with_retry(
398 &self,
399 payload: String,
400 rate_limit_keys: Option<&[Ustr]>,
401 ) -> BoxFuture<'static, Result<(), DeribitWsError>> {
402 self.send_secret_with_retry(payload.into(), rate_limit_keys)
403 }
404
405 fn send_secret_with_retry(
406 &self,
407 payload: SecretString,
408 rate_limit_keys: Option<&[Ustr]>,
409 ) -> BoxFuture<'static, Result<(), DeribitWsError>> {
410 let client = self.inner.clone();
411 let retry_manager = self.retry_manager.clone();
412 let keys_owned: Option<Vec<Ustr>> = rate_limit_keys.map(<[Ustr]>::to_vec);
413 Box::pin(async move {
414 let Some(client) = client else {
415 return Err(DeribitWsError::NotConnected);
416 };
417
418 retry_manager
419 .invocation(
420 "websocket_send",
421 || {
422 let payload = payload.clone();
423 let keys = keys_owned.clone();
424 let client = client.clone();
425 async move {
426 client
427 .send_text(payload.expose_secret().to_owned(), keys.as_deref())
428 .await
429 .map_err(|e| DeribitWsError::Send(e.to_string()))
430 }
431 },
432 |e| matches!(e, DeribitWsError::Send(_)),
433 |e| DeribitWsError::Timeout(e.to_string()),
434 )
435 .execute()
436 .await
437 })
438 }
439
440 async fn handle_authenticate(&mut self, auth_params: SecretString) {
441 let request_id = self.next_request_id();
442 log::debug!("Authenticating: request_id={request_id}");
443
444 let auth_params: serde_json::Value = match serde_json::from_str(auth_params.expose_secret())
445 {
446 Ok(auth_params) => auth_params,
447 Err(e) => {
448 log::error!("Failed to deserialize auth params: {e}");
449 self.auth_tracker.fail(format!("Serialization failed: {e}"));
450 return;
451 }
452 };
453 self.pending_requests
454 .insert(request_id, PendingRequestType::Authenticate);
455
456 let mut request = DeribitJsonRpcRequest::new(
457 request_id,
458 DeribitWsMethod::PublicAuth.as_method_str(),
459 auth_params,
460 );
461 let serialized = serde_json::to_string(&request).map(SecretString::from);
462 zeroize_json_value(&mut request.params);
463 drop(request);
464
465 match serialized {
466 Ok(payload) => {
467 if let Err(e) = self.send_secret_with_retry(payload, None).await {
468 self.pending_requests.remove(&request_id);
469 log::error!("Authentication send failed: {e}");
470 self.auth_tracker.fail(format!("Send failed: {e}"));
471 }
472 }
473 Err(e) => {
474 self.pending_requests.remove(&request_id);
475 log::error!("Failed to serialize auth request: {e}");
476 self.auth_tracker.fail(format!("Serialization failed: {e}"));
477 }
478 }
479 }
480
481 fn handle_subscribe(&mut self, channels: Vec<String>) -> Result<(), DeribitWsError> {
485 let result = self.queue_subscription_requests(channels, true);
486 if let Err(e) = &result {
487 self.subscribe_errors.lock().push(e.to_string());
488 }
489
490 result
491 }
492
493 fn handle_unsubscribe(&mut self, channels: Vec<String>) -> Result<(), DeribitWsError> {
495 self.queue_subscription_requests(channels, false)
496 }
497
498 fn queue_subscription_requests(
499 &mut self,
500 channels: Vec<String>,
501 subscribe: bool,
502 ) -> Result<(), DeribitWsError> {
503 let private = channels
504 .iter()
505 .any(|ch| DeribitWsChannel::requires_auth(ch));
506
507 let method = match (subscribe, private) {
508 (true, true) => DeribitWsMethod::PrivateSubscribe,
509 (true, false) => DeribitWsMethod::PublicSubscribe,
510 (false, true) => DeribitWsMethod::PrivateUnsubscribe,
511 (false, false) => DeribitWsMethod::PublicUnsubscribe,
512 };
513
514 let batches = subscription_batches(channels, &method)?;
515 for channels in batches {
516 let request_id = self.next_request_id();
517
518 let request = DeribitJsonRpcRequest::new(
519 request_id,
520 method.as_method_str(),
521 DeribitSubscribeParams { channels },
522 );
523 let payload =
524 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()))?;
525 let channels = request.params.channels;
526 let pending = if subscribe {
527 PendingRequestType::Subscribe { channels }
528 } else {
529 PendingRequestType::Unsubscribe { channels }
530 };
531
532 self.pending_requests.insert(request_id, pending);
533 self.subscription_requests.push_back((request_id, payload));
534 }
535
536 Ok(())
537 }
538
539 fn coalesce_subscriptions(&mut self, mut channels: Vec<String>) -> Vec<String> {
540 let private = channels
541 .iter()
542 .any(|ch| DeribitWsChannel::requires_auth(ch));
543 let mut bytes = channels.iter().map(|ch| ch.len() + 3).sum::<usize>();
544 while bytes < SUBSCRIPTION_REQUEST_MAX_BYTES {
545 let Ok(cmd) = self.cmd_rx.try_recv() else {
546 break;
547 };
548
549 match cmd {
550 HandlerCommand::Subscribe { channels: adjacent }
551 if adjacent
552 .iter()
553 .any(|ch| DeribitWsChannel::requires_auth(ch))
554 == private =>
555 {
556 bytes += adjacent.iter().map(|ch| ch.len() + 3).sum::<usize>() + 1;
557 channels.extend(adjacent);
558 }
559 cmd => {
560 self.pending_command = Some(cmd);
561 break;
562 }
563 }
564 }
565
566 channels
567 }
568
569 async fn handle_set_heartbeat(&mut self, interval: u64) -> Result<(), DeribitWsError> {
571 let request_id = self.next_request_id();
572
573 self.pending_requests
575 .insert(request_id, PendingRequestType::SetHeartbeat);
576
577 let request = DeribitJsonRpcRequest::new(
578 request_id,
579 DeribitWsMethod::SetHeartbeat.as_method_str(),
580 DeribitHeartbeatParams { interval },
581 );
582
583 let payload =
584 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
585
586 log::debug!(
587 "Enabling heartbeat with interval: request_id={request_id}, interval={interval} seconds"
588 );
589 self.send_tracked_request(request_id, payload, None).await
590 }
591
592 async fn handle_heartbeat_test_request(&mut self) -> Result<(), DeribitWsError> {
594 let request_id = self.next_request_id();
595
596 self.pending_requests
598 .insert(request_id, PendingRequestType::Test);
599
600 let request = DeribitJsonRpcRequest::new(
601 request_id,
602 DeribitWsMethod::Test.as_method_str(),
603 serde_json::json!({}),
604 );
605
606 let payload =
607 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
608
609 log::trace!("Responding to heartbeat test_request: request_id={request_id}");
610 self.send_tracked_request(request_id, payload, None).await
611 }
612
613 async fn handle_buy(
615 &mut self,
616 params: DeribitOrderParams,
617 client_order_id: ClientOrderId,
618 trader_id: TraderId,
619 strategy_id: StrategyId,
620 instrument_id: InstrumentId,
621 ) -> Result<(), DeribitWsError> {
622 let request_id = self.next_request_id();
623 let order_type = parse_deribit_order_type(¶ms.order_type);
624 let order_signature = (params.amount, params.price, params.trigger_price);
625
626 let request =
627 DeribitJsonRpcRequest::new(request_id, DeribitWsMethod::Buy.as_method_str(), params);
628
629 let payload =
631 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()))?;
632
633 self.submitted_order_contexts.insert(
634 client_order_id,
635 OrderContext {
636 client_order_id,
637 trader_id,
638 strategy_id,
639 instrument_id,
640 order_side: OrderSide::Buy,
641 order_type,
642 accepted: false,
643 last_order_signature: Some(order_signature),
644 },
645 );
646
647 self.pending_requests.insert(
648 request_id,
649 PendingRequestType::Buy {
650 client_order_id,
651 trader_id,
652 strategy_id,
653 instrument_id,
654 order_side: OrderSide::Buy,
655 order_type,
656 },
657 );
658
659 log::debug!("Sending buy order: request_id={request_id}");
660 self.send_with_retry(payload, Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()))
661 .await
662 }
663
664 async fn handle_sell(
666 &mut self,
667 params: DeribitOrderParams,
668 client_order_id: ClientOrderId,
669 trader_id: TraderId,
670 strategy_id: StrategyId,
671 instrument_id: InstrumentId,
672 ) -> Result<(), DeribitWsError> {
673 let request_id = self.next_request_id();
674 let order_type = parse_deribit_order_type(¶ms.order_type);
675 let order_signature = (params.amount, params.price, params.trigger_price);
676
677 let request =
678 DeribitJsonRpcRequest::new(request_id, DeribitWsMethod::Sell.as_method_str(), params);
679
680 let payload =
682 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()))?;
683
684 self.submitted_order_contexts.insert(
685 client_order_id,
686 OrderContext {
687 client_order_id,
688 trader_id,
689 strategy_id,
690 instrument_id,
691 order_side: OrderSide::Sell,
692 order_type,
693 accepted: false,
694 last_order_signature: Some(order_signature),
695 },
696 );
697
698 self.pending_requests.insert(
699 request_id,
700 PendingRequestType::Sell {
701 client_order_id,
702 trader_id,
703 strategy_id,
704 instrument_id,
705 order_side: OrderSide::Sell,
706 order_type,
707 },
708 );
709
710 log::debug!("Sending sell order: request_id={request_id}");
711 self.send_with_retry(payload, Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()))
712 .await
713 }
714
715 async fn handle_edit(
717 &mut self,
718 params: DeribitEditParams,
719 client_order_id: ClientOrderId,
720 trader_id: TraderId,
721 strategy_id: StrategyId,
722 instrument_id: InstrumentId,
723 ) -> Result<(), DeribitWsError> {
724 let request_id = self.next_request_id();
725 let order_id = params.order_id.clone();
726
727 self.pending_requests.insert(
728 request_id,
729 PendingRequestType::Edit {
730 client_order_id,
731 trader_id,
732 strategy_id,
733 instrument_id,
734 },
735 );
736
737 let request =
738 DeribitJsonRpcRequest::new(request_id, DeribitWsMethod::Edit.as_method_str(), params);
739
740 let payload =
741 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
742
743 log::debug!("Sending edit order: request_id={request_id}, order_id={order_id}");
744 self.send_tracked_request(
745 request_id,
746 payload,
747 Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()),
748 )
749 .await
750 }
751
752 async fn handle_cancel(
754 &mut self,
755 params: DeribitCancelParams,
756 client_order_id: ClientOrderId,
757 trader_id: TraderId,
758 strategy_id: StrategyId,
759 instrument_id: InstrumentId,
760 ) -> Result<(), DeribitWsError> {
761 let request_id = self.next_request_id();
762 let order_id = params.order_id.clone();
763
764 self.pending_requests.insert(
765 request_id,
766 PendingRequestType::Cancel {
767 client_order_id,
768 trader_id,
769 strategy_id,
770 instrument_id,
771 },
772 );
773
774 let request =
775 DeribitJsonRpcRequest::new(request_id, DeribitWsMethod::Cancel.as_method_str(), params);
776
777 let payload =
778 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
779
780 log::debug!("Sending cancel order: request_id={request_id}, order_id={order_id}");
781 self.send_tracked_request(
782 request_id,
783 payload,
784 Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()),
785 )
786 .await
787 }
788
789 async fn handle_cancel_all_by_instrument(
791 &mut self,
792 params: DeribitCancelAllByInstrumentParams,
793 instrument_id: InstrumentId,
794 ) -> Result<(), DeribitWsError> {
795 let request_id = self.next_request_id();
796 let instrument_name = params.instrument_name.clone();
797
798 self.pending_requests.insert(
800 request_id,
801 PendingRequestType::CancelAllByInstrument { instrument_id },
802 );
803
804 let request = DeribitJsonRpcRequest::new(
805 request_id,
806 DeribitWsMethod::CancelAllByInstrument.as_method_str(),
807 params,
808 );
809
810 let payload =
811 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
812
813 log::debug!(
814 "Sending cancel_all_by_instrument: request_id={request_id}, instrument={instrument_name}"
815 );
816 self.send_tracked_request(
817 request_id,
818 payload,
819 Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()),
820 )
821 .await
822 }
823
824 async fn handle_get_order_state(
826 &mut self,
827 order_id: String,
828 client_order_id: ClientOrderId,
829 trader_id: TraderId,
830 strategy_id: StrategyId,
831 instrument_id: InstrumentId,
832 ) -> Result<(), DeribitWsError> {
833 let request_id = self.next_request_id();
834
835 self.pending_requests.insert(
837 request_id,
838 PendingRequestType::GetOrderState {
839 client_order_id,
840 trader_id,
841 strategy_id,
842 instrument_id,
843 },
844 );
845
846 let params = serde_json::json!({
847 "order_id": order_id
848 });
849
850 let request = DeribitJsonRpcRequest::new(
851 request_id,
852 DeribitWsMethod::GetOrderState.as_method_str(),
853 params,
854 );
855
856 let payload =
857 serde_json::to_string(&request).map_err(|e| DeribitWsError::Json(e.to_string()));
858
859 log::debug!("Sending get_order_state: request_id={request_id}, order_id={order_id}");
860 self.send_tracked_request(
861 request_id,
862 payload,
863 Some(DERIBIT_RATE_LIMIT_KEY_ORDER.as_slice()),
864 )
865 .await
866 }
867
868 async fn process_command(&mut self, cmd: HandlerCommand) {
870 match cmd {
871 HandlerCommand::SetClient(client) => {
872 log::debug!("Setting WebSocket client");
873 self.inner = Some(Arc::new(client));
874 }
875 HandlerCommand::Disconnect => {
876 log::debug!("Disconnecting WebSocket");
877
878 if let Some(client) = self.inner.take() {
879 client.disconnect().await;
880 }
881 }
882 HandlerCommand::Authenticate { auth_params } => {
883 self.handle_authenticate(auth_params).await;
884 }
885 HandlerCommand::SetHeartbeat { interval } => {
886 if let Err(e) = self.handle_set_heartbeat(interval).await {
887 log::error!("Set heartbeat failed: {e}");
888 }
889 }
890 HandlerCommand::InitializeInstruments(instruments) => {
891 log::debug!("Handler received {} instruments", instruments.len());
892 self.instruments_cache.clear();
893 for inst in instruments {
894 self.instruments_cache
895 .insert(inst.raw_symbol().inner(), inst);
896 }
897 }
898 HandlerCommand::UpdateInstrument(instrument) => {
899 log::trace!("Updating instrument: {}", instrument.raw_symbol());
900 self.instruments_cache
901 .insert(instrument.raw_symbol().inner(), *instrument);
902 }
903 HandlerCommand::Subscribe { channels } => {
904 if let Err(e) = self.handle_subscribe(channels) {
905 log::error!("Subscribe failed: {e}");
906 }
907 }
908 HandlerCommand::Unsubscribe { channels } => {
909 self.pending_book_resync.retain(|ch| !channels.contains(ch));
912
913 if let Err(e) = self.handle_unsubscribe(channels) {
914 log::error!("Unsubscribe failed: {e}");
915 }
916 }
917 HandlerCommand::Buy {
918 params,
919 client_order_id,
920 trader_id,
921 strategy_id,
922 instrument_id,
923 } => {
924 if let Err(e) = self
925 .handle_buy(
926 params,
927 client_order_id,
928 trader_id,
929 strategy_id,
930 instrument_id,
931 )
932 .await
933 {
934 log::error!("Buy order failed: {e}");
935 }
936 }
937 HandlerCommand::Sell {
938 params,
939 client_order_id,
940 trader_id,
941 strategy_id,
942 instrument_id,
943 } => {
944 if let Err(e) = self
945 .handle_sell(
946 params,
947 client_order_id,
948 trader_id,
949 strategy_id,
950 instrument_id,
951 )
952 .await
953 {
954 log::error!("Sell order failed: {e}");
955 }
956 }
957 HandlerCommand::Edit {
958 params,
959 client_order_id,
960 trader_id,
961 strategy_id,
962 instrument_id,
963 } => {
964 if let Err(e) = self
965 .handle_edit(
966 params,
967 client_order_id,
968 trader_id,
969 strategy_id,
970 instrument_id,
971 )
972 .await
973 {
974 log::error!("Edit order failed: {e}");
975 }
976 }
977 HandlerCommand::Cancel {
978 params,
979 client_order_id,
980 trader_id,
981 strategy_id,
982 instrument_id,
983 } => {
984 if let Err(e) = self
985 .handle_cancel(
986 params,
987 client_order_id,
988 trader_id,
989 strategy_id,
990 instrument_id,
991 )
992 .await
993 {
994 log::error!("Cancel order failed: {e}");
995 }
996 }
997 HandlerCommand::CancelAllByInstrument {
998 params,
999 instrument_id,
1000 } => {
1001 if let Err(e) = self
1002 .handle_cancel_all_by_instrument(params, instrument_id)
1003 .await
1004 {
1005 log::error!("Cancel all by instrument failed: {e}");
1006 }
1007 }
1008 HandlerCommand::GetOrderState {
1009 order_id,
1010 client_order_id,
1011 trader_id,
1012 strategy_id,
1013 instrument_id,
1014 } => {
1015 if let Err(e) = self
1016 .handle_get_order_state(
1017 order_id,
1018 client_order_id,
1019 trader_id,
1020 strategy_id,
1021 instrument_id,
1022 )
1023 .await
1024 {
1025 log::error!("Get order state failed: {e}");
1026 }
1027 }
1028 }
1029 }
1030
1031 async fn process_raw_message(&mut self, text: &str) -> Option<NautilusWsMessage> {
1033 if text == RECONNECTED {
1034 log::info!("Received reconnection signal");
1035
1036 self.auth_tracker.invalidate();
1037 self.clear_state();
1038
1039 return Some(NautilusWsMessage::Reconnected);
1040 }
1041
1042 let ws_msg = match parse_raw_message(text) {
1044 Ok(msg) => msg,
1045 Err(e) => {
1046 log::warn!("Failed to parse message: {e}");
1047 return None;
1048 }
1049 };
1050
1051 let ts_init = self.ts_init();
1052
1053 match ws_msg {
1054 DeribitWsMessage::Response(response) => {
1055 if let Some(request_id) = response.id
1057 && let Some(request_type) = self.pending_requests.remove(&request_id)
1058 {
1059 match request_type {
1060 PendingRequestType::Authenticate => {
1061 if let Some(error) = &response.error {
1062 let reason = format!(
1063 "Authentication error code={}: {}",
1064 error.code, error.message
1065 );
1066 log::error!(
1067 "Authentication failed: code={}, message={}, request_id={}",
1068 error.code,
1069 error.message,
1070 request_id
1071 );
1072 self.auth_tracker.fail(reason.clone());
1073 return Some(NautilusWsMessage::AuthenticationFailed(reason));
1074 } else if let Some(result) = &response.result {
1075 match serde_json::from_str::<DeribitAuthResult>(result.get()) {
1076 Ok(auth_result) => {
1077 self.auth_tracker.succeed();
1078 log::debug!(
1079 "WebSocket authenticated successfully (request_id={}, scope={}, expires_in={}s)",
1080 request_id,
1081 auth_result.scope,
1082 auth_result.expires_in
1083 );
1084 return Some(NautilusWsMessage::Authenticated(Box::new(
1085 auth_result,
1086 )));
1087 }
1088 Err(e) => {
1089 let reason = format!("Failed to parse auth result: {e}");
1090 log::error!("{reason}: request_id={request_id}");
1091 self.auth_tracker.fail(reason.clone());
1092 return Some(NautilusWsMessage::AuthenticationFailed(
1093 reason,
1094 ));
1095 }
1096 }
1097 }
1098 }
1099 PendingRequestType::Subscribe { channels } => {
1100 if let Some(error) = &response.error {
1101 log::error!(
1102 "Subscribe failed: code={}, message={}, channels={:?}, request_id={}",
1103 error.code,
1104 error.message,
1105 channels,
1106 request_id
1107 );
1108
1109 self.subscribe_errors.lock().push(format!(
1110 "Subscribe rejected: code={}, message={}",
1111 error.code, error.message,
1112 ));
1113 } else {
1114 for ch in &channels {
1116 self.subscriptions_state.confirm_subscribe(ch);
1117 log::debug!("Subscription confirmed: {ch}");
1118 }
1119 }
1120 }
1121 PendingRequestType::Unsubscribe { channels } => {
1122 if let Some(error) = &response.error {
1123 log::error!(
1124 "Unsubscribe failed: code={}, message={}, channels={:?}, request_id={}",
1125 error.code,
1126 error.message,
1127 channels,
1128 request_id
1129 );
1130 } else {
1131 for ch in &channels {
1132 self.subscriptions_state.confirm_unsubscribe(ch);
1133 log::debug!("Unsubscription confirmed: {ch}");
1134 }
1135 }
1136
1137 if !self.pending_book_resync.is_empty() {
1140 let resync: Vec<String> = channels
1141 .iter()
1142 .filter(|ch| self.pending_book_resync.contains(ch))
1143 .cloned()
1144 .collect();
1145
1146 if !resync.is_empty() {
1147 let _ = self.handle_subscribe(resync);
1148 }
1149 }
1150 }
1151 PendingRequestType::SetHeartbeat => {
1152 if let Some(error) = &response.error {
1153 log::error!(
1154 "Set heartbeat failed: code={}, message={}, request_id={}",
1155 error.code,
1156 error.message,
1157 request_id
1158 );
1159 } else {
1160 log::debug!("Heartbeat enabled (request_id={request_id})");
1161 }
1162 }
1163 PendingRequestType::Test => {
1164 if let Some(error) = &response.error {
1165 log::warn!(
1166 "Heartbeat test failed: code={}, message={}, request_id={}",
1167 error.code,
1168 error.message,
1169 request_id
1170 );
1171 } else {
1172 log::trace!(
1173 "Heartbeat test acknowledged (request_id={request_id})"
1174 );
1175 }
1176 }
1177 PendingRequestType::Cancel {
1178 client_order_id,
1179 trader_id,
1180 strategy_id,
1181 instrument_id,
1182 } => {
1183 if let Some(result) = &response.result {
1184 match serde_json::from_str::<DeribitOrderMsg>(result.get()) {
1185 Ok(order_msg) => {
1186 let venue_order_id =
1187 VenueOrderId::new(order_msg.order_id.as_str());
1188 log::debug!(
1189 "Cancel confirmed: venue_order_id={venue_order_id}, \
1190 client_order_id={client_order_id}, state={}",
1191 order_msg.order_state
1192 );
1193
1194 if order_msg.order_state == "cancelled"
1199 && !self
1200 .terminal_order_contexts
1201 .contains_key(&venue_order_id)
1202 {
1203 let instrument_name_ustr = order_msg.instrument_name;
1204
1205 if let Some(instrument) =
1206 self.instruments_cache.get(&instrument_name_ustr)
1207 && let Some(account_id) = self.account_id
1208 {
1209 let event =
1210 parse_order_canceled_with_client_order_id(
1211 &order_msg,
1212 instrument,
1213 account_id,
1214 trader_id,
1215 strategy_id,
1216 client_order_id,
1217 ts_init,
1218 );
1219 let context = self
1220 .find_order_context(
1221 venue_order_id,
1222 Some(client_order_id),
1223 )
1224 .or_else(|| {
1225 let order_side =
1226 match order_msg.direction.as_str() {
1227 "buy" => OrderSide::Buy,
1228 "sell" => OrderSide::Sell,
1229 _ => return None,
1230 };
1231 Some(OrderContext {
1232 client_order_id,
1233 trader_id,
1234 strategy_id,
1235 instrument_id,
1236 order_side,
1237 order_type: parse_deribit_order_type(
1238 &order_msg.order_type,
1239 ),
1240 accepted: true,
1241 last_order_signature: Some(
1242 Self::order_signature(&order_msg),
1243 ),
1244 })
1245 });
1246
1247 if let Some(context) = context {
1248 self.finish_order_context(
1249 venue_order_id,
1250 &context,
1251 );
1252 }
1253 return Some(NautilusWsMessage::OrderCanceled(
1254 event,
1255 ));
1256 }
1257 }
1258 }
1259 Err(e) => {
1260 log::error!(
1261 "Failed to parse cancel response: request_id={request_id}, error={e}"
1262 );
1263 }
1264 }
1265 } else if let Some(error) = &response.error {
1266 log::warn!(
1267 "Cancel rejected: code={}, message={}, client_order_id={}",
1268 error.code,
1269 error.message,
1270 client_order_id
1271 );
1272 return Some(NautilusWsMessage::OrderCancelRejected(
1273 OrderCancelRejected::new(
1274 trader_id,
1275 strategy_id,
1276 instrument_id,
1277 client_order_id,
1278 ustr::ustr(&format!(
1279 "code={}: {}",
1280 error.code, error.message
1281 )),
1282 UUID4::new(),
1283 ts_init,
1284 ts_init,
1285 false,
1286 None, self.account_id,
1288 ),
1289 ));
1290 }
1291 }
1292 PendingRequestType::CancelAllByInstrument { instrument_id } => {
1293 if let Some(result) = &response.result {
1294 match serde_json::from_str::<u64>(result.get()) {
1295 Ok(count) => {
1296 log::debug!(
1297 "Cancelled {count} orders for instrument {instrument_id}"
1298 );
1299 }
1301 Err(e) => {
1302 log::warn!("Failed to parse cancel_all response: {e}");
1303 }
1304 }
1305 } else if let Some(error) = &response.error {
1306 log::error!(
1307 "Cancel all by instrument rejected: code={}, message={}, instrument_id={}",
1308 error.code,
1309 error.message,
1310 instrument_id
1311 );
1312 }
1313 }
1314 PendingRequestType::Buy {
1315 client_order_id,
1316 trader_id,
1317 strategy_id,
1318 instrument_id,
1319 order_side,
1320 order_type,
1321 }
1322 | PendingRequestType::Sell {
1323 client_order_id,
1324 trader_id,
1325 strategy_id,
1326 instrument_id,
1327 order_side,
1328 order_type,
1329 } => {
1330 if let Some(result) = &response.result {
1331 match serde_json::from_str::<DeribitOrderResponse>(result.get()) {
1332 Ok(order_response) => {
1333 let venue_order_id_str = &order_response.order.order_id;
1334 let venue_order_id =
1335 VenueOrderId::new(venue_order_id_str.as_str());
1336 let order_state = &order_response.order.order_state;
1337 log::debug!(
1338 "Order response: venue_order_id={venue_order_id}, client_order_id={client_order_id}, state={order_state}"
1339 );
1340
1341 let mut context = self
1342 .find_order_context(
1343 venue_order_id,
1344 Some(client_order_id),
1345 )
1346 .unwrap_or(OrderContext {
1347 client_order_id,
1348 trader_id,
1349 strategy_id,
1350 instrument_id,
1351 order_side,
1352 order_type,
1353 accepted: false,
1354 last_order_signature: Some(Self::order_signature(
1355 &order_response.order,
1356 )),
1357 });
1358 context.last_order_signature =
1359 Some(Self::order_signature(&order_response.order));
1360 self.bind_order_context(venue_order_id, context.clone());
1361
1362 if !order_response.trades.is_empty() {
1363 let outgoing = self
1364 .route_user_trades(&order_response.trades, ts_init);
1365 self.pending_outgoing.extend(outgoing);
1366 } else if order_state == "filled" {
1367 log::debug!(
1368 "Deferring acceptance for fast-filled order until its trade arrives: venue_order_id={venue_order_id}, client_order_id={client_order_id}"
1369 );
1370 self.finish_order_context(venue_order_id, &context);
1371 } else if context.accepted {
1372 log::trace!(
1373 "Skipping duplicate OrderAccepted response: venue_order_id={venue_order_id}"
1374 );
1375 } else {
1376 let instrument_name_ustr = Ustr::from(
1377 order_response.order.instrument_name.as_str(),
1378 );
1379
1380 if let Some(instrument) =
1381 self.instruments_cache.get(&instrument_name_ustr)
1382 {
1383 if let Some(account_id) = self.account_id {
1384 let event =
1385 parse_order_accepted_with_client_order_id(
1386 &order_response.order,
1387 instrument,
1388 account_id,
1389 trader_id,
1390 strategy_id,
1391 client_order_id,
1392 ts_init,
1393 );
1394 context.accepted = true;
1395 self.bind_order_context(
1396 venue_order_id,
1397 context,
1398 );
1399 return Some(NautilusWsMessage::OrderAccepted(
1400 event,
1401 ));
1402 } else {
1403 log::warn!(
1404 "Cannot create OrderAccepted: account_id not set"
1405 );
1406 }
1407 } else {
1408 log::warn!(
1409 "Instrument {instrument_name_ustr} not found in cache for order response"
1410 );
1411 }
1412 }
1413 }
1414 Err(e) => {
1415 log::error!(
1416 "Failed to parse order response: request_id={request_id}, error={e}"
1417 );
1418 }
1419 }
1420 } else if let Some(error) = &response.error {
1421 let due_post_only = error.code == DERIBIT_POST_ONLY_ERROR_CODE;
1422 let reason = if let Some(data) = &error.data {
1423 format!(
1424 "code={}: {} (data: {})",
1425 error.code, error.message, data
1426 )
1427 } else {
1428 format!("code={}: {}", error.code, error.message)
1429 };
1430
1431 log::debug!(
1432 "Order rejected: {reason}, client_order_id={client_order_id}"
1433 );
1434 self.submitted_order_contexts.remove(&client_order_id);
1435 return Some(NautilusWsMessage::OrderRejected(OrderRejected::new(
1436 trader_id,
1437 strategy_id,
1438 instrument_id,
1439 client_order_id,
1440 self.account_id.unwrap_or(AccountId::new("DERIBIT-UNKNOWN")),
1441 ustr::ustr(&reason),
1442 UUID4::new(),
1443 ts_init,
1444 ts_init,
1445 false,
1446 due_post_only,
1447 )));
1448 }
1449 }
1450 PendingRequestType::Edit {
1451 client_order_id,
1452 trader_id,
1453 strategy_id,
1454 instrument_id,
1455 } => {
1456 if let Some(result) = &response.result {
1457 match serde_json::from_str::<DeribitOrderResponse>(result.get()) {
1458 Ok(order_response) => {
1459 let venue_order_id =
1460 VenueOrderId::new(&order_response.order.order_id);
1461 log::debug!(
1462 "Order updated: venue_order_id={}, client_order_id={}, state={}",
1463 venue_order_id,
1464 client_order_id,
1465 order_response.order.order_state
1466 );
1467
1468 let instrument_name_ustr = Ustr::from(
1469 order_response.order.instrument_name.as_str(),
1470 );
1471
1472 if let Some(instrument) =
1473 self.instruments_cache.get(&instrument_name_ustr)
1474 {
1475 if let Some(account_id) = self.account_id {
1476 let Some(mut context) = self.find_order_context(
1477 venue_order_id,
1478 Some(client_order_id),
1479 ) else {
1480 let report = parse_user_order_msg(
1481 &order_response.order,
1482 instrument,
1483 account_id,
1484 ts_init,
1485 );
1486 let outgoing = self.route_user_trades(
1487 &order_response.trades,
1488 ts_init,
1489 );
1490 self.pending_outgoing.extend(outgoing);
1491 return report
1492 .map(|report| {
1493 NautilusWsMessage::OrderStatusReports(vec![
1494 report,
1495 ])
1496 })
1497 .map_err(|e| {
1498 log::warn!(
1499 "Failed to parse external edit response: {e}"
1500 );
1501 })
1502 .ok();
1503 };
1504 let was_terminal = self
1505 .terminal_order_contexts
1506 .contains_key(&venue_order_id);
1507 let signature =
1508 Self::order_signature(&order_response.order);
1509 let duplicate_update = was_terminal
1510 || context.last_order_signature
1511 == Some(signature);
1512 let event = (!duplicate_update).then(|| {
1513 parse_order_updated_with_client_order_id(
1514 &order_response.order,
1515 instrument,
1516 account_id,
1517 context.trader_id,
1518 context.strategy_id,
1519 context.client_order_id,
1520 ts_init,
1521 )
1522 });
1523 context.accepted = true;
1524 context.last_order_signature = Some(signature);
1525
1526 if was_terminal {
1527 self.finish_order_context(
1528 venue_order_id,
1529 &context,
1530 );
1531 } else {
1532 self.bind_order_context(
1533 venue_order_id,
1534 context,
1535 );
1536 }
1537 let outgoing = self.route_user_trades(
1538 &order_response.trades,
1539 ts_init,
1540 );
1541 self.pending_outgoing.extend(outgoing);
1542 return event.map(NautilusWsMessage::OrderUpdated);
1543 } else {
1544 log::warn!(
1545 "Cannot create OrderUpdated: account_id not set"
1546 );
1547 }
1548 } else {
1549 log::warn!(
1550 "Instrument {instrument_name_ustr} not found in cache for edit response"
1551 );
1552 }
1553 }
1554 Err(e) => {
1555 log::error!(
1556 "Failed to parse edit response: request_id={request_id}, error={e}"
1557 );
1558 }
1559 }
1560 } else if let Some(error) = &response.error {
1561 log::warn!(
1562 "Order modify rejected: code={}, message={}, client_order_id={}",
1563 error.code,
1564 error.message,
1565 client_order_id
1566 );
1567 return Some(NautilusWsMessage::OrderModifyRejected(
1568 OrderModifyRejected::new(
1569 trader_id,
1570 strategy_id,
1571 instrument_id,
1572 client_order_id,
1573 ustr::ustr(&format!(
1574 "code={}: {}",
1575 error.code, error.message
1576 )),
1577 UUID4::new(),
1578 ts_init,
1579 ts_init,
1580 false,
1581 None, self.account_id,
1583 ),
1584 ));
1585 }
1586 }
1587 PendingRequestType::GetOrderState {
1588 client_order_id,
1589 trader_id: _,
1590 strategy_id: _,
1591 instrument_id: _,
1592 } => {
1593 if let Some(result) = &response.result {
1594 match serde_json::from_str::<DeribitOrderMsg>(result.get()) {
1595 Ok(order_msg) => {
1596 log::debug!(
1597 "Order state received: venue_order_id={}, client_order_id={}, state={}",
1598 order_msg.order_id,
1599 client_order_id,
1600 order_msg.order_state
1601 );
1602
1603 let instrument_name_ustr = order_msg.instrument_name;
1605
1606 if let Some(instrument) =
1607 self.instruments_cache.get(&instrument_name_ustr)
1608 {
1609 if let Some(account_id) = self.account_id {
1610 match parse_user_order_msg(
1611 &order_msg, instrument, account_id, ts_init,
1612 ) {
1613 Ok(report) => {
1614 return Some(
1615 NautilusWsMessage::OrderStatusReports(
1616 vec![report],
1617 ),
1618 );
1619 }
1620 Err(e) => {
1621 log::warn!(
1622 "Failed to parse get_order_state response to report: {e}"
1623 );
1624 }
1625 }
1626 } else {
1627 log::warn!(
1628 "Cannot create OrderStatusReport: account_id not set"
1629 );
1630 }
1631 } else {
1632 log::warn!(
1633 "Instrument {instrument_name_ustr} not found in cache for get_order_state response"
1634 );
1635 }
1636 }
1637 Err(e) => {
1638 log::error!(
1639 "Failed to parse get_order_state response: request_id={request_id}, error={e}"
1640 );
1641 }
1642 }
1643 } else if let Some(error) = &response.error {
1644 log::error!(
1645 "Get order state failed: code={}, message={}, client_order_id={}",
1646 error.code,
1647 error.message,
1648 client_order_id
1649 );
1650 }
1651 }
1652 }
1653 } else if let Some(request_id) = response.id {
1654 if let Some(error) = &response.error {
1656 log::error!(
1658 "Deribit error for unknown request: code={}, message={}, request_id={}, data={:?}",
1659 error.code,
1660 error.message,
1661 request_id,
1662 error.data
1663 );
1664 return Some(NautilusWsMessage::Error(DeribitWsError::DeribitError {
1665 code: error.code,
1666 message: error.message.clone(),
1667 }));
1668 } else {
1669 log::debug!(
1671 "Received response for unknown request_id={}, result present: {}",
1672 request_id,
1673 response.result.is_some()
1674 );
1675 }
1676 } else if let Some(error) = &response.error {
1677 log::error!(
1679 "Deribit error with no request_id: code={}, message={}, data={:?}",
1680 error.code,
1681 error.message,
1682 error.data
1683 );
1684 return Some(NautilusWsMessage::Error(DeribitWsError::DeribitError {
1685 code: error.code,
1686 message: error.message.clone(),
1687 }));
1688 }
1689 None
1690 }
1691 DeribitWsMessage::Notification(notification) => {
1692 let channel = ¬ification.params.channel;
1693 let data = ¬ification.params.data;
1694
1695 if let Some(channel_type) = DeribitWsChannel::from_channel_string(channel) {
1697 match channel_type {
1698 DeribitWsChannel::Trades => {
1699 match serde_json::from_str::<Vec<DeribitTradeMsg>>(data.get()) {
1701 Ok(trades) => {
1702 log::debug!("Received {} trades", trades.len());
1703 let data_vec = parse_trades_data(
1704 &trades,
1705 &self.instruments_cache,
1706 ts_init,
1707 );
1708
1709 if data_vec.is_empty() && !trades.is_empty() {
1710 let missing: Vec<&Ustr> = trades
1711 .iter()
1712 .map(|t| &t.instrument_name)
1713 .filter(|name| {
1714 !self.instruments_cache.contains_key(name)
1715 })
1716 .collect();
1717
1718 if missing.is_empty() {
1719 log::warn!(
1720 "Received {} trades but parsed 0 (parse failures); cache size: {}",
1721 trades.len(),
1722 self.instruments_cache.len()
1723 );
1724 } else {
1725 log::warn!(
1726 "Trade message received but instrument(s) not found in cache: {:?} (cache size: {})",
1727 missing,
1728 self.instruments_cache.len()
1729 );
1730 }
1731 } else if !data_vec.is_empty() {
1732 log::debug!("Parsed {} trade ticks", data_vec.len());
1733 return Some(NautilusWsMessage::Data(data_vec));
1734 }
1735 }
1736 Err(e) => {
1737 log::warn!("Failed to deserialize trades: {e}");
1738 }
1739 }
1740 }
1741 DeribitWsChannel::Book => {
1742 return self.handle_book_message(channel, data.get(), ts_init).await;
1743 }
1744 DeribitWsChannel::Ticker => {
1745 match serde_json::from_str::<DeribitTickerMsg>(data.get()) {
1746 Ok(ticker_msg) => {
1747 if let Some(instrument) =
1748 self.instruments_cache.get(&ticker_msg.instrument_name)
1749 {
1750 if self.option_greeks_subs.contains(&instrument.id())
1752 && let Some(option_greeks) =
1753 parse_ticker_to_option_greeks(
1754 &ticker_msg,
1755 instrument,
1756 ts_init,
1757 )
1758 {
1759 let _ = self.out_tx.send(
1760 NautilusWsMessage::OptionGreeks(option_greeks),
1761 );
1762 }
1763
1764 let instrument_id = instrument.id();
1765 let mut data_vec = Vec::new();
1766
1767 if self.mark_price_subs.contains(&instrument_id) {
1769 match parse_ticker_to_mark_price(
1770 &ticker_msg,
1771 instrument,
1772 ts_init,
1773 ) {
1774 Ok(mark_price) => {
1775 data_vec.push(Data::MarkPrice(mark_price));
1776 }
1777 Err(e) => {
1778 log::warn!("Failed to parse mark price: {e}");
1779 }
1780 }
1781 }
1782
1783 if self.index_price_subs.contains(&instrument_id) {
1785 match parse_ticker_to_index_price(
1786 &ticker_msg,
1787 instrument,
1788 ts_init,
1789 ) {
1790 Ok(index_price) => {
1791 data_vec.push(Data::IndexPrice(index_price));
1792 }
1793 Err(e) => {
1794 log::warn!("Failed to parse index price: {e}");
1795 }
1796 }
1797 }
1798
1799 if !data_vec.is_empty() {
1800 return Some(NautilusWsMessage::Data(data_vec));
1801 }
1802 } else {
1803 log::warn!(
1804 "Ticker message received but instrument '{}' not found in cache (cache size: {})",
1805 ticker_msg.instrument_name,
1806 self.instruments_cache.len()
1807 );
1808 }
1809 }
1810 Err(e) => {
1811 log::warn!(
1812 "Failed to deserialize ticker message: {e}, channel: {channel}"
1813 );
1814 }
1815 }
1816 }
1817 DeribitWsChannel::Perpetual => {
1818 match serde_json::from_str::<DeribitPerpetualMsg>(data.get()) {
1822 Ok(perpetual_msg) => {
1823 let parts: Vec<&str> = channel.split('.').collect();
1825 if parts.len() >= 2 {
1826 let instrument_name = Ustr::from(parts[1]);
1827
1828 if let Some(instrument) =
1829 self.instruments_cache.get(&instrument_name)
1830 {
1831 let funding_rate = parse_perpetual_to_funding_rate(
1832 &perpetual_msg,
1833 instrument,
1834 ts_init,
1835 );
1836 return Some(NautilusWsMessage::FundingRates(vec![
1837 funding_rate,
1838 ]));
1839 } else {
1840 log::warn!(
1841 "Instrument {} not found in cache (cache size: {})",
1842 instrument_name,
1843 self.instruments_cache.len()
1844 );
1845 }
1846 }
1847 }
1848 Err(e) => {
1849 log::warn!(
1850 "Failed to deserialize perpetual message: {e}, data: {data}"
1851 );
1852 }
1853 }
1854 }
1855 DeribitWsChannel::Quote => {
1856 match serde_json::from_str::<DeribitQuoteMsg>(data.get()) {
1858 Ok(quote_msg) => {
1859 if let Some(instrument) =
1860 self.instruments_cache.get("e_msg.instrument_name)
1861 {
1862 match parse_quote_msg("e_msg, instrument, ts_init) {
1863 Ok(quote) => {
1864 return Some(NautilusWsMessage::Data(vec![
1865 Data::Quote(quote),
1866 ]));
1867 }
1868 Err(e) => {
1869 log::warn!("Failed to parse quote message: {e}");
1870 }
1871 }
1872 } else {
1873 log::warn!(
1874 "Quote message received but instrument '{}' not found in cache (cache size: {})",
1875 quote_msg.instrument_name,
1876 self.instruments_cache.len()
1877 );
1878 }
1879 }
1880 Err(e) => {
1881 log::warn!(
1882 "Failed to deserialize quote message: {e}, channel: {channel}"
1883 );
1884 }
1885 }
1886 }
1887 DeribitWsChannel::VolatilityIndex => {
1888 match serde_json::from_str::<DeribitVolatilityIndexMsg>(data.get()) {
1889 Ok(msg) => {
1890 let ts_event = UnixNanos::from(msg.timestamp * 1_000_000);
1891
1892 let metadata = nautilus_core::Params::new(
1893 [(
1894 "index_name".to_string(),
1895 serde_json::Value::String(msg.index_name.clone()),
1896 )]
1897 .into(),
1898 );
1899 let data_type = DataType::new(
1900 "DeribitVolatilityIndex",
1901 Some(metadata),
1902 None,
1903 );
1904
1905 let dvol = DeribitVolatilityIndex::new(
1906 msg.index_name,
1907 msg.volatility,
1908 ts_event,
1909 ts_init,
1910 );
1911
1912 return Some(NautilusWsMessage::Data(vec![Data::Custom(
1913 CustomData::new(Arc::new(dvol), data_type),
1914 )]));
1915 }
1916 Err(e) => {
1917 log::warn!("Failed to deserialize volatility index: {e}");
1918 }
1919 }
1920 }
1921 DeribitWsChannel::InstrumentState => {
1922 match serde_json::from_str::<DeribitInstrumentStateMsg>(data.get()) {
1923 Ok(state_msg) => {
1924 log::debug!(
1925 "Instrument state change: {} -> {} (timestamp: {})",
1926 state_msg.instrument_name,
1927 state_msg.state,
1928 state_msg.timestamp
1929 );
1930
1931 let instrument_id = if let Some(instrument) =
1932 self.instruments_cache.get(&state_msg.instrument_name)
1933 {
1934 instrument.id()
1935 } else {
1936 log::debug!(
1937 "Instrument '{}' not in cache, constructing ID",
1938 state_msg.instrument_name
1939 );
1940 InstrumentId::new(
1941 Symbol::new(state_msg.instrument_name),
1942 *DERIBIT_VENUE,
1943 )
1944 };
1945
1946 let action = MarketStatusAction::from(state_msg.state);
1947 let is_trading =
1948 Some(state_msg.state == DeribitInstrumentState::Started);
1949 let ts_event = UnixNanos::from(state_msg.timestamp * 1_000_000);
1950 let status = InstrumentStatus::new(
1951 instrument_id,
1952 action,
1953 ts_event,
1954 ts_init,
1955 None,
1956 None,
1957 is_trading,
1958 None,
1959 None,
1960 );
1961 return Some(NautilusWsMessage::InstrumentStatus(status));
1962 }
1963 Err(e) => {
1964 log::warn!("Failed to parse instrument status message: {e}");
1965 }
1966 }
1967 }
1968 DeribitWsChannel::ChartTrades => {
1969 if let Ok(chart_msg) =
1974 serde_json::from_str::<DeribitChartMsg>(data.get())
1975 {
1976 let parts: Vec<&str> = channel.split('.').collect();
1979 if parts.len() >= 4 {
1980 let instrument_name = Ustr::from(parts[2]);
1981 let resolution = parts[3];
1982
1983 if let Some(instrument) =
1984 self.instruments_cache.get(&instrument_name)
1985 {
1986 let instrument_id = instrument.id();
1987
1988 match resolution_to_bar_type(instrument_id, resolution) {
1989 Ok(bar_type) => {
1990 let price_precision = instrument.price_precision();
1991 let size_precision = instrument.size_precision();
1992 let use_cost_for_volume =
1993 use_cost_for_bar_volume(instrument);
1994
1995 match parse_chart_msg(
1996 &chart_msg,
1997 bar_type,
1998 price_precision,
1999 size_precision,
2000 use_cost_for_volume,
2001 self.bars_timestamp_on_close,
2002 ts_init,
2003 ) {
2004 Ok(new_bar) => {
2005 let channel_key = channel.clone();
2007
2008 if let Some(pending_bar) =
2009 self.pending_bars.get(&channel_key)
2010 {
2011 if new_bar.ts_event
2013 != pending_bar.ts_event
2014 {
2015 let closed_bar = *pending_bar;
2016 self.pending_bars
2017 .insert(channel_key, new_bar);
2018 log::debug!(
2019 "Emitting closed bar: {closed_bar:?}"
2020 );
2021 return Some(
2022 NautilusWsMessage::Data(vec![
2023 Data::Bar(closed_bar),
2024 ]),
2025 );
2026 }
2027 self.pending_bars
2029 .insert(channel_key, new_bar);
2030 } else {
2031 self.pending_bars
2033 .insert(channel_key, new_bar);
2034 }
2035 }
2036 Err(e) => {
2037 log::warn!(
2038 "Failed to parse chart message to bar: {e}"
2039 );
2040 }
2041 }
2042 }
2043 Err(e) => {
2044 log::warn!(
2045 "Failed to create BarType from resolution {resolution}: {e}"
2046 );
2047 }
2048 }
2049 } else {
2050 log::warn!(
2051 "Instrument {instrument_name} not found in cache for chart data"
2052 );
2053 }
2054 }
2055 }
2056 }
2057 DeribitWsChannel::UserOrders => {
2058 let orders_result = serde_json::from_str::<Vec<DeribitOrderMsg>>(
2060 data.get(),
2061 )
2062 .or_else(|_| {
2063 serde_json::from_str::<DeribitOrderMsg>(data.get())
2064 .map(|order| vec![order])
2065 });
2066
2067 match orders_result {
2068 Ok(orders) => {
2069 log::debug!("Received {} user order updates", orders.len());
2070
2071 let Some(account_id) = self.account_id else {
2073 log::warn!("Cannot parse user orders: account_id not set");
2074 return Some(NautilusWsMessage::Raw(data.clone()));
2075 };
2076
2077 let mut outgoing = Vec::new();
2078
2079 for order in &orders {
2081 let venue_order_id_str = &order.order_id;
2082 let venue_order_id =
2083 VenueOrderId::new(venue_order_id_str.as_str());
2084 let instrument_name = order.instrument_name;
2085
2086 let Some(instrument) =
2087 self.instruments_cache.get(&instrument_name)
2088 else {
2089 log::warn!(
2090 "Instrument {instrument_name} not found in cache"
2091 );
2092 continue;
2093 };
2094
2095 let label_client_order_id = order
2096 .label
2097 .as_ref()
2098 .filter(|l| !l.is_empty())
2099 .map(ClientOrderId::new);
2100 let was_terminal = self
2101 .terminal_order_contexts
2102 .contains_key(&venue_order_id);
2103 let Some(mut context) = self.find_order_context(
2104 venue_order_id,
2105 label_client_order_id,
2106 ) else {
2107 match parse_user_order_msg(
2108 order, instrument, account_id, ts_init,
2109 ) {
2110 Ok(report) => outgoing.push(
2111 NautilusWsMessage::OrderStatusReports(vec![
2112 report,
2113 ]),
2114 ),
2115 Err(e) => log::warn!(
2116 "Failed to parse external order update: {e}"
2117 ),
2118 }
2119 continue;
2120 };
2121
2122 let signature = Self::order_signature(order);
2123
2124 let event_type = determine_order_event_type(
2126 &order.order_state,
2127 !context.accepted,
2128 order.replaced
2129 && context.last_order_signature != Some(signature),
2130 );
2131 context.last_order_signature = Some(signature);
2132
2133 let trader_id = context.trader_id;
2134 let strategy_id = context.strategy_id;
2135 let client_order_id = context.client_order_id;
2136
2137 match event_type {
2138 OrderEventType::Accepted => {
2139 if self
2141 .terminal_order_contexts
2142 .contains_key(&venue_order_id)
2143 {
2144 log::debug!(
2145 "Skipping OrderAccepted for terminal order: client_order_id={client_order_id}"
2146 );
2147 continue;
2148 }
2149
2150 let event =
2151 parse_order_accepted_with_client_order_id(
2152 order,
2153 instrument,
2154 account_id,
2155 trader_id,
2156 strategy_id,
2157 client_order_id,
2158 ts_init,
2159 );
2160 context.accepted = true;
2161 self.bind_order_context(venue_order_id, context);
2162
2163 log::debug!(
2164 "Emitting OrderAccepted: venue_order_id={venue_order_id}"
2165 );
2166 outgoing
2167 .push(NautilusWsMessage::OrderAccepted(event));
2168 }
2169 OrderEventType::Canceled => {
2170 if self
2173 .terminal_order_contexts
2174 .contains_key(&venue_order_id)
2175 {
2176 log::trace!(
2177 "Skipping duplicate OrderCanceled: client_order_id={client_order_id}"
2178 );
2179 continue;
2180 }
2181
2182 if !context.accepted {
2183 outgoing
2184 .push(NautilusWsMessage::OrderAccepted(
2185 parse_order_accepted_with_client_order_id(
2186 order,
2187 instrument,
2188 account_id,
2189 trader_id,
2190 strategy_id,
2191 client_order_id,
2192 ts_init,
2193 ),
2194 ));
2195 context.accepted = true;
2196 }
2197
2198 let event =
2199 parse_order_canceled_with_client_order_id(
2200 order,
2201 instrument,
2202 account_id,
2203 trader_id,
2204 strategy_id,
2205 client_order_id,
2206 ts_init,
2207 );
2208 log::debug!(
2209 "Emitting OrderCanceled: venue_order_id={venue_order_id}"
2210 );
2211 self.finish_order_context(venue_order_id, &context);
2212 outgoing
2213 .push(NautilusWsMessage::OrderCanceled(event));
2214 }
2215 OrderEventType::Expired => {
2216 if self
2217 .terminal_order_contexts
2218 .contains_key(&venue_order_id)
2219 {
2220 log::trace!(
2221 "Skipping duplicate OrderExpired: client_order_id={client_order_id}"
2222 );
2223 continue;
2224 }
2225
2226 if !context.accepted {
2227 outgoing
2228 .push(NautilusWsMessage::OrderAccepted(
2229 parse_order_accepted_with_client_order_id(
2230 order,
2231 instrument,
2232 account_id,
2233 trader_id,
2234 strategy_id,
2235 client_order_id,
2236 ts_init,
2237 ),
2238 ));
2239 context.accepted = true;
2240 }
2241
2242 let event =
2243 parse_order_expired_with_client_order_id(
2244 order,
2245 instrument,
2246 account_id,
2247 trader_id,
2248 strategy_id,
2249 client_order_id,
2250 ts_init,
2251 );
2252 log::debug!(
2253 "Emitting OrderExpired: venue_order_id={venue_order_id}"
2254 );
2255 self.finish_order_context(venue_order_id, &context);
2256 outgoing
2257 .push(NautilusWsMessage::OrderExpired(event));
2258 }
2259 OrderEventType::Updated => {
2260 if was_terminal {
2261 log::trace!(
2262 "Skipping amendment for terminal order: venue_order_id={venue_order_id}"
2263 );
2264 continue;
2265 }
2266
2267 if !context.accepted {
2268 outgoing
2269 .push(NautilusWsMessage::OrderAccepted(
2270 parse_order_accepted_with_client_order_id(
2271 order,
2272 instrument,
2273 account_id,
2274 trader_id,
2275 strategy_id,
2276 client_order_id,
2277 ts_init,
2278 ),
2279 ));
2280 context.accepted = true;
2281 }
2282
2283 let event =
2284 parse_order_updated_with_client_order_id(
2285 order,
2286 instrument,
2287 account_id,
2288 trader_id,
2289 strategy_id,
2290 client_order_id,
2291 ts_init,
2292 );
2293 self.bind_order_context(venue_order_id, context);
2294 log::debug!(
2295 "Emitting OrderUpdated: venue_order_id={venue_order_id}"
2296 );
2297 outgoing
2298 .push(NautilusWsMessage::OrderUpdated(event));
2299 }
2300 OrderEventType::None => {
2301 if order.order_state == "filled" {
2304 log::debug!(
2305 "Recording terminal order: venue_order_id={venue_order_id}, state={}",
2306 order.order_state
2307 );
2308 self.finish_order_context(
2309 venue_order_id,
2310 &context,
2311 );
2312 } else if order.order_state == "rejected" {
2313 log::debug!(
2314 "Recording rejected order: venue_order_id={venue_order_id}"
2315 );
2316 self.finish_order_context(
2317 venue_order_id,
2318 &context,
2319 );
2320 } else if was_terminal {
2321 self.finish_order_context(
2322 venue_order_id,
2323 &context,
2324 );
2325 } else {
2326 log::trace!(
2327 "No event to emit for order {}, state={}",
2328 venue_order_id,
2329 order.order_state
2330 );
2331 self.bind_order_context(
2332 venue_order_id,
2333 context,
2334 );
2335 }
2336 }
2337 }
2338 }
2339
2340 if !outgoing.is_empty() {
2341 self.pending_outgoing.extend(outgoing);
2342 }
2343 }
2344 Err(e) => {
2345 log::warn!("Failed to deserialize user orders: {e}");
2346 }
2347 }
2348 }
2349 DeribitWsChannel::UserTrades => {
2350 let trades_result =
2352 serde_json::from_str::<Vec<DeribitUserTradeMsg>>(data.get())
2353 .or_else(|_| {
2354 serde_json::from_str::<DeribitUserTradeMsg>(data.get())
2355 .map(|trade| vec![trade])
2356 });
2357
2358 match trades_result {
2359 Ok(trades) => {
2360 log::debug!("Received {} user trade updates", trades.len());
2361 if self.account_id.is_none() {
2362 log::warn!("Cannot parse user trades: account_id not set");
2363 return Some(NautilusWsMessage::Raw(data.clone()));
2364 }
2365 let outgoing = self.route_user_trades(&trades, ts_init);
2366 if !outgoing.is_empty() {
2367 self.pending_outgoing.extend(outgoing);
2368 }
2369 }
2370 Err(e) => {
2371 log::warn!("Failed to deserialize user trades: {e}");
2372 }
2373 }
2374 }
2375 DeribitWsChannel::UserPortfolio => {
2376 match serde_json::from_str::<DeribitPortfolioMsg>(data.get()) {
2377 Ok(portfolio) => {
2378 if portfolio.equity.is_zero() && portfolio.balance.is_zero() {
2382 log::trace!(
2383 "Skipping zero-balance portfolio for {}",
2384 portfolio.currency
2385 );
2386 return None;
2387 }
2388
2389 let Some(account_id) = self.account_id else {
2391 log::warn!("Cannot parse portfolio: account_id not set");
2392 return None;
2393 };
2394
2395 match parse_portfolio_to_account_state(
2396 &portfolio, account_id, ts_init,
2397 ) {
2398 Ok(account_state) => {
2399 let currency_key = portfolio.currency.clone();
2401
2402 if let Some(last) =
2403 self.last_account_states.get(¤cy_key)
2404 && account_state.has_same_balances_and_margins(last)
2405 {
2406 log::trace!(
2407 "Skipping duplicate portfolio update for {}",
2408 portfolio.currency
2409 );
2410 return None;
2411 }
2412
2413 self.last_account_states
2414 .insert(currency_key, account_state.clone());
2415 return Some(NautilusWsMessage::AccountState(
2416 account_state,
2417 ));
2418 }
2419 Err(e) => {
2420 log::warn!(
2421 "Failed to parse portfolio to AccountState: {e}"
2422 );
2423 }
2424 }
2425 }
2426 Err(e) => {
2427 log::warn!("Failed to deserialize portfolio: {e}");
2428 }
2429 }
2430 }
2431 _ => {
2432 log::trace!("Unhandled channel: {channel}");
2434 return Some(NautilusWsMessage::Raw(data.clone()));
2435 }
2436 }
2437 } else {
2438 log::trace!("Unknown channel: {channel}");
2439 return Some(NautilusWsMessage::Raw(data.clone()));
2440 }
2441 None
2442 }
2443 DeribitWsMessage::Heartbeat(heartbeat) => {
2444 match heartbeat.heartbeat_type {
2445 DeribitHeartbeatType::TestRequest => {
2446 log::trace!(
2447 "Received heartbeat test_request - responding with public/test"
2448 );
2449
2450 if let Err(e) = self.handle_heartbeat_test_request().await {
2451 log::error!("Failed to respond to heartbeat test_request: {e}");
2452
2453 return Some(NautilusWsMessage::Error(DeribitWsError::Send(format!(
2455 "Heartbeat response failed: {e}"
2456 ))));
2457 }
2458 }
2459 DeribitHeartbeatType::Heartbeat => {
2460 log::trace!("Received heartbeat acknowledgment");
2461 }
2462 }
2463 None
2464 }
2465 DeribitWsMessage::Error(err) => {
2466 log::error!("Deribit error {}: {}", err.code, err.message);
2467 Some(NautilusWsMessage::Error(DeribitWsError::DeribitError {
2468 code: err.code,
2469 message: err.message,
2470 }))
2471 }
2472 DeribitWsMessage::Reconnected => Some(NautilusWsMessage::Reconnected),
2473 }
2474 }
2475
2476 async fn handle_book_message(
2477 &mut self,
2478 channel: &str,
2479 data: &str,
2480 ts_init: UnixNanos,
2481 ) -> Option<NautilusWsMessage> {
2482 let book_msg = match serde_json::from_str::<DeribitBookMsg>(data) {
2483 Ok(msg) => msg,
2484 Err(e) => {
2485 log::warn!("Failed to deserialize book message: {e}, channel: {channel}");
2486 return None;
2487 }
2488 };
2489
2490 if is_grouped_book_channel(channel) {
2491 return match self.handle_grouped_book(channel, &book_msg, ts_init).await {
2492 Ok(message) => message,
2493 Err(e) => {
2494 log::error!(
2495 "Failed to parse book snapshot for {} on {channel}: {e}, waiting for the next snapshot",
2496 book_msg.instrument_name
2497 );
2498 None
2499 }
2500 };
2501 }
2502
2503 let Some(instrument) = self.instruments_cache.get(&book_msg.instrument_name) else {
2504 log::warn!(
2505 "Book message received but instrument '{}' not found in cache (cache size: {})",
2506 book_msg.instrument_name,
2507 self.instruments_cache.len()
2508 );
2509 return None;
2510 };
2511
2512 let inst_name = book_msg.instrument_name.as_str();
2513 let awaiting_resync = self
2514 .pending_book_resync
2515 .iter()
2516 .any(|ch| is_instrument_book_channel(ch, inst_name));
2517
2518 if awaiting_resync && book_msg.msg_type == DeribitBookMsgType::Change {
2519 return None;
2520 }
2521
2522 if book_msg.msg_type == DeribitBookMsgType::Change
2523 && let Some(prev_id) = book_msg.prev_change_id
2524 && let Some(&last_id) = self.book_sequence.get(&book_msg.instrument_name)
2525 && prev_id != last_id
2526 {
2527 log::warn!(
2528 "Book sequence gap for {}: expected prev_change_id={}, was {} \
2529 - dropping delta, forcing resync",
2530 book_msg.instrument_name,
2531 last_id,
2532 prev_id
2533 );
2534 self.resync_book(book_msg.instrument_name).await;
2535 return None;
2536 }
2537
2538 match parse_book_msg(&book_msg, instrument, ts_init) {
2539 Ok(deltas) => {
2540 if awaiting_resync {
2541 self.pending_book_resync.retain(|ch| ch != channel);
2542 }
2543
2544 self.book_sequence
2545 .insert(book_msg.instrument_name, book_msg.change_id);
2546 Some(NautilusWsMessage::Deltas(deltas))
2547 }
2548 Err(e) => {
2549 log::error!(
2550 "Failed to parse book message for {}: {e}, forcing resync",
2551 book_msg.instrument_name
2552 );
2553 self.resync_book(book_msg.instrument_name).await;
2554 None
2555 }
2556 }
2557 }
2558
2559 async fn handle_grouped_book(
2560 &mut self,
2561 channel: &str,
2562 book_msg: &DeribitBookMsg<'_>,
2563 ts_init: UnixNanos,
2564 ) -> anyhow::Result<Option<NautilusWsMessage>> {
2565 let depth_subs = self.book_depth_subs.lock().await;
2566 let depth_refs = depth_subs.get_reference_count(channel);
2567 let deltas_subscribed = self.subscriptions_state.get_reference_count(channel) > depth_refs;
2568 if depth_refs == 0 && !deltas_subscribed {
2569 return Ok(None);
2570 }
2571
2572 let Some(instrument) = self.instruments_cache.get(&book_msg.instrument_name) else {
2573 log::warn!(
2574 "Book snapshot instrument '{}' not found in cache",
2575 book_msg.instrument_name
2576 );
2577 return Ok(None);
2578 };
2579
2580 let depth = if depth_refs > 0 {
2581 Some(parse_book_depth(book_msg, instrument, ts_init)?)
2582 } else {
2583 None
2584 };
2585
2586 let deltas = if deltas_subscribed {
2587 Some(parse_book_msg(book_msg, instrument, ts_init)?)
2588 } else {
2589 None
2590 };
2591
2592 if let Some(depth) = depth {
2593 if let Some(deltas) = deltas {
2594 self.pending_outgoing
2595 .push_back(NautilusWsMessage::Deltas(deltas));
2596 }
2597
2598 return Ok(Some(NautilusWsMessage::Data(vec![Data::BookDepth(
2599 Box::new(depth),
2600 )])));
2601 }
2602
2603 Ok(deltas.map(NautilusWsMessage::Deltas))
2604 }
2605
2606 async fn resync_book(&mut self, instrument_name: Ustr) {
2610 self.book_sequence.remove(&instrument_name);
2611
2612 let book_channels: Vec<String> = self
2613 .subscriptions_state
2614 .all_topics()
2615 .into_iter()
2616 .filter(|t| {
2617 is_instrument_book_channel(t, &instrument_name) && !is_grouped_book_channel(t)
2618 })
2619 .collect();
2620
2621 if book_channels.is_empty() {
2622 return;
2623 }
2624
2625 for ch in &book_channels {
2626 self.subscriptions_state.mark_failure(ch);
2627
2628 if !self.pending_book_resync.contains(ch) {
2629 self.pending_book_resync.push(ch.clone());
2630 }
2631 }
2632
2633 let _ = self.handle_unsubscribe(book_channels);
2635 }
2636
2637 fn order_signature(order: &DeribitOrderMsg) -> OrderSignature {
2638 (order.amount, order.price, order.trigger_price)
2639 }
2640
2641 fn find_order_context(
2642 &self,
2643 venue_order_id: VenueOrderId,
2644 client_order_id: Option<ClientOrderId>,
2645 ) -> Option<OrderContext> {
2646 self.order_contexts
2647 .get(&venue_order_id)
2648 .or_else(|| self.terminal_order_contexts.get(&venue_order_id))
2649 .cloned()
2650 .or_else(|| {
2651 client_order_id.and_then(|client_order_id| {
2652 self.submitted_order_contexts.get(&client_order_id).cloned()
2653 })
2654 })
2655 }
2656
2657 fn bind_order_context(&mut self, venue_order_id: VenueOrderId, context: OrderContext) {
2658 self.submitted_order_contexts
2659 .remove(&context.client_order_id);
2660 self.terminal_order_contexts.remove(&venue_order_id);
2661 self.order_contexts.insert(venue_order_id, context);
2662 }
2663
2664 fn finish_order_context(&mut self, venue_order_id: VenueOrderId, context: &OrderContext) {
2665 self.order_contexts.remove(&venue_order_id);
2666 self.submitted_order_contexts
2667 .remove(&context.client_order_id);
2668 self.terminal_order_contexts
2669 .insert(venue_order_id, context.clone());
2670 }
2671
2672 fn route_user_trades(
2673 &mut self,
2674 trades: &[DeribitUserTradeMsg],
2675 ts_init: UnixNanos,
2676 ) -> Vec<NautilusWsMessage> {
2677 let Some(account_id) = self.account_id else {
2678 log::warn!("Cannot parse user trades: account_id not set");
2679 return Vec::new();
2680 };
2681
2682 let mut outgoing = Vec::with_capacity(trades.len() + 1);
2683 let mut reports = Vec::new();
2684
2685 for trade in trades {
2686 let instrument_name = trade.instrument_name;
2687 let Some((report, quote_currency)) =
2688 self.instruments_cache
2689 .get(&instrument_name)
2690 .map(|instrument| {
2691 (
2692 parse_user_trade_msg(trade, instrument, account_id, ts_init),
2693 instrument.quote_currency(),
2694 )
2695 })
2696 else {
2697 log::warn!("Instrument {instrument_name} not found in cache");
2698 continue;
2699 };
2700
2701 let report = match report {
2702 Ok(report) => report,
2703 Err(e) => {
2704 log::warn!("Failed to parse trade {}: {e}", trade.trade_id);
2705 continue;
2706 }
2707 };
2708 let venue_order_id = report.venue_order_id;
2709 let was_terminal = self.terminal_order_contexts.contains_key(&venue_order_id);
2710 let Some(mut context) = self.find_order_context(venue_order_id, report.client_order_id)
2711 else {
2712 log::debug!(
2713 "Parsed external fill report: {} @ {}",
2714 report.trade_id,
2715 report.last_px
2716 );
2717 reports.push(report);
2718 continue;
2719 };
2720
2721 if !context.accepted {
2722 outgoing.push(NautilusWsMessage::OrderAccepted(OrderAccepted::new(
2723 context.trader_id,
2724 context.strategy_id,
2725 context.instrument_id,
2726 context.client_order_id,
2727 venue_order_id,
2728 account_id,
2729 UUID4::new(),
2730 report.ts_event,
2731 report.ts_init,
2732 false,
2733 )));
2734 context.accepted = true;
2735 }
2736
2737 log::debug!(
2738 "Parsed tracked fill event: {} @ {}",
2739 report.trade_id,
2740 report.last_px
2741 );
2742 outgoing.push(NautilusWsMessage::OrderFilled(OrderFilled::new(
2743 context.trader_id,
2744 context.strategy_id,
2745 context.instrument_id,
2746 context.client_order_id,
2747 venue_order_id,
2748 account_id,
2749 report.trade_id,
2750 context.order_side,
2751 context.order_type,
2752 report.last_qty,
2753 report.last_px,
2754 quote_currency,
2755 report.liquidity_side,
2756 UUID4::new(),
2757 report.ts_event,
2758 report.ts_init,
2759 false,
2760 report.venue_position_id,
2761 Some(report.commission),
2762 None,
2763 )));
2764
2765 if was_terminal || trade.state == "filled" {
2766 self.finish_order_context(venue_order_id, &context);
2767 } else {
2768 self.bind_order_context(venue_order_id, context);
2769 }
2770 }
2771
2772 outgoing.retain(|message| match message {
2774 NautilusWsMessage::OrderFilled(fill) => self
2775 .emitted_trades
2776 .insert((fill.venue_order_id, fill.trade_id)),
2777 _ => true,
2778 });
2779
2780 reports.retain(|report| {
2781 self.emitted_trades
2782 .insert((report.venue_order_id, report.trade_id))
2783 });
2784
2785 if !reports.is_empty() {
2786 outgoing.push(NautilusWsMessage::FillReports(reports));
2787 }
2788 outgoing
2789 }
2790
2791 pub async fn next(&mut self) -> Option<NautilusWsMessage> {
2797 loop {
2798 if self.signal.load(Ordering::Relaxed) {
2799 self.subscription_send = None;
2800 self.process_command(HandlerCommand::Disconnect).await;
2801 return None;
2802 }
2803
2804 if let Some(msg) = self.pending_outgoing.pop_front() {
2805 match msg {
2806 NautilusWsMessage::Reconnected
2807 | NautilusWsMessage::Authenticated(_)
2808 | NautilusWsMessage::AuthenticationFailed(_) => {
2809 return Some(msg);
2810 }
2811 _ => {
2812 let _ = self.out_tx.send(msg);
2813 continue;
2814 }
2815 }
2816 }
2817
2818 if self.subscription_send.is_none()
2819 && let Some((request_id, payload)) = self.subscription_requests.pop_front()
2820 {
2821 let send = self.send_secret_with_retry(
2822 payload.into(),
2823 Some(&[Ustr::from(DERIBIT_WS_SUBSCRIPTION_KEY)]),
2824 );
2825 self.subscription_send = Some(Box::pin(async move { (request_id, send.await) }));
2826 }
2827
2828 tokio::select! {
2829 (request_id, result) = async {
2830 match &mut self.subscription_send {
2831 Some(send) => send.await,
2832 None => std::future::pending().await,
2833 }
2834 } => {
2835 self.subscription_send = None;
2836
2837 if let Err(e) = result {
2838 if matches!(
2839 self.pending_requests.remove(&request_id),
2840 Some(PendingRequestType::Subscribe { .. })
2841 ) {
2842 self.subscribe_errors.lock().push(e.to_string());
2843 }
2844 log::error!("Subscription request send failed: request_id={request_id}, {e}");
2845 }
2846 }
2847 Some(cmd) = async {
2849 match self.pending_command.take() {
2850 Some(cmd) => Some(cmd),
2851 None => self.cmd_rx.recv().await,
2852 }
2853 }, if self.subscription_send.is_none() && self.subscription_requests.is_empty() => {
2854 let cmd = match cmd {
2855 HandlerCommand::Subscribe { channels } => {
2856 let channels = self.coalesce_subscriptions(channels);
2857 HandlerCommand::Subscribe { channels }
2858 }
2859 cmd => cmd,
2860 };
2861 self.process_command(cmd).await;
2862 }
2863 Some(msg) = self.raw_rx.recv() => {
2865 match msg {
2866 Message::Text(text) => {
2867 if let Some(nautilus_msg) = self.process_raw_message(&text).await {
2868 match &nautilus_msg {
2870 NautilusWsMessage::Data(_)
2871 | NautilusWsMessage::Deltas(_)
2872 | NautilusWsMessage::Instrument(_)
2873 | NautilusWsMessage::InstrumentStatus(_)
2874 | NautilusWsMessage::OptionGreeks(_)
2875 | NautilusWsMessage::Raw(_)
2876 | NautilusWsMessage::Error(_) => {
2877 let _ = self.out_tx.send(nautilus_msg);
2878 }
2879 NautilusWsMessage::FundingRates(rates) => {
2880 let msg_to_send =
2881 NautilusWsMessage::FundingRates(rates.clone());
2882
2883 if let Err(e) = self.out_tx.send(msg_to_send) {
2884 log::error!("Failed to send funding rates: {e}");
2885 }
2886 }
2887 NautilusWsMessage::OrderStatusReports(_)
2888 | NautilusWsMessage::FillReports(_)
2889 | NautilusWsMessage::OrderFilled(_)
2890 | NautilusWsMessage::OrderAccepted(_)
2891 | NautilusWsMessage::OrderCanceled(_)
2892 | NautilusWsMessage::OrderExpired(_)
2893 | NautilusWsMessage::OrderUpdated(_)
2894 | NautilusWsMessage::OrderRejected(_)
2895 | NautilusWsMessage::OrderCancelRejected(_)
2896 | NautilusWsMessage::OrderModifyRejected(_)
2897 | NautilusWsMessage::AccountState(_) => {
2898 let _ = self.out_tx.send(nautilus_msg);
2899 }
2900 NautilusWsMessage::Reconnected
2902 | NautilusWsMessage::Authenticated(_)
2903 | NautilusWsMessage::AuthenticationFailed(_) => {
2904 return Some(nautilus_msg);
2905 }
2906 }
2907 }
2908 }
2909 Message::Ping(data) => {
2910 if let Some(client) = &self.inner {
2912 let _ = client.send_pong(data.to_vec()).await;
2913 }
2914 }
2915 Message::Close(_) => {
2916 log::debug!("Received close frame");
2917 }
2918 _ => {}
2919 }
2920 }
2921 () = tokio::time::sleep(tokio::time::Duration::from_millis(100)) => {
2923 }
2924 }
2925 }
2926 }
2927}
2928
2929type SubscriptionSend = BoxFuture<'static, (u64, Result<(), DeribitWsError>)>;
2930
2931fn subscription_batches(
2932 channels: Vec<String>,
2933 method: &DeribitWsMethod,
2934) -> Result<Vec<Vec<String>>, DeribitWsError> {
2935 let envelope = DeribitJsonRpcRequest::new(
2936 u64::MAX,
2937 method.as_method_str(),
2938 DeribitSubscribeParams {
2939 channels: Vec::new(),
2940 },
2941 );
2942
2943 let overhead = serde_json::to_string(&envelope)
2944 .map_err(|e| DeribitWsError::Json(e.to_string()))?
2945 .len();
2946 let mut batches = Vec::new();
2947 let mut batch = Vec::new();
2948 let mut bytes = overhead;
2949
2950 for channel in channels {
2951 let encoded =
2952 serde_json::to_string(&channel).map_err(|e| DeribitWsError::Json(e.to_string()))?;
2953
2954 if overhead + encoded.len() > SUBSCRIPTION_REQUEST_MAX_BYTES {
2955 return Err(DeribitWsError::Subscribe(
2956 "Channel exceeds the WebSocket request byte limit".to_string(),
2957 ));
2958 }
2959
2960 let separator = usize::from(!batch.is_empty());
2961 if bytes + separator + encoded.len() > SUBSCRIPTION_REQUEST_MAX_BYTES {
2962 batches.push(std::mem::take(&mut batch));
2963 bytes = overhead;
2964 }
2965
2966 bytes += usize::from(!batch.is_empty()) + encoded.len();
2967 batch.push(channel);
2968 }
2969
2970 if !batch.is_empty() {
2971 batches.push(batch);
2972 }
2973
2974 Ok(batches)
2975}
2976
2977fn is_instrument_book_channel(channel: &str, instrument_name: &str) -> bool {
2978 channel.starts_with("book.")
2979 && channel
2980 .split('.')
2981 .nth(1)
2982 .is_some_and(|s| s == instrument_name)
2983}
2984
2985fn is_grouped_book_channel(channel: &str) -> bool {
2987 channel.starts_with("book.") && channel.split('.').count() == 5
2988}
2989
2990#[cfg(test)]
2991mod tests {
2992 use std::time::Duration;
2993
2994 use futures_util::{SinkExt, StreamExt};
2995 use nautilus_model::{
2996 enums::{BookType, LiquiditySide},
2997 instruments::Instrument,
2998 orderbook::OrderBook,
2999 types::{Money, Price, Quantity},
3000 };
3001 use nautilus_network::{
3002 ratelimiter::quota::Quota,
3003 websocket::{WebSocketConfig, channel_message_handler},
3004 };
3005 use rstest::rstest;
3006
3007 use super::*;
3008 use crate::{
3009 common::{parse::parse_deribit_instrument_any, testing::load_test_json},
3010 http::models::{DeribitInstrument, DeribitJsonRpcResponse},
3011 };
3012
3013 fn routing_test_handler() -> DeribitWsFeedHandler {
3014 let signal = Arc::new(AtomicBool::new(false));
3015 let (_cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
3016 let (_raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
3017 let (out_tx, _out_rx) = tokio::sync::mpsc::unbounded_channel();
3018 let mut handler = DeribitWsFeedHandler::new(
3019 signal,
3020 cmd_rx,
3021 raw_rx,
3022 out_tx,
3023 AuthTracker::new(),
3024 SubscriptionState::new('.'),
3025 Arc::new(AtomicSet::new()),
3026 Arc::new(AtomicSet::new()),
3027 Arc::new(AtomicSet::new()),
3028 Some(AccountId::from("DERIBIT-001")),
3029 true,
3030 Arc::new(Mutex::new(Vec::new())),
3031 );
3032 let json = load_test_json("http_get_instruments.json");
3033 let response: DeribitJsonRpcResponse<Vec<DeribitInstrument>> =
3034 serde_json::from_str(&json).unwrap();
3035 let instrument = parse_deribit_instrument_any(
3036 &response.result.unwrap()[0],
3037 UnixNanos::default(),
3038 UnixNanos::default(),
3039 )
3040 .unwrap()
3041 .unwrap();
3042 handler
3043 .instruments_cache
3044 .insert(instrument.raw_symbol().inner(), instrument);
3045 handler
3046 }
3047
3048 fn order_data(order_state: &str, replaced: bool) -> serde_json::Value {
3049 serde_json::json!({
3050 "order_id": "ETH-584830574",
3051 "label": "O-19700101-000000-001-001-1",
3052 "instrument_name": "BTC-PERPETUAL",
3053 "direction": "buy",
3054 "order_type": "market",
3055 "order_state": order_state,
3056 "replaced": replaced,
3057 "price": 203.8,
3058 "amount": 2.0,
3059 "filled_amount": if order_state == "filled" { 2.0 } else { 0.0 },
3060 "average_price": 203.8,
3061 "creation_timestamp": 1_590_480_712_700_u64,
3062 "last_update_timestamp": 1_590_480_712_800_u64,
3063 "time_in_force": "good_til_cancelled",
3064 "commission": 0.00073602,
3065 "post_only": false,
3066 "reduce_only": false,
3067 "trigger_price": null,
3068 "trigger": null,
3069 "max_show": null,
3070 "api": true,
3071 "reject_reason": null,
3072 "cancel_reason": null
3073 })
3074 }
3075
3076 fn trade_data() -> serde_json::Value {
3077 serde_json::json!({
3078 "trade_id": "ETH-2696068",
3079 "order_id": "ETH-584830574",
3080 "instrument_name": "BTC-PERPETUAL",
3081 "direction": "buy",
3082 "price": 203.8,
3083 "amount": 2.0,
3084 "fee": 0.00073602,
3085 "fee_currency": "USDT",
3086 "timestamp": 1_590_480_712_800_u64,
3087 "trade_seq": 1_966_042_u64,
3088 "liquidity": "T",
3089 "order_type": "market",
3090 "index_price": 203.89,
3091 "mark_price": 203.78,
3092 "tick_direction": 3,
3093 "state": "filled",
3094 "label": "O-19700101-000000-001-001-1",
3095 "reduce_only": false,
3096 "post_only": false,
3097 "liquidation": null,
3098 "profit_loss": null
3099 })
3100 }
3101
3102 fn subscription(channel: &str, data: &serde_json::Value) -> String {
3103 serde_json::json!({
3104 "jsonrpc": "2.0",
3105 "method": "subscription",
3106 "params": { "channel": channel, "data": data }
3107 })
3108 .to_string()
3109 }
3110
3111 #[rstest]
3112 #[tokio::test]
3113 async fn tracked_fast_fill_synthesizes_accepted_before_filled() {
3114 let mut handler = routing_test_handler();
3115 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3116 let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
3117 handler.submitted_order_contexts.insert(
3118 client_order_id,
3119 OrderContext {
3120 client_order_id,
3121 trader_id: TraderId::from("TRADER-001"),
3122 strategy_id: StrategyId::from("S-001"),
3123 instrument_id,
3124 order_side: OrderSide::Buy,
3125 order_type: OrderType::Market,
3126 accepted: false,
3127 last_order_signature: None,
3128 },
3129 );
3130 handler.pending_requests.insert(
3131 42,
3132 PendingRequestType::Buy {
3133 client_order_id,
3134 trader_id: TraderId::from("TRADER-001"),
3135 strategy_id: StrategyId::from("S-001"),
3136 instrument_id,
3137 order_side: OrderSide::Buy,
3138 order_type: OrderType::Market,
3139 },
3140 );
3141 let response = serde_json::json!({
3142 "jsonrpc": "2.0",
3143 "id": 42,
3144 "result": { "order": order_data("filled", false), "trades": [] }
3145 });
3146
3147 handler.process_raw_message(&response.to_string()).await;
3148 assert!(handler.pending_outgoing.is_empty());
3149 assert!(
3150 !handler
3151 .terminal_order_contexts
3152 .get(&VenueOrderId::from("ETH-584830574"))
3153 .unwrap()
3154 .accepted
3155 );
3156
3157 handler
3158 .process_raw_message(&subscription(
3159 "user.trades.any.any.raw",
3160 &serde_json::json!([trade_data()]),
3161 ))
3162 .await;
3163
3164 assert!(matches!(
3165 handler.pending_outgoing.pop_front().unwrap(),
3166 NautilusWsMessage::OrderAccepted(event)
3167 if event.client_order_id == client_order_id
3168 ));
3169 assert!(matches!(
3170 handler.pending_outgoing.pop_front().unwrap(),
3171 NautilusWsMessage::OrderFilled(event)
3172 if event.client_order_id == client_order_id
3173 && event.trade_id.to_string() == "ETH-2696068"
3174 && event.order_side == OrderSide::Buy
3175 && event.order_type == OrderType::Market
3176 && event.last_qty.to_string() == "2"
3177 && event.last_px.to_string() == "203.8"
3178 && event.liquidity_side == LiquiditySide::Taker
3179 && event.commission == Some(Money::from("0.00073602 USDT"))
3180 ));
3181 assert!(handler.pending_outgoing.is_empty());
3182
3183 handler
3184 .process_raw_message(&subscription(
3185 "user.trades.any.any.raw",
3186 &serde_json::json!([trade_data()]),
3187 ))
3188 .await;
3189
3190 assert!(handler.pending_outgoing.is_empty());
3191 }
3192
3193 #[rstest]
3194 #[case::response_first(false)]
3195 #[case::subscription_first(true)]
3196 #[tokio::test]
3197 async fn submit_response_trades_use_tracked_event_path(
3198 #[case] subscription_first: bool,
3199 #[values("filled", "open")] trade_state: &str,
3200 ) {
3201 let mut handler = routing_test_handler();
3202 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3203 let context = OrderContext {
3204 client_order_id,
3205 trader_id: TraderId::from("TRADER-001"),
3206 strategy_id: StrategyId::from("S-001"),
3207 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3208 order_side: OrderSide::Buy,
3209 order_type: OrderType::Market,
3210 accepted: false,
3211 last_order_signature: None,
3212 };
3213 handler
3214 .submitted_order_contexts
3215 .insert(client_order_id, context.clone());
3216 handler.pending_requests.insert(
3217 42,
3218 PendingRequestType::Buy {
3219 client_order_id,
3220 trader_id: context.trader_id,
3221 strategy_id: context.strategy_id,
3222 instrument_id: context.instrument_id,
3223 order_side: context.order_side,
3224 order_type: context.order_type,
3225 },
3226 );
3227
3228 let mut trade = trade_data();
3229 trade["state"] = serde_json::json!(trade_state);
3230 trade["amount"] = serde_json::json!(if trade_state == "filled" { 2 } else { 1 });
3231 let mut order = order_data(trade_state, false);
3232 order["filled_amount"] = trade["amount"].clone();
3233 let response = serde_json::json!({
3234 "jsonrpc": "2.0",
3235 "id": 42,
3236 "result": { "order": order, "trades": [trade.clone()] }
3237 });
3238 let trade_update = subscription("user.trades.any.any.raw", &serde_json::json!([trade]));
3239
3240 let (raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
3241 let (out_tx, mut out_rx) = tokio::sync::mpsc::unbounded_channel();
3242 handler.raw_rx = raw_rx;
3243 handler.out_tx = out_tx;
3244 let response = response.to_string();
3245 let messages = if subscription_first {
3246 [trade_update, response]
3247 } else {
3248 [response, trade_update]
3249 };
3250
3251 for message in messages {
3252 raw_tx.send(Message::Text(message.into())).unwrap();
3253 }
3254
3255 raw_tx.send(Message::Text(RECONNECTED.into())).unwrap();
3256 let reconnected = tokio::time::timeout(std::time::Duration::from_secs(2), handler.next())
3257 .await
3258 .unwrap();
3259
3260 assert!(matches!(reconnected, Some(NautilusWsMessage::Reconnected)));
3261 assert!(matches!(
3262 out_rx.try_recv().unwrap(),
3263 NautilusWsMessage::OrderAccepted(event)
3264 if event.client_order_id == client_order_id
3265 ));
3266 assert!(matches!(
3267 out_rx.try_recv().unwrap(),
3268 NautilusWsMessage::OrderFilled(event)
3269 if event.client_order_id == client_order_id
3270 && event.trade_id.to_string() == "ETH-2696068"
3271 && event.last_qty.to_string() == if trade_state == "filled" { "2" } else { "1" }
3272 ));
3273 assert!(out_rx.try_recv().is_err());
3274 assert!(handler.pending_outgoing.is_empty());
3275 let venue_order_id = VenueOrderId::from("ETH-584830574");
3276 assert_eq!(
3277 handler
3278 .terminal_order_contexts
3279 .contains_key(&venue_order_id),
3280 trade_state == "filled",
3281 );
3282 assert_eq!(
3283 handler.order_contexts.contains_key(&venue_order_id),
3284 trade_state == "open",
3285 );
3286 }
3287
3288 #[rstest]
3289 #[case::tracked(true)]
3290 #[case::external(false)]
3291 #[tokio::test]
3292 async fn trade_replays_after_reconnect_preserve_new_fills(#[case] tracked: bool) {
3293 let mut handler = routing_test_handler();
3294 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3295 let venue_order_id = VenueOrderId::from("ETH-584830574");
3296
3297 if tracked {
3298 handler.order_contexts.insert(
3299 venue_order_id,
3300 OrderContext {
3301 client_order_id,
3302 trader_id: TraderId::from("TRADER-001"),
3303 strategy_id: StrategyId::from("S-001"),
3304 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3305 order_side: OrderSide::Buy,
3306 order_type: OrderType::Market,
3307 accepted: true,
3308 last_order_signature: None,
3309 },
3310 );
3311 }
3312
3313 let mut first_trade = trade_data();
3314 first_trade["amount"] = serde_json::json!(1);
3315 let mut second_trade = first_trade.clone();
3316 second_trade["trade_id"] = serde_json::json!("ETH-2696069");
3317 second_trade["price"] = serde_json::json!(204.1);
3318 second_trade["fee"] = serde_json::json!(0.00083602);
3319 let first_update = subscription(
3320 "user.trades.any.any.raw",
3321 &serde_json::json!([first_trade.clone(), first_trade.clone()]),
3322 );
3323 let replay_update = subscription(
3324 "user.trades.any.any.raw",
3325 &serde_json::json!([first_trade, second_trade.clone(), second_trade]),
3326 );
3327
3328 handler.process_raw_message(&first_update).await;
3329 let first = handler.pending_outgoing.pop_front().unwrap();
3330 assert!(handler.pending_outgoing.is_empty());
3331 let reconnected = handler.process_raw_message(RECONNECTED).await;
3332 handler.process_raw_message(&replay_update).await;
3333 let second = handler.pending_outgoing.pop_front().unwrap();
3334
3335 let fills = [first, second].map(|message| match message {
3336 NautilusWsMessage::OrderFilled(fill) => {
3337 assert!(tracked);
3338 assert_eq!(fill.client_order_id, client_order_id);
3339 (
3340 fill.trade_id,
3341 fill.last_qty,
3342 fill.last_px,
3343 fill.commission.unwrap(),
3344 )
3345 }
3346 NautilusWsMessage::FillReports(mut reports) => {
3347 assert!(!tracked);
3348 assert_eq!(reports.len(), 1);
3349 let report = reports.pop().unwrap();
3350 (
3351 report.trade_id,
3352 report.last_qty,
3353 report.last_px,
3354 report.commission,
3355 )
3356 }
3357 message => panic!("unexpected fill message: {message:?}"),
3358 });
3359
3360 assert!(matches!(reconnected, Some(NautilusWsMessage::Reconnected)));
3361 assert!(handler.pending_outgoing.is_empty());
3362 assert_eq!(
3363 fills,
3364 [
3365 (
3366 TradeId::from("ETH-2696068"),
3367 Quantity::from(1),
3368 Price::from("203.8"),
3369 Money::from("0.00073602 USDT")
3370 ),
3371 (
3372 TradeId::from("ETH-2696069"),
3373 Quantity::from(1),
3374 Price::from("204.1"),
3375 Money::from("0.00083602 USDT")
3376 ),
3377 ],
3378 );
3379 assert_eq!(
3380 handler
3381 .terminal_order_contexts
3382 .contains_key(&venue_order_id),
3383 tracked
3384 );
3385 assert!(!handler.order_contexts.contains_key(&venue_order_id));
3386 }
3387
3388 #[rstest]
3389 fn trade_dedup_is_scoped_to_venue_order() {
3390 let mut handler = routing_test_handler();
3391 let first_trade = trade_data();
3392 let mut second_trade = first_trade.clone();
3393 second_trade["order_id"] = serde_json::json!("ETH-584830575");
3394 let trades: Vec<DeribitUserTradeMsg> = serde_json::from_str(
3395 &serde_json::json!([first_trade.clone(), first_trade, second_trade]).to_string(),
3396 )
3397 .unwrap();
3398
3399 let messages = handler.route_user_trades(&trades, UnixNanos::default());
3400
3401 assert_eq!(messages.len(), 1);
3402
3403 let NautilusWsMessage::FillReports(reports) = &messages[0] else {
3404 panic!("expected external fill reports");
3405 };
3406
3407 assert_eq!(
3408 reports
3409 .iter()
3410 .map(|report| (report.venue_order_id, report.trade_id))
3411 .collect::<Vec<_>>(),
3412 vec![
3413 (
3414 VenueOrderId::from("ETH-584830574"),
3415 TradeId::from("ETH-2696068")
3416 ),
3417 (
3418 VenueOrderId::from("ETH-584830575"),
3419 TradeId::from("ETH-2696068")
3420 ),
3421 ],
3422 );
3423 }
3424
3425 #[rstest]
3426 #[tokio::test]
3427 async fn reconnect_preserves_submit_identity_before_response() {
3428 let mut handler = routing_test_handler();
3429 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3430 handler.submitted_order_contexts.insert(
3431 client_order_id,
3432 OrderContext {
3433 client_order_id,
3434 trader_id: TraderId::from("TRADER-001"),
3435 strategy_id: StrategyId::from("S-001"),
3436 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3437 order_side: OrderSide::Buy,
3438 order_type: OrderType::Market,
3439 accepted: false,
3440 last_order_signature: None,
3441 },
3442 );
3443 handler
3444 .pending_requests
3445 .insert(42, PendingRequestType::Test);
3446
3447 handler.clear_state();
3448 handler
3449 .process_raw_message(&subscription(
3450 "user.orders.any.any.raw",
3451 &serde_json::json!([order_data("open", false)]),
3452 ))
3453 .await;
3454
3455 assert!(handler.pending_requests.is_empty());
3456 assert!(matches!(
3457 handler.pending_outgoing.pop_front().unwrap(),
3458 NautilusWsMessage::OrderAccepted(event)
3459 if event.client_order_id == client_order_id
3460 ));
3461 assert!(
3462 handler
3463 .order_contexts
3464 .get(&VenueOrderId::from("ETH-584830574"))
3465 .unwrap()
3466 .accepted
3467 );
3468 }
3469
3470 #[rstest]
3471 #[case("cancelled")]
3472 #[case("expired")]
3473 #[tokio::test]
3474 async fn tracked_terminal_order_synthesizes_accepted_first(#[case] order_state: &str) {
3475 let mut handler = routing_test_handler();
3476 let venue_order_id = VenueOrderId::from("ETH-584830574");
3477 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3478 handler.order_contexts.insert(
3479 venue_order_id,
3480 OrderContext {
3481 client_order_id,
3482 trader_id: TraderId::from("TRADER-001"),
3483 strategy_id: StrategyId::from("S-001"),
3484 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3485 order_side: OrderSide::Buy,
3486 order_type: OrderType::Market,
3487 accepted: false,
3488 last_order_signature: None,
3489 },
3490 );
3491
3492 handler
3493 .process_raw_message(&subscription(
3494 "user.orders.any.any.raw",
3495 &serde_json::json!([order_data(order_state, false)]),
3496 ))
3497 .await;
3498
3499 let accepted = handler.pending_outgoing.pop_front().unwrap();
3500 let terminal = handler.pending_outgoing.pop_front().unwrap();
3501 assert!(matches!(
3502 accepted,
3503 NautilusWsMessage::OrderAccepted(event)
3504 if event.client_order_id == client_order_id
3505 ));
3506 assert!(
3507 matches!(
3508 (order_state, terminal),
3509 ("cancelled", NautilusWsMessage::OrderCanceled(_))
3510 | ("expired", NautilusWsMessage::OrderExpired(_))
3511 ),
3512 "unexpected terminal message for {order_state}",
3513 );
3514 assert!(handler.pending_outgoing.is_empty());
3515 assert!(!handler.order_contexts.contains_key(&venue_order_id));
3516 }
3517
3518 #[rstest]
3519 #[case("filled")]
3520 #[case("open")]
3521 #[tokio::test]
3522 async fn late_fill_after_cancel_stays_on_tracked_event_path(#[case] trade_state: &str) {
3523 let mut handler = routing_test_handler();
3524 let venue_order_id = VenueOrderId::from("ETH-584830574");
3525 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3526 handler.order_contexts.insert(
3527 venue_order_id,
3528 OrderContext {
3529 client_order_id,
3530 trader_id: TraderId::from("TRADER-001"),
3531 strategy_id: StrategyId::from("S-001"),
3532 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3533 order_side: OrderSide::Buy,
3534 order_type: OrderType::Market,
3535 accepted: true,
3536 last_order_signature: None,
3537 },
3538 );
3539
3540 handler
3541 .process_raw_message(&subscription(
3542 "user.orders.any.any.raw",
3543 &serde_json::json!([order_data("cancelled", false)]),
3544 ))
3545 .await;
3546 let mut trade = trade_data();
3547 trade["state"] = serde_json::Value::String(trade_state.to_string());
3548 handler
3549 .process_raw_message(&subscription(
3550 "user.trades.any.any.raw",
3551 &serde_json::json!([trade]),
3552 ))
3553 .await;
3554
3555 assert!(matches!(
3556 handler.pending_outgoing.pop_front().unwrap(),
3557 NautilusWsMessage::OrderCanceled(event)
3558 if event.client_order_id == client_order_id
3559 ));
3560 assert!(matches!(
3561 handler.pending_outgoing.pop_front().unwrap(),
3562 NautilusWsMessage::OrderFilled(event)
3563 if event.client_order_id == client_order_id
3564 ));
3565 assert!(handler.pending_outgoing.is_empty());
3566 assert!(!handler.order_contexts.contains_key(&venue_order_id));
3567 assert!(
3568 handler
3569 .terminal_order_contexts
3570 .contains_key(&venue_order_id)
3571 );
3572 }
3573
3574 #[rstest]
3575 #[tokio::test]
3576 async fn tracked_order_uses_stored_client_id_when_label_changes() {
3577 let mut handler = routing_test_handler();
3578 let venue_order_id = VenueOrderId::from("ETH-584830574");
3579 let client_order_id = ClientOrderId::from("ORIGINAL-CLIENT-ID");
3580 handler.order_contexts.insert(
3581 venue_order_id,
3582 OrderContext {
3583 client_order_id,
3584 trader_id: TraderId::from("TRADER-001"),
3585 strategy_id: StrategyId::from("S-001"),
3586 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3587 order_side: OrderSide::Buy,
3588 order_type: OrderType::Market,
3589 accepted: false,
3590 last_order_signature: None,
3591 },
3592 );
3593 let mut order = order_data("open", false);
3594 order["label"] = serde_json::Value::Null;
3595
3596 handler
3597 .process_raw_message(&subscription(
3598 "user.orders.any.any.raw",
3599 &serde_json::json!([order]),
3600 ))
3601 .await;
3602
3603 assert!(matches!(
3604 handler.pending_outgoing.pop_front().unwrap(),
3605 NautilusWsMessage::OrderAccepted(event)
3606 if event.client_order_id == client_order_id
3607 ));
3608 assert!(handler.pending_outgoing.is_empty());
3609 }
3610
3611 #[rstest]
3612 #[tokio::test]
3613 async fn subscription_accept_before_submit_response_is_not_duplicated() {
3614 let mut handler = routing_test_handler();
3615 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3616 handler.submitted_order_contexts.insert(
3617 client_order_id,
3618 OrderContext {
3619 client_order_id,
3620 trader_id: TraderId::from("TRADER-001"),
3621 strategy_id: StrategyId::from("S-001"),
3622 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3623 order_side: OrderSide::Buy,
3624 order_type: OrderType::Market,
3625 accepted: false,
3626 last_order_signature: None,
3627 },
3628 );
3629 handler.pending_requests.insert(
3630 42,
3631 PendingRequestType::Buy {
3632 client_order_id,
3633 trader_id: TraderId::from("TRADER-001"),
3634 strategy_id: StrategyId::from("S-001"),
3635 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3636 order_side: OrderSide::Buy,
3637 order_type: OrderType::Market,
3638 },
3639 );
3640
3641 handler
3642 .process_raw_message(&subscription(
3643 "user.orders.any.any.raw",
3644 &serde_json::json!([order_data("open", false)]),
3645 ))
3646 .await;
3647 let accepted = handler.pending_outgoing.pop_front().unwrap();
3648 let response = serde_json::json!({
3649 "jsonrpc": "2.0",
3650 "id": 42,
3651 "result": { "order": order_data("open", false), "trades": [] }
3652 });
3653 let duplicate = handler.process_raw_message(&response.to_string()).await;
3654 handler.clear_state();
3655 handler
3656 .process_raw_message(&subscription(
3657 "user.orders.any.any.raw",
3658 &serde_json::json!([order_data("open", false)]),
3659 ))
3660 .await;
3661
3662 assert!(matches!(accepted, NautilusWsMessage::OrderAccepted(_)));
3663 assert!(duplicate.is_none());
3664 assert!(handler.pending_outgoing.is_empty());
3665 assert!(
3666 handler
3667 .order_contexts
3668 .get(&VenueOrderId::from("ETH-584830574"))
3669 .unwrap()
3670 .accepted
3671 );
3672 }
3673
3674 #[rstest]
3675 #[tokio::test]
3676 async fn untracked_live_order_and_fill_use_report_paths() {
3677 let mut handler = routing_test_handler();
3678
3679 handler
3680 .process_raw_message(&subscription(
3681 "user.orders.any.any.raw",
3682 &serde_json::json!([order_data("open", false)]),
3683 ))
3684 .await;
3685 handler
3686 .process_raw_message(&subscription(
3687 "user.trades.any.any.raw",
3688 &serde_json::json!([trade_data()]),
3689 ))
3690 .await;
3691
3692 assert!(matches!(
3693 handler.pending_outgoing.pop_front().unwrap(),
3694 NautilusWsMessage::OrderStatusReports(reports) if reports.len() == 1
3695 ));
3696 assert!(matches!(
3697 handler.pending_outgoing.pop_front().unwrap(),
3698 NautilusWsMessage::FillReports(reports) if reports.len() == 1
3699 ));
3700 assert!(handler.pending_outgoing.is_empty());
3701 }
3702
3703 #[rstest]
3704 #[tokio::test]
3705 async fn edit_response_without_tracked_context_uses_report_path() {
3706 let mut handler = routing_test_handler();
3707 handler.pending_requests.insert(
3708 42,
3709 PendingRequestType::Edit {
3710 client_order_id: ClientOrderId::from("UNKNOWN-CLIENT-ID"),
3711 trader_id: TraderId::from("TRADER-001"),
3712 strategy_id: StrategyId::from("S-001"),
3713 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3714 },
3715 );
3716 let response = serde_json::json!({
3717 "jsonrpc": "2.0",
3718 "id": 42,
3719 "result": { "order": order_data("open", true), "trades": [trade_data()] }
3720 });
3721
3722 let message = handler.process_raw_message(&response.to_string()).await;
3723 let fill = handler.pending_outgoing.pop_front().unwrap();
3724
3725 assert!(matches!(
3726 message,
3727 Some(NautilusWsMessage::OrderStatusReports(reports)) if reports.len() == 1
3728 ));
3729 assert!(matches!(
3730 fill,
3731 NautilusWsMessage::FillReports(reports) if reports.len() == 1
3732 ));
3733 assert!(handler.order_contexts.is_empty());
3734 assert!(handler.pending_outgoing.is_empty());
3735 }
3736
3737 #[rstest]
3738 #[tokio::test]
3739 async fn tracked_edit_response_routes_fill_and_deduplicates_subscription_echo() {
3740 let mut handler = routing_test_handler();
3741 let venue_order_id = VenueOrderId::from("ETH-584830574");
3742 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3743 handler.order_contexts.insert(
3744 venue_order_id,
3745 OrderContext {
3746 client_order_id,
3747 trader_id: TraderId::from("TRADER-001"),
3748 strategy_id: StrategyId::from("S-001"),
3749 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3750 order_side: OrderSide::Buy,
3751 order_type: OrderType::Market,
3752 accepted: true,
3753 last_order_signature: None,
3754 },
3755 );
3756 handler.pending_requests.insert(
3757 42,
3758 PendingRequestType::Edit {
3759 client_order_id,
3760 trader_id: TraderId::from("TRADER-001"),
3761 strategy_id: StrategyId::from("S-001"),
3762 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3763 },
3764 );
3765 let response = serde_json::json!({
3766 "jsonrpc": "2.0",
3767 "id": 42,
3768 "result": { "order": order_data("open", true), "trades": [trade_data()] }
3769 });
3770
3771 let updated = handler.process_raw_message(&response.to_string()).await;
3772 let filled = handler.pending_outgoing.pop_front().unwrap();
3773 handler
3774 .process_raw_message(&subscription(
3775 "user.orders.any.any.raw",
3776 &serde_json::json!([order_data("open", true)]),
3777 ))
3778 .await;
3779
3780 assert!(matches!(
3781 updated,
3782 Some(NautilusWsMessage::OrderUpdated(event))
3783 if event.client_order_id == client_order_id
3784 ));
3785 assert!(matches!(
3786 filled,
3787 NautilusWsMessage::OrderFilled(event)
3788 if event.client_order_id == client_order_id
3789 ));
3790 assert!(handler.pending_outgoing.is_empty());
3791 assert!(!handler.order_contexts.contains_key(&venue_order_id));
3792 assert!(
3793 handler
3794 .terminal_order_contexts
3795 .get(&venue_order_id)
3796 .unwrap()
3797 .last_order_signature
3798 .is_some(),
3799 );
3800 }
3801
3802 #[rstest]
3803 #[tokio::test]
3804 async fn subscription_edit_echo_before_response_is_not_duplicated() {
3805 let mut handler = routing_test_handler();
3806 let venue_order_id = VenueOrderId::from("ETH-584830574");
3807 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3808 handler.order_contexts.insert(
3809 venue_order_id,
3810 OrderContext {
3811 client_order_id,
3812 trader_id: TraderId::from("TRADER-001"),
3813 strategy_id: StrategyId::from("S-001"),
3814 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3815 order_side: OrderSide::Buy,
3816 order_type: OrderType::Market,
3817 accepted: true,
3818 last_order_signature: None,
3819 },
3820 );
3821 handler.pending_requests.insert(
3822 42,
3823 PendingRequestType::Edit {
3824 client_order_id,
3825 trader_id: TraderId::from("TRADER-001"),
3826 strategy_id: StrategyId::from("S-001"),
3827 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3828 },
3829 );
3830 handler
3831 .process_raw_message(&subscription(
3832 "user.orders.any.any.raw",
3833 &serde_json::json!([order_data("open", true)]),
3834 ))
3835 .await;
3836 let updated = handler.pending_outgoing.pop_front().unwrap();
3837 let response = serde_json::json!({
3838 "jsonrpc": "2.0",
3839 "id": 42,
3840 "result": { "order": order_data("open", true), "trades": [trade_data()] }
3841 });
3842
3843 let duplicate = handler.process_raw_message(&response.to_string()).await;
3844 let filled = handler.pending_outgoing.pop_front().unwrap();
3845
3846 assert!(matches!(
3847 updated,
3848 NautilusWsMessage::OrderUpdated(event)
3849 if event.client_order_id == client_order_id
3850 ));
3851 assert!(duplicate.is_none());
3852 assert!(matches!(
3853 filled,
3854 NautilusWsMessage::OrderFilled(event)
3855 if event.client_order_id == client_order_id
3856 ));
3857 assert!(handler.pending_outgoing.is_empty());
3858 }
3859
3860 #[rstest]
3861 #[case::new_trade(true)]
3862 #[case::repeated_trade(false)]
3863 #[tokio::test]
3864 async fn delayed_edit_response_keeps_partial_fill_terminal(#[case] new_trade: bool) {
3865 let mut handler = routing_test_handler();
3866 let venue_order_id = VenueOrderId::from("ETH-584830574");
3867 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3868 handler.order_contexts.insert(
3869 venue_order_id,
3870 OrderContext {
3871 client_order_id,
3872 trader_id: TraderId::from("TRADER-001"),
3873 strategy_id: StrategyId::from("S-001"),
3874 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3875 order_side: OrderSide::Buy,
3876 order_type: OrderType::Market,
3877 accepted: true,
3878 last_order_signature: None,
3879 },
3880 );
3881 handler
3882 .process_raw_message(&subscription(
3883 "user.trades.any.any.raw",
3884 &serde_json::json!([trade_data()]),
3885 ))
3886 .await;
3887 handler.pending_outgoing.clear();
3888 handler.pending_requests.insert(
3889 42,
3890 PendingRequestType::Edit {
3891 client_order_id,
3892 trader_id: TraderId::from("TRADER-001"),
3893 strategy_id: StrategyId::from("S-001"),
3894 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3895 },
3896 );
3897 let mut partial_trade = trade_data();
3898 partial_trade["state"] = serde_json::Value::String("open".to_string());
3899 if new_trade {
3900 partial_trade["trade_id"] = serde_json::json!("ETH-2696069");
3901 }
3902
3903 let response = serde_json::json!({
3904 "jsonrpc": "2.0",
3905 "id": 42,
3906 "result": { "order": order_data("open", true), "trades": [partial_trade] }
3907 });
3908
3909 let updated = handler.process_raw_message(&response.to_string()).await;
3910 let filled = handler.pending_outgoing.pop_front();
3911
3912 assert!(updated.is_none());
3913
3914 match filled {
3915 Some(NautilusWsMessage::OrderFilled(event)) => {
3916 assert!(new_trade);
3917 assert_eq!(event.client_order_id, client_order_id);
3918 assert_eq!(event.trade_id, TradeId::from("ETH-2696069"));
3919 }
3920 None => assert!(!new_trade),
3921 message => panic!("unexpected fill message: {message:?}"),
3922 }
3923
3924 assert!(!handler.order_contexts.contains_key(&venue_order_id));
3925 assert!(
3926 handler
3927 .terminal_order_contexts
3928 .contains_key(&venue_order_id)
3929 );
3930 assert!(handler.pending_outgoing.is_empty());
3931 }
3932
3933 #[rstest]
3934 #[tokio::test]
3935 async fn tracked_replaced_order_emits_updated_event() {
3936 let mut handler = routing_test_handler();
3937 let venue_order_id = VenueOrderId::from("ETH-584830574");
3938 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
3939 handler.order_contexts.insert(
3940 venue_order_id,
3941 OrderContext {
3942 client_order_id,
3943 trader_id: TraderId::from("TRADER-001"),
3944 strategy_id: StrategyId::from("S-001"),
3945 instrument_id: InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
3946 order_side: OrderSide::Buy,
3947 order_type: OrderType::Market,
3948 accepted: true,
3949 last_order_signature: None,
3950 },
3951 );
3952
3953 handler
3954 .process_raw_message(&subscription(
3955 "user.orders.any.any.raw",
3956 &serde_json::json!([order_data("open", true)]),
3957 ))
3958 .await;
3959
3960 assert!(matches!(
3961 handler.pending_outgoing.pop_front().unwrap(),
3962 NautilusWsMessage::OrderUpdated(event)
3963 if event.client_order_id == client_order_id
3964 ));
3965
3966 let mut partial_fill = order_data("open", true);
3967 partial_fill["filled_amount"] = serde_json::json!(1.0);
3968 partial_fill["last_update_timestamp"] = serde_json::json!(1_590_480_712_900_u64);
3969 handler
3970 .process_raw_message(&subscription(
3971 "user.orders.any.any.raw",
3972 &serde_json::json!([partial_fill]),
3973 ))
3974 .await;
3975
3976 assert!(handler.pending_outgoing.is_empty());
3977 }
3978
3979 const BOOK_CHANNEL: &str = "book.BTC-PERPETUAL.100ms";
3980 const GROUPED_CHANNEL: &str = "book.BTC-PERPETUAL.none.10.100ms";
3981
3982 const UNCONVERTIBLE_PRICE: f64 = 1e20;
3984
3985 fn book_message(
3986 msg_type: &str,
3987 change_id: u64,
3988 prev_change_id: Option<u64>,
3989 bid_price: f64,
3990 ) -> String {
3991 subscription(
3992 BOOK_CHANNEL,
3993 &serde_json::json!({
3994 "type": msg_type,
3995 "instrument_name": "BTC-PERPETUAL",
3996 "timestamp": 1_700_000_000_000_u64,
3997 "change_id": change_id,
3998 "prev_change_id": prev_change_id,
3999 "bids": [["new", bid_price, 10.0]],
4000 "asks": [],
4001 }),
4002 )
4003 }
4004
4005 fn grouped_book_message(change_id: u64, bid_price: f64) -> String {
4007 subscription(
4008 GROUPED_CHANNEL,
4009 &serde_json::json!({
4010 "instrument_name": "BTC-PERPETUAL",
4011 "timestamp": 1_700_000_000_000_u64,
4012 "change_id": change_id,
4013 "bids": [[bid_price, 10.0]],
4014 "asks": [],
4015 }),
4016 )
4017 }
4018
4019 fn book_test_handler() -> DeribitWsFeedHandler {
4020 let handler = routing_test_handler();
4021
4022 for channel in [
4023 BOOK_CHANNEL,
4024 "book.ETH-PERPETUAL.100ms",
4025 "trades.BTC-PERPETUAL.100ms",
4026 ] {
4027 handler.subscriptions_state.mark_subscribe(channel);
4028 handler.subscriptions_state.confirm_subscribe(channel);
4029 }
4030
4031 handler
4032 }
4033
4034 fn pending_unsubscribe_channels(handler: &DeribitWsFeedHandler) -> Vec<Vec<String>> {
4035 handler
4036 .pending_requests
4037 .values()
4038 .filter_map(|request| match request {
4039 PendingRequestType::Unsubscribe { channels } => Some(channels.clone()),
4040 _ => None,
4041 })
4042 .collect()
4043 }
4044
4045 #[rstest]
4046 #[tokio::test]
4047 async fn book_sequence_gap_forces_resync_after_contiguous_delta() {
4048 let mut handler = book_test_handler();
4049
4050 let snapshot = handler
4051 .process_raw_message(&book_message("snapshot", 100, None, 42500.0))
4052 .await;
4053 let contiguous = handler
4054 .process_raw_message(&book_message("change", 101, Some(100), 42499.5))
4055 .await;
4056
4057 assert!(matches!(
4058 snapshot,
4059 Some(NautilusWsMessage::Deltas(deltas)) if deltas.sequence == 100
4060 ));
4061 assert!(matches!(
4062 contiguous,
4063 Some(NautilusWsMessage::Deltas(deltas)) if deltas.sequence == 101
4064 ));
4065 assert_eq!(
4066 handler.book_sequence.get(&Ustr::from("BTC-PERPETUAL")),
4067 Some(&101)
4068 );
4069 assert!(handler.pending_book_resync.is_empty());
4070
4071 let gap = handler
4072 .process_raw_message(&book_message("change", 103, Some(102), 42499.0))
4073 .await;
4074
4075 assert!(gap.is_none());
4076 assert!(handler.book_sequence.is_empty());
4077 assert_eq!(handler.pending_book_resync, vec![BOOK_CHANNEL.to_string()]);
4078 assert_eq!(
4079 pending_unsubscribe_channels(&handler),
4080 vec![vec![BOOK_CHANNEL.to_string()]]
4081 );
4082 }
4083
4084 #[tokio::test]
4085 async fn grouped_book_waits_for_depth_subscription_update() {
4086 let mut handler = book_test_handler();
4087 handler
4088 .book_depth_subs
4089 .lock()
4090 .await
4091 .add_reference(GROUPED_CHANNEL);
4092 handler.subscriptions_state.add_reference(GROUPED_CHANNEL);
4093 handler
4094 .process_raw_message(&book_message("snapshot", 100, None, 42500.0))
4095 .await;
4096 let depth_subs = Arc::clone(&handler.book_depth_subs);
4097 let subscriptions = handler.subscriptions_state.clone();
4098 let depth_guard = depth_subs.lock().await;
4099 depth_guard.add_reference(GROUPED_CHANNEL);
4100 let message = grouped_book_message(999, 89532.5);
4101 let mut processing = Box::pin(handler.process_raw_message(&message));
4102
4103 assert!(futures_util::poll!(processing.as_mut()).is_pending());
4104
4105 subscriptions.add_reference(GROUPED_CHANNEL);
4106 drop(depth_guard);
4107 let output = processing.await.unwrap();
4108
4109 let NautilusWsMessage::Data(data) = output else {
4110 panic!("expected grouped depth");
4111 };
4112
4113 let [Data::BookDepth(depth)] = data.as_slice() else {
4114 panic!("expected only depth data");
4115 };
4116
4117 assert_eq!(depth.sequence, 999);
4118 assert!(handler.pending_outgoing.is_empty());
4119 assert_eq!(
4120 handler.book_sequence.get(&Ustr::from("BTC-PERPETUAL")),
4121 Some(&100)
4122 );
4123 assert!(handler.pending_book_resync.is_empty());
4124 }
4125
4126 #[rstest]
4127 #[case::depth_only(false)]
4128 #[case::shared_grouped_channel(true)]
4129 #[tokio::test]
4130 async fn grouped_depth_does_not_change_raw_book_sequence(#[case] grouped_deltas: bool) {
4131 let mut handler = book_test_handler();
4132 handler
4133 .book_depth_subs
4134 .lock()
4135 .await
4136 .add_reference(GROUPED_CHANNEL);
4137 handler.subscriptions_state.add_reference(GROUPED_CHANNEL);
4138 if grouped_deltas {
4139 handler.subscriptions_state.add_reference(GROUPED_CHANNEL);
4140 }
4141
4142 let snapshot = handler
4143 .process_raw_message(&book_message("snapshot", 100, None, 42500.0))
4144 .await
4145 .unwrap();
4146 let grouped = handler
4147 .process_raw_message(&grouped_book_message(999, 89532.5))
4148 .await
4149 .unwrap();
4150 let sequence_after_grouped = handler
4151 .book_sequence
4152 .get(&Ustr::from("BTC-PERPETUAL"))
4153 .copied();
4154 let contiguous = handler
4155 .process_raw_message(&book_message("change", 101, Some(100), 42499.5))
4156 .await
4157 .unwrap();
4158
4159 let mut book = OrderBook::new(
4160 InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
4161 BookType::L2_MBP,
4162 );
4163
4164 let NautilusWsMessage::Deltas(snapshot) = snapshot else {
4165 panic!("expected raw snapshot");
4166 };
4167
4168 book.apply_deltas(&snapshot).unwrap();
4169
4170 let NautilusWsMessage::Data(data) = grouped else {
4171 panic!("expected grouped depth");
4172 };
4173
4174 let [Data::BookDepth(depth)] = data.as_slice() else {
4175 panic!("expected only depth data");
4176 };
4177
4178 let NautilusWsMessage::Deltas(contiguous) = contiguous else {
4179 panic!("expected raw delta");
4180 };
4181
4182 book.apply_deltas(&contiguous).unwrap();
4183
4184 assert_eq!(sequence_after_grouped, Some(100));
4185 assert_eq!(
4186 handler.book_sequence.get(&Ustr::from("BTC-PERPETUAL")),
4187 Some(&101)
4188 );
4189 assert_eq!(depth.sequence, 999);
4190 assert_eq!(depth.bids[0].price, Price::from("89532.5"));
4191 assert_eq!(
4192 book.bids_as_map(None).into_iter().collect::<Vec<_>>(),
4193 vec![
4194 (Decimal::from(42500), Decimal::from(10)),
4195 (Decimal::new(424995, 1), Decimal::from(10)),
4196 ]
4197 );
4198 assert!(handler.pending_book_resync.is_empty());
4199 assert!(pending_unsubscribe_channels(&handler).is_empty());
4200 assert_eq!(handler.pending_outgoing.len(), usize::from(grouped_deltas));
4201
4202 if grouped_deltas {
4203 let Some(NautilusWsMessage::Deltas(deltas)) = handler.pending_outgoing.pop_front()
4204 else {
4205 panic!("expected requested grouped deltas");
4206 };
4207
4208 assert_eq!(deltas.sequence, 999);
4209 }
4210 }
4211
4212 #[rstest]
4213 #[case::deltas(false)]
4214 #[case::depth(true)]
4215 #[tokio::test]
4216 async fn rejected_grouped_book_snapshot_waits_for_next_snapshot(#[case] depth: bool) {
4217 let mut handler = book_test_handler();
4218 handler.subscriptions_state.add_reference(GROUPED_CHANNEL);
4219 if depth {
4220 handler
4221 .book_depth_subs
4222 .lock()
4223 .await
4224 .add_reference(GROUPED_CHANNEL);
4225 }
4226
4227 handler.subscriptions_state.mark_subscribe(GROUPED_CHANNEL);
4228 handler
4229 .subscriptions_state
4230 .confirm_subscribe(GROUPED_CHANNEL);
4231
4232 let rejected = handler
4233 .process_raw_message(&grouped_book_message(100, UNCONVERTIBLE_PRICE))
4234 .await;
4235
4236 assert!(rejected.is_none());
4237 assert!(handler.book_sequence.is_empty());
4238 assert!(handler.pending_book_resync.is_empty());
4239 assert!(pending_unsubscribe_channels(&handler).is_empty());
4240
4241 let next = handler
4242 .process_raw_message(&grouped_book_message(101, 42500.0))
4243 .await;
4244
4245 match next.unwrap() {
4246 NautilusWsMessage::Deltas(deltas) if !depth => assert_eq!(deltas.sequence, 101),
4247 NautilusWsMessage::Data(data) if depth => {
4248 let [Data::BookDepth(snapshot)] = data.as_slice() else {
4249 panic!("expected depth");
4250 };
4251
4252 assert_eq!(snapshot.sequence, 101);
4253 }
4254 message => panic!("unexpected message: {message:?}"),
4255 }
4256
4257 assert!(handler.book_sequence.is_empty());
4258 }
4259
4260 #[rstest]
4261 #[case::deltas(false)]
4262 #[case::depth(true)]
4263 #[tokio::test]
4264 async fn delta_channel_resync_survives_grouped_snapshot_before_unsubscribe_ack(
4265 #[case] depth: bool,
4266 ) {
4267 let mut handler = book_test_handler();
4268 handler.subscriptions_state.add_reference(GROUPED_CHANNEL);
4269 if depth {
4270 handler
4271 .book_depth_subs
4272 .lock()
4273 .await
4274 .add_reference(GROUPED_CHANNEL);
4275 }
4276
4277 handler.subscriptions_state.mark_subscribe(GROUPED_CHANNEL);
4278 handler
4279 .subscriptions_state
4280 .confirm_subscribe(GROUPED_CHANNEL);
4281
4282 handler
4283 .process_raw_message(&book_message("snapshot", 100, None, 42500.0))
4284 .await;
4285 handler
4286 .process_raw_message(&book_message("change", 101, Some(100), UNCONVERTIBLE_PRICE))
4287 .await;
4288 let grouped = handler
4289 .process_raw_message(&grouped_book_message(102, 42500.0))
4290 .await;
4291
4292 match grouped.unwrap() {
4293 NautilusWsMessage::Deltas(deltas) if !depth => assert_eq!(deltas.sequence, 102),
4294 NautilusWsMessage::Data(data) if depth => {
4295 let [Data::BookDepth(snapshot)] = data.as_slice() else {
4296 panic!("expected depth");
4297 };
4298
4299 assert_eq!(snapshot.sequence, 102);
4300 }
4301 message => panic!("unexpected message: {message:?}"),
4302 }
4303
4304 assert!(handler.book_sequence.is_empty());
4305 assert_eq!(handler.pending_book_resync, vec![BOOK_CHANNEL.to_string()]);
4306 assert_eq!(
4307 pending_unsubscribe_channels(&handler),
4308 vec![vec![BOOK_CHANNEL.to_string()]]
4309 );
4310
4311 let unsubscribe_id = handler
4312 .pending_requests
4313 .iter()
4314 .find_map(|(id, request)| {
4315 matches!(request, PendingRequestType::Unsubscribe { .. }).then_some(*id)
4316 })
4317 .unwrap();
4318
4319 let ack = serde_json::json!({
4320 "jsonrpc": "2.0",
4321 "id": unsubscribe_id,
4322 "result": [BOOK_CHANNEL],
4323 });
4324 handler.process_raw_message(&ack.to_string()).await;
4325
4326 let resubscribed: Vec<Vec<String>> = handler
4327 .pending_requests
4328 .values()
4329 .filter_map(|request| match request {
4330 PendingRequestType::Subscribe { channels } => Some(channels.clone()),
4331 _ => None,
4332 })
4333 .collect();
4334
4335 assert_eq!(resubscribed, vec![vec![BOOK_CHANNEL.to_string()]]);
4336 }
4337
4338 #[rstest]
4339 #[tokio::test]
4340 async fn rejected_book_delta_forces_resync_and_drops_successor() {
4341 let mut handler = book_test_handler();
4342
4343 let snapshot = handler
4344 .process_raw_message(&book_message("snapshot", 100, None, 42500.0))
4345 .await;
4346 let rejected = handler
4347 .process_raw_message(&book_message("change", 101, Some(100), UNCONVERTIBLE_PRICE))
4348 .await;
4349 let successor = handler
4350 .process_raw_message(&book_message("change", 102, Some(101), 42499.5))
4351 .await;
4352
4353 assert!(matches!(
4354 snapshot,
4355 Some(NautilusWsMessage::Deltas(deltas)) if deltas.sequence == 100
4356 ));
4357 assert!(rejected.is_none());
4358 assert!(successor.is_none());
4359 assert!(handler.book_sequence.is_empty());
4360 assert_eq!(handler.pending_book_resync, vec![BOOK_CHANNEL.to_string()]);
4361 assert_eq!(
4362 pending_unsubscribe_channels(&handler),
4363 vec![vec![BOOK_CHANNEL.to_string()]]
4364 );
4365 assert_eq!(
4366 handler.subscriptions_state.pending_subscribe_topics(),
4367 vec![BOOK_CHANNEL.to_string()]
4368 );
4369
4370 let recovered = handler
4371 .process_raw_message(&book_message("snapshot", 200, None, 42500.0))
4372 .await;
4373
4374 assert!(matches!(
4375 recovered,
4376 Some(NautilusWsMessage::Deltas(deltas)) if deltas.sequence == 200
4377 ));
4378 assert!(handler.pending_book_resync.is_empty());
4379 assert_eq!(
4380 handler.book_sequence.get(&Ustr::from("BTC-PERPETUAL")),
4381 Some(&200)
4382 );
4383 }
4384
4385 #[rstest]
4386 #[tokio::test]
4387 async fn rejected_resync_book_snapshot_keeps_resync_pending() {
4388 let mut handler = book_test_handler();
4389 handler.pending_book_resync.push(BOOK_CHANNEL.to_string());
4390
4391 let rejected = handler
4392 .process_raw_message(&book_message("snapshot", 200, None, UNCONVERTIBLE_PRICE))
4393 .await;
4394 let delta = handler
4395 .process_raw_message(&book_message("change", 201, Some(200), 42499.5))
4396 .await;
4397
4398 assert!(rejected.is_none());
4399 assert!(delta.is_none());
4400 assert!(handler.book_sequence.is_empty());
4401 assert_eq!(handler.pending_book_resync, vec![BOOK_CHANNEL.to_string()]);
4402 assert_eq!(
4403 pending_unsubscribe_channels(&handler),
4404 vec![vec![BOOK_CHANNEL.to_string()]]
4405 );
4406 }
4407
4408 #[rstest]
4409 #[case(OrderSide::Buy)]
4410 #[case(OrderSide::Sell)]
4411 #[tokio::test]
4412 async fn order_serialization_failure_leaves_no_order_context(#[case] side: OrderSide) {
4413 let mut handler = routing_test_handler();
4414 let client_order_id = ClientOrderId::from("O-19700101-000000-001-001-1");
4415 let trader_id = TraderId::from("TRADER-001");
4416 let strategy_id = StrategyId::from("S-001");
4417 let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
4418
4419 let params = DeribitOrderParams {
4420 instrument_name: "BTC-PERPETUAL".to_string(),
4421 amount: Decimal::from_str_exact("100000000.005000001").unwrap(),
4422 order_type: "limit".to_string(),
4423 label: Some(client_order_id.to_string()),
4424 price: Some(Decimal::from_str_exact("65000.5").unwrap()),
4425 time_in_force: Some("good_til_cancelled".to_string()),
4426 post_only: None,
4427 reject_post_only: None,
4428 reduce_only: None,
4429 trigger_price: None,
4430 trigger: None,
4431 max_show: None,
4432 valid_until: None,
4433 };
4434
4435 let result = match side {
4436 OrderSide::Buy => {
4437 handler
4438 .handle_buy(
4439 params,
4440 client_order_id,
4441 trader_id,
4442 strategy_id,
4443 instrument_id,
4444 )
4445 .await
4446 }
4447 OrderSide::Sell => {
4448 handler
4449 .handle_sell(
4450 params,
4451 client_order_id,
4452 trader_id,
4453 strategy_id,
4454 instrument_id,
4455 )
4456 .await
4457 }
4458 };
4459
4460 if "0.10".parse::<serde_json::Number>().unwrap().to_string() == "0.10" {
4462 assert!(matches!(result, Err(DeribitWsError::NotConnected)));
4463 assert!(
4464 handler
4465 .submitted_order_contexts
4466 .get(&client_order_id)
4467 .is_some()
4468 );
4469 assert_eq!(handler.pending_requests.len(), 1);
4470 } else {
4471 assert!(matches!(
4472 result,
4473 Err(DeribitWsError::Json(message))
4474 if message == "exact decimal number serialization requires serde_json/arbitrary_precision"
4475 ));
4476 assert!(
4477 handler
4478 .submitted_order_contexts
4479 .get(&client_order_id)
4480 .is_none()
4481 );
4482 assert!(handler.pending_requests.is_empty());
4483 }
4484 }
4485
4486 #[rstest]
4487 #[case(DeribitWsMethod::PublicSubscribe)]
4488 #[case(DeribitWsMethod::PrivateSubscribe)]
4489 #[case(DeribitWsMethod::PublicUnsubscribe)]
4490 #[case(DeribitWsMethod::PrivateUnsubscribe)]
4491 fn subscription_batches_bound_serialized_bytes(#[case] method: DeribitWsMethod) {
4492 let mut channels: Vec<String> = (0..1500)
4493 .map(|i| format!("book.BTC-27DEC30-{}-C.none.20.100ms", 100000 + i))
4494 .collect();
4495 channels.push("quoted.\"\\\n.é".repeat(100));
4496 let batches = subscription_batches(channels.clone(), &method).unwrap();
4497 assert_eq!(batches.len(), 3);
4498 let mut actual = Vec::new();
4499
4500 for channels in batches {
4501 let request = DeribitJsonRpcRequest::new(
4502 u64::MAX,
4503 method.as_method_str(),
4504 DeribitSubscribeParams { channels },
4505 );
4506 let payload = serde_json::to_string(&request).unwrap();
4507 assert!(payload.len() <= SUBSCRIPTION_REQUEST_MAX_BYTES);
4508 actual.extend(request.params.channels);
4509 }
4510
4511 assert_eq!(actual, channels);
4512 }
4513
4514 #[rstest]
4515 fn subscription_batches_exact_boundary_and_oversized_channel() {
4516 let method = DeribitWsMethod::PrivateSubscribe;
4517
4518 let request = DeribitJsonRpcRequest::new(
4519 u64::MAX,
4520 method.as_method_str(),
4521 DeribitSubscribeParams {
4522 channels: vec![String::new()],
4523 },
4524 );
4525
4526 let overhead = serde_json::to_string(&request).unwrap().len();
4527 let channel = "a".repeat(SUBSCRIPTION_REQUEST_MAX_BYTES - overhead);
4528 assert_eq!(
4529 subscription_batches(vec![channel.clone()], &method).unwrap(),
4530 vec![vec![channel.clone()]],
4531 );
4532 assert!(matches!(
4533 subscription_batches(vec![format!("{channel}a")], &method),
4534 Err(DeribitWsError::Subscribe(_)),
4535 ));
4536 assert!(
4537 subscription_batches(Vec::new(), &method)
4538 .unwrap()
4539 .is_empty()
4540 );
4541 }
4542
4543 #[tokio::test]
4544 async fn subscription_batches_correlate_rejections_and_reconnect() {
4545 let mut handler = routing_test_handler();
4546 let channels: Vec<String> = (0..1500)
4547 .map(|i| format!("book.BTC-27DEC30-{}-C.none.20.100ms", 100000 + i))
4548 .collect();
4549
4550 for channel in &channels {
4551 handler.subscriptions_state.add_reference(channel);
4552 handler.subscriptions_state.mark_subscribe(channel);
4553 }
4554
4555 handler.handle_subscribe(channels.clone()).unwrap();
4556 assert_eq!(handler.subscription_requests.len(), 2);
4557 let requests: Vec<_> = handler.subscription_requests.iter().cloned().collect();
4558 for (index, (request_id, payload)) in requests.iter().enumerate() {
4559 let request: serde_json::Value = serde_json::from_str(payload).unwrap();
4560
4561 let result = if index == 0 {
4562 serde_json::json!({"jsonrpc": "2.0", "id": request_id, "result": request["params"]["channels"]})
4563 } else {
4564 serde_json::json!({"jsonrpc": "2.0", "id": request_id, "error": {"code": 10028, "message": "too_many_requests"}})
4565 };
4566
4567 handler.process_raw_message(&result.to_string()).await;
4568 }
4569
4570 let rejected: serde_json::Value = serde_json::from_str(&requests[1].1).unwrap();
4571 let rejected: Vec<String> =
4572 serde_json::from_value(rejected["params"]["channels"].clone()).unwrap();
4573 let mut pending = handler.subscriptions_state.pending_subscribe_topics();
4574 pending.sort();
4575 assert_eq!(pending, rejected);
4576 assert!(handler.pending_requests.is_empty());
4577 assert_eq!(handler.subscribe_errors.lock().len(), 1);
4578 let mut restored = handler.subscriptions_state.reset_after_reconnect();
4579 restored.sort();
4580 assert_eq!(restored, channels);
4581 handler.process_raw_message(RECONNECTED).await;
4582 assert!(handler.subscription_requests.is_empty());
4583 assert!(handler.subscription_send.is_none());
4584 assert_eq!(
4585 handler
4586 .subscriptions_state
4587 .get_reference_count(&channels[0]),
4588 1
4589 );
4590 }
4591
4592 #[rstest]
4593 fn subscription_coalescing_preserves_barriers_and_authentication() {
4594 let mut handler = routing_test_handler();
4595 let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
4596 handler.cmd_rx = cmd_rx;
4597 cmd_tx
4598 .send(HandlerCommand::Subscribe {
4599 channels: vec!["ticker.ETH-PERPETUAL.100ms".into()],
4600 })
4601 .unwrap();
4602
4603 cmd_tx
4604 .send(HandlerCommand::Unsubscribe {
4605 channels: vec!["ticker.BTC-PERPETUAL.100ms".into()],
4606 })
4607 .unwrap();
4608
4609 cmd_tx
4610 .send(HandlerCommand::Subscribe {
4611 channels: vec!["ticker.SOL_USDC-PERPETUAL.100ms".into()],
4612 })
4613 .unwrap();
4614
4615 let combined = handler.coalesce_subscriptions(vec!["ticker.BTC-PERPETUAL.100ms".into()]);
4616 assert_eq!(
4617 combined,
4618 vec!["ticker.BTC-PERPETUAL.100ms", "ticker.ETH-PERPETUAL.100ms"]
4619 );
4620 assert!(matches!(
4621 handler.pending_command.take(),
4622 Some(HandlerCommand::Unsubscribe { .. })
4623 ));
4624 assert!(matches!(
4625 handler.cmd_rx.try_recv(),
4626 Ok(HandlerCommand::Subscribe { .. })
4627 ));
4628 cmd_tx
4629 .send(HandlerCommand::Subscribe {
4630 channels: vec!["user.orders.any.any.raw".into()],
4631 })
4632 .unwrap();
4633
4634 let combined = handler.coalesce_subscriptions(vec!["ticker.BTC-PERPETUAL.100ms".into()]);
4635 assert_eq!(combined, vec!["ticker.BTC-PERPETUAL.100ms"]);
4636 assert!(matches!(
4637 handler.pending_command,
4638 Some(HandlerCommand::Subscribe { .. })
4639 ));
4640 }
4641
4642 #[rstest]
4643 #[case::reconnect(true)]
4644 #[case::shutdown(false)]
4645 #[tokio::test]
4646 async fn subscription_quota_wait_keeps_messages_and_lifecycle_responsive(
4647 #[case] reconnect: bool,
4648 ) {
4649 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4650 let addr = listener.local_addr().unwrap();
4651 let server = tokio::spawn(async move {
4652 let (stream, _) = listener.accept().await.unwrap();
4653 tokio_tungstenite::accept_async(stream).await.unwrap()
4654 });
4655
4656 let (message_handler, raw_rx) = channel_message_handler();
4657 let config = WebSocketConfig::builder()
4658 .url(format!("ws://{addr}"))
4659 .build()
4660 .unwrap();
4661 let quota = Quota::with_period(Duration::from_secs(60)).unwrap();
4662 let ws_client = WebSocketClient::builder()
4663 .config(config)
4664 .message_handler(message_handler)
4665 .keyed_quotas(vec![(DERIBIT_WS_SUBSCRIPTION_KEY.to_string(), quota)])
4666 .connect()
4667 .await
4668 .unwrap();
4669 let mut socket = server.await.unwrap();
4670 ws_client
4671 .send_text(
4672 "reserve".to_string(),
4673 Some(&[Ustr::from(DERIBIT_WS_SUBSCRIPTION_KEY)]),
4674 )
4675 .await
4676 .unwrap();
4677 assert_eq!(
4678 socket.next().await.unwrap().unwrap(),
4679 Message::Text("reserve".into())
4680 );
4681
4682 let mut handler = routing_test_handler();
4683 handler.inner = Some(Arc::new(ws_client));
4684 handler.raw_rx = raw_rx;
4685 let (out_tx, mut out_rx) = tokio::sync::mpsc::unbounded_channel();
4686 handler.out_tx = out_tx;
4687 handler
4688 .handle_authenticate(
4689 serde_json::json!({"grant_type":"client_credentials"})
4690 .to_string()
4691 .into(),
4692 )
4693 .await;
4694 let auth: serde_json::Value =
4695 serde_json::from_str(socket.next().await.unwrap().unwrap().to_text().unwrap()).unwrap();
4696 assert_eq!(auth["method"], "public/auth");
4697 assert_eq!(auth["id"], 1);
4698 let channels = vec![
4699 "ticker.BTC-PERPETUAL.100ms".to_string(),
4700 "ticker.ETH-PERPETUAL.100ms".to_string(),
4701 ];
4702
4703 for channel in &channels {
4704 handler.subscriptions_state.add_reference(channel);
4705 handler.subscriptions_state.mark_subscribe(channel);
4706 handler.handle_subscribe(vec![channel.clone()]).unwrap();
4707 }
4708
4709 handler.subscriptions_state.add_reference(&channels[0]);
4710 assert!(
4711 tokio::time::timeout(Duration::from_millis(20), handler.next())
4712 .await
4713 .is_err()
4714 );
4715 assert!(handler.subscription_send.is_some());
4716 assert_eq!(handler.subscription_requests.len(), 1);
4717 assert!(
4718 tokio::time::timeout(Duration::from_millis(20), socket.next())
4719 .await
4720 .is_err()
4721 );
4722
4723 let incoming = serde_json::json!({"value":17});
4724 socket
4725 .send(Message::Text(
4726 subscription("unknown.channel", &incoming).into(),
4727 ))
4728 .await
4729 .unwrap();
4730 socket.send(Message::Text(serde_json::json!({"jsonrpc":"2.0","method":"heartbeat","params":{"type":"test_request"}}).to_string().into())).await.unwrap();
4731 socket
4732 .send(Message::Ping(vec![7, 11].into()))
4733 .await
4734 .unwrap();
4735 socket
4736 .send(Message::Text(
4737 serde_json::json!({
4738 "jsonrpc":"2.0", "id":1, "result":{
4739 "access_token":"access", "refresh_token":"refresh", "expires_in":901,
4740 "scope":"session:test", "token_type":"bearer", "enabled_features":[]
4741 }
4742 })
4743 .to_string()
4744 .into(),
4745 ))
4746 .await
4747 .unwrap();
4748 let authenticated = tokio::time::timeout(Duration::from_millis(500), handler.next())
4749 .await
4750 .unwrap();
4751
4752 let Some(NautilusWsMessage::Authenticated(auth)) = authenticated else {
4753 panic!("Expected authentication response during quota wait");
4754 };
4755
4756 assert_eq!(auth.access_token.expose_secret(), "access");
4757 assert_eq!(auth.refresh_token.expose_secret(), "refresh");
4758 assert_eq!(auth.expires_in, 901);
4759 assert_eq!(auth.scope, "session:test");
4760 assert_eq!(auth.token_type, "bearer");
4761 assert_eq!(auth.enabled_features, Vec::<String>::new());
4762 assert!(
4763 matches!(out_rx.try_recv().unwrap(), NautilusWsMessage::Raw(data) if serde_json::from_str::<serde_json::Value>(data.get()).unwrap() == incoming)
4764 );
4765 let mut heartbeat = None;
4766 let mut pong = None;
4767
4768 for _ in 0..2 {
4769 let message = tokio::time::timeout(Duration::from_millis(500), socket.next())
4770 .await
4771 .unwrap()
4772 .unwrap()
4773 .unwrap();
4774
4775 match message {
4776 Message::Text(text) => {
4777 heartbeat = Some(serde_json::from_str::<serde_json::Value>(&text).unwrap());
4778 }
4779 Message::Pong(data) => pong = Some(data.to_vec()),
4780 message => panic!("Unexpected response: {message:?}"),
4781 }
4782 }
4783
4784 assert_eq!(
4785 heartbeat,
4786 Some(serde_json::json!({"jsonrpc":"2.0", "id":4, "method":"public/test", "params":{}}))
4787 );
4788 assert_eq!(pong, Some(vec![7, 11]));
4789 assert!(handler.subscription_send.is_some());
4790 assert_eq!(handler.subscription_requests.len(), 1);
4791
4792 if reconnect {
4793 socket
4794 .send(Message::Text(RECONNECTED.into()))
4795 .await
4796 .unwrap();
4797 let result = tokio::time::timeout(Duration::from_millis(500), handler.next())
4798 .await
4799 .unwrap();
4800 assert!(matches!(result, Some(NautilusWsMessage::Reconnected)));
4801 assert!(handler.subscription_send.is_none());
4802 assert!(handler.subscription_requests.is_empty());
4803 assert!(handler.pending_requests.is_empty());
4804 let mut restored = handler.subscriptions_state.reset_after_reconnect();
4805 restored.sort();
4806 assert_eq!(restored, channels);
4807 assert_eq!(
4808 handler
4809 .subscriptions_state
4810 .get_reference_count(&channels[0]),
4811 2
4812 );
4813 assert_eq!(
4814 handler
4815 .subscriptions_state
4816 .get_reference_count(&channels[1]),
4817 1
4818 );
4819 }
4820
4821 handler.signal.store(true, Ordering::Relaxed);
4822 let result = tokio::time::timeout(Duration::from_millis(500), handler.next())
4823 .await
4824 .unwrap();
4825 assert!(result.is_none());
4826 assert!(handler.subscription_send.is_none());
4827 let close = tokio::time::timeout(Duration::from_millis(500), socket.next())
4828 .await
4829 .unwrap()
4830 .unwrap()
4831 .unwrap();
4832 assert!(matches!(close, Message::Close(_)));
4833 }
4834
4835 #[rstest]
4836 #[case(true, "private/subscribe")]
4837 #[case(false, "private/unsubscribe")]
4838 fn mixed_subscription_batches_keep_original_private_method(
4839 #[case] subscribe: bool,
4840 #[case] method: &str,
4841 ) {
4842 let mut handler = routing_test_handler();
4843 let mut channels: Vec<String> = (0..1500)
4844 .map(|i| format!("ticker.BTC-27DEC30-{}-C.100ms", 100000 + i))
4845 .collect();
4846 channels.push("user.orders.any.any.raw".into());
4847 handler
4848 .queue_subscription_requests(channels.clone(), subscribe)
4849 .unwrap();
4850 assert_eq!(handler.subscription_requests.len(), 2);
4851 let mut actual = Vec::new();
4852
4853 for (_, payload) in &handler.subscription_requests {
4854 let request: serde_json::Value = serde_json::from_str(payload).unwrap();
4855 assert_eq!(request["method"], method);
4856 actual.extend(
4857 serde_json::from_value::<Vec<String>>(request["params"]["channels"].clone())
4858 .unwrap(),
4859 );
4860 }
4861
4862 assert_eq!(actual, channels);
4863 }
4864
4865 #[tokio::test]
4866 async fn subscription_send_wait_processes_heartbeat_and_authentication() {
4867 let mut handler = routing_test_handler();
4868 let (raw_tx, raw_rx) = tokio::sync::mpsc::unbounded_channel();
4869 handler.raw_rx = raw_rx;
4870 handler.subscription_send = Some(Box::pin(std::future::pending()));
4871 handler
4872 .pending_requests
4873 .insert(42, PendingRequestType::Authenticate);
4874 raw_tx
4875 .send(Message::Text(load_test_json("ws_test_request.json").into()))
4876 .unwrap();
4877 let response = serde_json::json!({
4878 "jsonrpc": "2.0", "id": 42,
4879 "result": {
4880 "access_token": "test-access", "refresh_token": "test-refresh", "expires_in": 900,
4881 "scope": "connection", "token_type": "bearer"
4882 }
4883 });
4884 raw_tx
4885 .send(Message::Text(response.to_string().into()))
4886 .unwrap();
4887 let result = tokio::time::timeout(std::time::Duration::from_millis(500), handler.next())
4888 .await
4889 .unwrap();
4890 assert!(matches!(result, Some(NautilusWsMessage::Authenticated(_))));
4891 assert!(handler.subscription_send.is_some());
4892 assert!(!handler.pending_requests.contains_key(&42));
4893 assert!(matches!(
4894 handler.pending_requests.get(&1),
4895 Some(PendingRequestType::Test)
4896 ));
4897 }
4898
4899 #[tokio::test]
4900 async fn subscription_send_wait_preserves_command_order() {
4901 let mut handler = routing_test_handler();
4902 let (cmd_tx, cmd_rx) = tokio::sync::mpsc::unbounded_channel();
4903 handler.cmd_rx = cmd_rx;
4904 handler.subscription_send = Some(Box::pin(std::future::pending()));
4905 cmd_tx
4906 .send(HandlerCommand::SetHeartbeat { interval: 37 })
4907 .unwrap();
4908 {
4909 let next = handler.next();
4910 tokio::pin!(next);
4911 assert!(futures_util::poll!(next).is_pending());
4912 }
4913
4914 assert!(handler.pending_requests.is_empty());
4915 assert_eq!(handler.cmd_rx.len(), 1);
4916 handler.subscription_send = None;
4917 {
4918 let next = handler.next();
4919 tokio::pin!(next);
4920 assert!(futures_util::poll!(next).is_pending());
4921 }
4922
4923 assert_eq!(handler.cmd_rx.len(), 0);
4924 assert_eq!(handler.pending_requests.len(), 1);
4925 assert!(matches!(
4926 handler.pending_requests.get(&1),
4927 Some(PendingRequestType::SetHeartbeat)
4928 ));
4929 }
4930
4931 #[rstest]
4932 #[case(true)]
4933 #[case(false)]
4934 #[tokio::test]
4935 async fn subscription_send_failure_removes_request_correlation(#[case] subscribe: bool) {
4936 let mut handler = routing_test_handler();
4937 let channels = vec!["ticker.BTC-PERPETUAL.100ms".into()];
4938 let request = if subscribe {
4939 PendingRequestType::Subscribe { channels }
4940 } else {
4941 PendingRequestType::Unsubscribe { channels }
4942 };
4943
4944 handler.pending_requests.insert(73, request);
4945
4946 handler.subscription_send =
4947 Some(Box::pin(async { (73, Err(DeribitWsError::NotConnected)) }));
4948 {
4949 let next = handler.next();
4950 tokio::pin!(next);
4951 assert!(futures_util::poll!(next).is_pending());
4952 }
4953
4954 assert!(handler.pending_requests.is_empty());
4955 assert!(handler.subscription_send.is_none());
4956
4957 let expected_errors = if subscribe {
4958 vec![DeribitWsError::NotConnected.to_string()]
4959 } else {
4960 Vec::new()
4961 };
4962
4963 assert_eq!(*handler.subscribe_errors.lock(), expected_errors);
4964 }
4965}