nautilus_deribit/
execution.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Live execution client implementation for the Deribit adapter.
17
18use std::{future::Future, io, time::Duration};
19
20use anyhow::Context;
21use async_trait::async_trait;
22use futures_util::{StreamExt, pin_mut};
23use nautilus_common::{
24    clients::ExecutionClient,
25    live::runner::get_exec_event_sender,
26    messages::execution::{
27        BatchCancelOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
28        GenerateFillReportsBuilder, GenerateOrderStatusReport, GenerateOrderStatusReports,
29        GenerateOrderStatusReportsBuilder, GeneratePositionStatusReports,
30        GeneratePositionStatusReportsBuilder, ModifyOrder, QueryAccount, QueryOrder, SubmitOrder,
31        SubmitOrderList,
32    },
33};
34use nautilus_core::{
35    DurationNanos, Params, UnixNanos,
36    time::{AtomicTime, get_atomic_clock_realtime},
37};
38use nautilus_live::{
39    ExecutionClientCore, ExecutionEventEmitter, SocketControl,
40    task::{TaskGroup, TaskGroupGuard},
41};
42use nautilus_model::{
43    accounts::AccountAny,
44    enums::{AccountType, OmsType, OrderType, TimeInForce},
45    events::OrderEventAny,
46    identifiers::{AccountId, ClientId, Venue},
47    orders::{Order, OrderAny},
48    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
49    types::{AccountBalance, MarginBalance},
50};
51use serde::Serialize;
52
53use crate::{
54    common::{
55        consts::{DERIBIT_VENUE, DERIBIT_WS_HEARTBEAT_SECS},
56        enums::resolve_trigger_type,
57    },
58    config::DeribitExecutionClientConfig,
59    http::{client::DeribitHttpClient, models::DeribitCurrency, query::GetOrderStateParams},
60    websocket::{
61        auth::DERIBIT_EXECUTION_SESSION_NAME,
62        client::DeribitWebSocketClient,
63        messages::{DeribitEditParams, DeribitOrderParams, NautilusWsMessage},
64        parse::parse_user_order_msg,
65    },
66};
67
68/// Deribit live execution client.
69#[derive(Debug)]
70pub struct DeribitExecutionClient {
71    core: ExecutionClientCore,
72    clock: &'static AtomicTime,
73    config: DeribitExecutionClientConfig,
74    emitter: ExecutionEventEmitter,
75    http_client: DeribitHttpClient,
76    ws_client: DeribitWebSocketClient,
77    session_tasks: TaskGroup,
78    pending_tasks: TaskGroup,
79}
80
81impl DeribitExecutionClient {
82    /// Creates a new [`DeribitExecutionClient`].
83    ///
84    /// # Errors
85    ///
86    /// Returns an error if the client fails to initialize.
87    pub fn new(
88        core: ExecutionClientCore,
89        config: DeribitExecutionClientConfig,
90    ) -> anyhow::Result<Self> {
91        let api_key = config
92            .api_key
93            .as_ref()
94            .map(|value| value.expose_secret().to_owned());
95        let api_secret = config
96            .api_secret
97            .as_ref()
98            .map(|value| value.expose_secret().to_owned());
99        let proxy_url = config
100            .proxy_url
101            .as_ref()
102            .map(|value| value.expose_secret().to_owned());
103        let http_client = if config.has_api_credentials() {
104            DeribitHttpClient::new_with_env(
105                api_key.clone(),
106                api_secret.clone(),
107                config.base_url_http.clone(),
108                config.environment,
109                config.http_timeout_secs,
110                config.max_retries,
111                config.retry_delay_initial_ms,
112                config.retry_delay_max_ms,
113                proxy_url.clone(),
114            )?
115        } else {
116            DeribitHttpClient::new(
117                config.base_url_http.clone(),
118                config.environment,
119                config.http_timeout_secs,
120                config.max_retries,
121                config.retry_delay_initial_ms,
122                config.retry_delay_max_ms,
123                proxy_url.clone(),
124            )?
125        };
126
127        let mut ws_client = DeribitWebSocketClient::new(
128            config.base_url_ws.clone(),
129            api_key,
130            api_secret,
131            DERIBIT_WS_HEARTBEAT_SECS,
132            config.auth_timeout_secs,
133            config.environment,
134            config.transport_backend,
135            proxy_url,
136        )
137        .context("failed to create WebSocket client for execution")?
138        .with_socket_control(SocketControl::new(
139            core.client_id,
140            Some(*DERIBIT_VENUE),
141            "deribit-user-streams",
142        ));
143        // Set account ID for order/fill reports
144        ws_client.set_account_id(core.account_id);
145
146        let clock = get_atomic_clock_realtime();
147        let emitter = ExecutionEventEmitter::new(
148            clock,
149            core.trader_id,
150            core.account_id,
151            AccountType::Margin,
152            None,
153        );
154
155        let session_tasks = TaskGroup::new();
156        let pending_tasks = TaskGroup::new();
157
158        Ok(Self {
159            core,
160            clock,
161            config,
162            emitter,
163            http_client,
164            ws_client,
165            session_tasks,
166            pending_tasks,
167        })
168    }
169
170    /// Spawns an async task for execution operations.
171    fn spawn_task<F>(&self, description: &'static str, fut: F)
172    where
173        F: Future<Output = anyhow::Result<()>> + Send + 'static,
174    {
175        let future = async move {
176            if let Err(e) = fut.await {
177                log::warn!("{description} failed: {e:?}");
178            }
179        };
180
181        if let Err(e) = self.pending_tasks.spawn(future) {
182            log::warn!("Skipping Deribit {description} after shutdown began: {e}");
183        }
184    }
185
186    /// Aborts all pending async tasks.
187    fn abort_pending_tasks(&self) {
188        self.pending_tasks.begin_shutdown();
189    }
190
191    fn abort_session_tasks(&self) {
192        self.session_tasks.begin_shutdown();
193        self.ws_client.begin_shutdown();
194    }
195
196    async fn await_pending_tasks(&self) -> anyhow::Result<()> {
197        self.pending_tasks.begin_shutdown();
198        self.pending_tasks
199            .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
200            .await
201            .map_err(|e| anyhow::anyhow!("Failed to terminate Deribit execution tasks: {e}"))?;
202        Ok(())
203    }
204
205    async fn await_session_tasks(&self) -> anyhow::Result<()> {
206        self.session_tasks.begin_shutdown();
207        self.session_tasks
208            .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2))
209            .await
210            .map_err(|e| anyhow::anyhow!("Failed to terminate Deribit session tasks: {e}"))?;
211        Ok(())
212    }
213
214    async fn teardown_partial_connect(&self) -> anyhow::Result<()> {
215        self.abort_session_tasks();
216        self.abort_pending_tasks();
217
218        let mut errors = Vec::new();
219        if let Err(e) = self.ws_client.close().await {
220            errors.push(format!("WebSocket shutdown failed: {e}"));
221        }
222        let (session_result, pending_result) =
223            tokio::join!(self.await_session_tasks(), self.await_pending_tasks());
224
225        if let Err(e) = session_result {
226            errors.push(e.to_string());
227        }
228
229        if let Err(e) = pending_result {
230            errors.push(e.to_string());
231        }
232        self.core.set_disconnected();
233
234        if errors.is_empty() {
235            Ok(())
236        } else {
237            anyhow::bail!(errors.join("; "))
238        }
239    }
240
241    // Rejects unsupported order types, time-in-force values, and params without exact JSON numbers
242    fn build_order_params(order: &dyn Order) -> anyhow::Result<DeribitOrderParams> {
243        let order_type = match order.order_type() {
244            OrderType::Limit => "limit",
245            OrderType::Market => "market",
246            OrderType::StopLimit => "stop_limit",
247            OrderType::StopMarket => "stop_market",
248            OrderType::LimitIfTouched => "take_limit",
249            OrderType::MarketIfTouched => "take_market",
250            other => {
251                anyhow::bail!("Unsupported order type {other:?} for Deribit");
252            }
253        }
254        .to_string();
255
256        let time_in_force = if matches!(
257            order.order_type(),
258            OrderType::Market | OrderType::StopMarket | OrderType::MarketIfTouched
259        ) {
260            // Deribit rejects `time_in_force` on market-style order types
261            None
262        } else {
263            Some(
264                match order.time_in_force() {
265                    TimeInForce::Gtc => "good_til_cancelled",
266                    TimeInForce::Ioc => "immediate_or_cancel",
267                    TimeInForce::Fok => "fill_or_kill",
268                    TimeInForce::Gtd => {
269                        if order.expire_time().is_some() {
270                            log::warn!(
271                                "Deribit GTD orders expire at 8:00 UTC only - custom expire_time is ignored. \
272                                For custom expiry times, use managed GTD with emulation_trigger"
273                            );
274                        }
275                        "good_til_day"
276                    }
277                    other => {
278                        anyhow::bail!("Unsupported time_in_force {other:?} for Deribit");
279                    }
280                }
281                .to_string(),
282            )
283        };
284
285        // Deribit's `valid_until` is a REQUEST timeout, not order expiry.
286        // Deribit's `good_til_day` expires at end of trading session (8 UTC).
287        let valid_until = None;
288
289        let trigger = resolve_trigger_type(order.trigger_type());
290
291        let params = DeribitOrderParams {
292            instrument_name: order.instrument_id().symbol.to_string(),
293            amount: order.quantity().as_decimal(),
294            order_type,
295            label: Some(order.client_order_id().to_string()),
296            price: order.price().map(|p| p.as_decimal()),
297            time_in_force,
298            post_only: if order.is_post_only() {
299                Some(true)
300            } else {
301                None
302            },
303            reject_post_only: if order.is_post_only() {
304                Some(true)
305            } else {
306                None
307            },
308            reduce_only: if order.is_reduce_only() {
309                Some(true)
310            } else {
311                None
312            },
313            trigger_price: order.trigger_price().map(|p| p.as_decimal()),
314            trigger,
315            max_show: None,
316            valid_until,
317        };
318
319        check_params_serialize(&params)?;
320
321        Ok(params)
322    }
323
324    /// Submits a single order to Deribit.
325    ///
326    /// This is the core submission logic shared by `submit_order` and `submit_order_list`.
327    fn submit_single_order(&self, order: &OrderAny, task_name: &'static str) {
328        if order.is_closed() {
329            log::warn!("Cannot submit closed order {}", order.client_order_id());
330            return;
331        }
332
333        let params = match Self::build_order_params(order) {
334            Ok(params) => params,
335            Err(e) => {
336                let ts_event = self.clock.get_time_ns();
337                self.emitter.emit_order_rejected_event(
338                    order.strategy_id(),
339                    order.instrument_id(),
340                    order.client_order_id(),
341                    &format!("{e}"),
342                    ts_event,
343                    false,
344                );
345                return;
346            }
347        };
348        let client_order_id = order.client_order_id();
349        let trader_id = order.trader_id();
350        let strategy_id = order.strategy_id();
351        let instrument_id = order.instrument_id();
352        let order_side = order.order_side();
353
354        log::debug!("OrderSubmitted client_order_id={client_order_id}");
355        self.emitter.emit_order_submitted(order);
356
357        let ws_client = self.ws_client.clone();
358
359        self.spawn_task(task_name, async move {
360            let result = ws_client
361                .submit_order(
362                    order_side,
363                    params,
364                    client_order_id,
365                    trader_id,
366                    strategy_id,
367                    instrument_id,
368                )
369                .await;
370
371            if let Err(e) = result {
372                log::error!(
373                    "Submit order request failed: task={task_name}, client_order_id={client_order_id}, error={e}"
374                );
375                return Err(e.into());
376            }
377
378            Ok(())
379        });
380    }
381
382    /// Spawns a stream handler to dispatch WebSocket messages to the execution engine.
383    fn spawn_stream_handler(
384        &self,
385        stream: impl futures_util::Stream<Item = NautilusWsMessage> + Send + 'static,
386    ) -> anyhow::Result<()> {
387        let emitter = self.emitter.clone();
388
389        self.session_tasks.spawn(async move {
390            pin_mut!(stream);
391            while let Some(message) = stream.next().await {
392                dispatch_ws_message(message, &emitter);
393            }
394        })?;
395
396        log::debug!("WebSocket stream handler started");
397        Ok(())
398    }
399}
400
401#[async_trait(?Send)]
402impl ExecutionClient for DeribitExecutionClient {
403    fn is_connected(&self) -> bool {
404        self.core.is_connected()
405    }
406
407    fn client_id(&self) -> ClientId {
408        self.core.client_id
409    }
410
411    fn account_id(&self) -> AccountId {
412        self.core.account_id
413    }
414
415    fn venue(&self) -> Venue {
416        *DERIBIT_VENUE
417    }
418
419    fn oms_type(&self) -> OmsType {
420        self.core.oms_type
421    }
422
423    fn get_account(&self) -> Option<AccountAny> {
424        self.core.cache().account_owned(&self.core.account_id)
425    }
426
427    fn generate_account_state(
428        &self,
429        balances: Vec<AccountBalance>,
430        margins: Vec<MarginBalance>,
431        reported: bool,
432        ts_event: UnixNanos,
433        info: Option<Params>,
434    ) -> anyhow::Result<()> {
435        self.emitter
436            .emit_account_state(balances, margins, reported, ts_event, info);
437        Ok(())
438    }
439
440    fn start(&mut self) -> anyhow::Result<()> {
441        if self.core.is_started() {
442            return Ok(());
443        }
444
445        let sender = get_exec_event_sender();
446        self.emitter.set_sender(sender);
447        self.core.set_started();
448
449        log::info!(
450            "Started: client_id={}, account_id={}, account_type={:?}, product_types={:?}, environment={}",
451            self.core.client_id,
452            self.core.account_id,
453            self.core.account_type,
454            self.config.product_types,
455            self.config.environment
456        );
457        Ok(())
458    }
459
460    fn stop(&mut self) -> anyhow::Result<()> {
461        if self.core.is_stopped() {
462            return Ok(());
463        }
464
465        self.core.set_stopped();
466        self.core.set_disconnected();
467        self.abort_session_tasks();
468        self.abort_pending_tasks();
469        log::info!("Stopped: client_id={}", self.core.client_id);
470        Ok(())
471    }
472
473    async fn connect(&mut self) -> anyhow::Result<()> {
474        if self.core.is_connected() && self.pending_tasks.is_open() && self.session_tasks.is_open()
475        {
476            return Ok(());
477        }
478
479        if !self.pending_tasks.is_open() {
480            self.await_pending_tasks().await?;
481            self.pending_tasks
482                .start_generation()
483                .map_err(|e| anyhow::anyhow!("Failed to start Deribit task generation: {e}"))?;
484        }
485
486        if !self.session_tasks.is_open() || !self.session_tasks.is_empty() {
487            self.abort_session_tasks();
488
489            if self.ws_client.is_active() {
490                self.ws_client
491                    .close()
492                    .await
493                    .context("failed to close stale Deribit WebSocket")?;
494            }
495            self.await_session_tasks().await?;
496            self.session_tasks
497                .start_generation()
498                .map_err(|e| anyhow::anyhow!("Failed to start Deribit session generation: {e}"))?;
499        } else if self.ws_client.is_active() {
500            self.ws_client
501                .close()
502                .await
503                .context("failed to close stale Deribit WebSocket")?;
504        }
505        let ws_client = self.ws_client.clone();
506        let setup_guard =
507            TaskGroupGuard::new(&[&self.session_tasks, &self.pending_tasks], move || {
508                ws_client.begin_shutdown();
509            });
510
511        // Check if credentials are available before requesting account state
512        if !self.config.has_api_credentials() {
513            anyhow::bail!("Missing API credentials; set Deribit environment variables");
514        }
515
516        // Set account ID for order/fill reports
517        self.ws_client.set_account_id(self.core.account_id);
518
519        // Fetch and cache instruments in both HTTP client and WebSocket client
520        if !self.core.instruments_initialized() {
521            for product_type in &self.config.product_types {
522                let instruments = self
523                    .http_client
524                    .request_instruments(DeribitCurrency::ANY, Some(*product_type))
525                    .await
526                    .with_context(|| {
527                        format!("failed to request instruments for {product_type:?}")
528                    })?;
529
530                if instruments.is_empty() {
531                    log::warn!("No instruments returned for {product_type:?}");
532                    continue;
533                }
534
535                log::debug!("Fetched {} {product_type:?} instruments", instruments.len());
536                self.ws_client.cache_instruments(&instruments);
537                self.http_client.cache_instruments(&instruments);
538            }
539            self.core.set_instruments_initialized();
540        }
541
542        // Fetch initial account state
543        let account_state = self
544            .http_client
545            .request_account_state(self.core.account_id)
546            .await
547            .context("failed to request account state")?;
548
549        self.emitter.send_account_state(account_state);
550
551        let session_result = async {
552            self.ws_client
553                .connect()
554                .await
555                .context("failed to connect WebSocket client for execution")?;
556
557            self.ws_client
558                .authenticate_session(DERIBIT_EXECUTION_SESSION_NAME)
559                .await
560                .map_err(|e| anyhow::anyhow!("failed to authenticate WebSocket session: {e}"))?;
561
562            log::debug!("WebSocket client authenticated for execution");
563
564            // Subscribe to user order and trade updates for all instruments
565            self.ws_client
566                .subscribe_user_orders()
567                .await
568                .map_err(|e| anyhow::anyhow!("failed to subscribe to user orders: {e}"))?;
569            self.ws_client
570                .subscribe_user_trades()
571                .await
572                .map_err(|e| anyhow::anyhow!("failed to subscribe to user trades: {e}"))?;
573            self.ws_client
574                .subscribe_user_portfolio()
575                .await
576                .map_err(|e| anyhow::anyhow!("failed to subscribe to user portfolio: {e}"))?;
577
578            if let Err(e) = self.ws_client.wait_for_subscriptions_confirmed(30.0).await {
579                // Roll back subscription state so a retry re-sends subscribe requests
580                let _ = self.ws_client.unsubscribe_user_orders().await;
581                let _ = self.ws_client.unsubscribe_user_trades().await;
582                let _ = self.ws_client.unsubscribe_user_portfolio().await;
583                anyhow::bail!("subscription confirmation failed: {e}");
584            }
585
586            log::debug!("Subscribed to user order, trade, and portfolio updates");
587
588            // Spawn stream handler to dispatch WebSocket messages to the execution engine
589            let stream = self.ws_client.stream()?;
590            self.spawn_stream_handler(stream)?;
591
592            Ok::<(), anyhow::Error>(())
593        }
594        .await;
595
596        if let Err(e) = session_result {
597            if let Err(teardown_error) = self.teardown_partial_connect().await {
598                return Err(e.context(format!(
599                    "Deribit execution startup teardown failed: {teardown_error}"
600                )));
601            }
602            return Err(e);
603        }
604
605        self.core.set_connected();
606        setup_guard.disarm();
607        log::info!("Connected: client_id={}", self.core.client_id);
608        Ok(())
609    }
610
611    async fn disconnect(&mut self) -> anyhow::Result<()> {
612        self.teardown_partial_connect().await?;
613        log::info!("Disconnected: client_id={}", self.core.client_id);
614        Ok(())
615    }
616
617    async fn generate_order_status_report(
618        &self,
619        cmd: &GenerateOrderStatusReport,
620    ) -> anyhow::Result<Option<OrderStatusReport>> {
621        // If venue_order_id is provided, fetch the specific order by ID
622        if let Some(venue_order_id) = &cmd.venue_order_id {
623            let params = GetOrderStateParams {
624                order_id: venue_order_id.to_string(),
625            };
626            let ts_init = self.clock.get_time_ns();
627
628            match self.http_client.inner.get_order_state(params).await {
629                Ok(response) => {
630                    if let Some(order) = response.result {
631                        let symbol = order.instrument_name;
632                        if let Some(instrument) = self.http_client.get_instrument(&symbol) {
633                            let report = parse_user_order_msg(
634                                &order,
635                                &instrument,
636                                self.core.account_id,
637                                ts_init,
638                            )?;
639                            return Ok(Some(report));
640                        } else {
641                            log::warn!(
642                                "Instrument {} not in cache for order {}",
643                                order.instrument_name,
644                                order.order_id
645                            );
646                        }
647                    }
648                }
649                Err(e) => {
650                    log::warn!("Failed to get order state: {e}");
651                }
652            }
653            return Ok(None);
654        }
655
656        // If client_order_id is provided, search open then closed orders
657        if let Some(client_order_id) = &cmd.client_order_id {
658            let reports = self
659                .http_client
660                .request_order_status_reports(
661                    self.core.account_id,
662                    cmd.instrument_id,
663                    None,
664                    None,
665                    false, // search all orders, not just open
666                )
667                .await?;
668
669            // Filter by client_order_id
670            for report in reports {
671                if report.client_order_id == Some(*client_order_id) {
672                    return Ok(Some(report));
673                }
674            }
675        }
676
677        Ok(None)
678    }
679
680    async fn generate_order_status_reports(
681        &self,
682        cmd: &GenerateOrderStatusReports,
683    ) -> anyhow::Result<Vec<OrderStatusReport>> {
684        self.http_client
685            .request_order_status_reports(
686                self.core.account_id,
687                cmd.instrument_id,
688                cmd.start,
689                cmd.end,
690                cmd.open_only,
691            )
692            .await
693    }
694
695    async fn generate_fill_reports(
696        &self,
697        cmd: GenerateFillReports,
698    ) -> anyhow::Result<Vec<FillReport>> {
699        let mut reports = self
700            .http_client
701            .request_fill_reports(self.core.account_id, cmd.instrument_id, cmd.start, cmd.end)
702            .await?;
703
704        // Filter by venue_order_id if provided
705        if let Some(venue_order_id) = &cmd.venue_order_id {
706            reports.retain(|r| r.venue_order_id == *venue_order_id);
707        }
708
709        Ok(reports)
710    }
711
712    async fn generate_position_status_reports(
713        &self,
714        cmd: &GeneratePositionStatusReports,
715    ) -> anyhow::Result<Vec<PositionStatusReport>> {
716        self.http_client
717            .request_position_status_reports(self.core.account_id, cmd.instrument_id)
718            .await
719    }
720
721    async fn generate_mass_status(
722        &self,
723        lookback_mins: Option<u64>,
724    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
725        log::info!("Generating ExecutionMassStatus (lookback_mins={lookback_mins:?})");
726        let ts_now = self.clock.get_time_ns();
727        let start = lookback_mins
728            .map(DurationNanos::try_from_mins)
729            .transpose()?
730            .map(|lookback| ts_now.saturating_sub(lookback));
731
732        let order_cmd = GenerateOrderStatusReportsBuilder::default()
733            .ts_init(ts_now)
734            .open_only(false) // get all orders for mass status
735            .start(start)
736            .build()
737            .context("Failed to build GenerateOrderStatusReports")?;
738
739        let fill_cmd = GenerateFillReportsBuilder::default()
740            .ts_init(ts_now)
741            .start(start)
742            .build()
743            .context("Failed to build GenerateFillReports")?;
744
745        let position_cmd = GeneratePositionStatusReportsBuilder::default()
746            .ts_init(ts_now)
747            .start(start)
748            .build()
749            .context("Failed to build GeneratePositionStatusReports")?;
750
751        let (order_reports, fill_reports, position_reports) = tokio::try_join!(
752            self.generate_order_status_reports(&order_cmd),
753            self.generate_fill_reports(fill_cmd),
754            self.generate_position_status_reports(&position_cmd),
755        )?;
756
757        log::info!("Received {} OrderStatusReports", order_reports.len());
758        log::info!("Received {} FillReports", fill_reports.len());
759        log::info!("Received {} PositionReports", position_reports.len());
760
761        let mut mass_status = ExecutionMassStatus::new(
762            self.core.client_id,
763            self.core.account_id,
764            *DERIBIT_VENUE,
765            ts_now,
766            None,
767        );
768
769        mass_status.add_order_reports(order_reports);
770        mass_status.add_fill_reports(fill_reports);
771        mass_status.add_position_reports(position_reports);
772
773        Ok(Some(mass_status))
774    }
775
776    fn query_account(&self, _cmd: QueryAccount) -> anyhow::Result<()> {
777        let http_client = self.http_client.clone();
778        let account_id = self.core.account_id;
779        let emitter = self.emitter.clone();
780
781        self.spawn_task("query_account", async move {
782            let account_state = http_client
783                .request_account_state(account_id)
784                .await
785                .context("failed to query account state (check API credentials are valid)")?;
786
787            emitter.send_account_state(account_state);
788            Ok(())
789        });
790
791        Ok(())
792    }
793
794    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
795        let ws_client = self.ws_client.clone();
796
797        // Extract venue order ID (Deribit's order_id)
798        let order_id = cmd
799            .venue_order_id
800            .as_ref()
801            .ok_or_else(|| anyhow::anyhow!("venue_order_id required for query_order"))?
802            .to_string();
803
804        let client_order_id = cmd.client_order_id;
805        let trader_id = cmd.trader_id;
806        let strategy_id = cmd.strategy_id;
807        let instrument_id = cmd.instrument_id;
808
809        log::debug!("Querying order state: order_id={order_id}, client_order_id={client_order_id}");
810
811        // Spawn async task to query order state via WebSocket
812        // Response will be dispatched through the WebSocket stream handler as OrderStatusReport
813        self.spawn_task("query_order", async move {
814            ws_client
815                .query_order(
816                    &order_id,
817                    client_order_id,
818                    trader_id,
819                    strategy_id,
820                    instrument_id,
821                )
822                .await
823                .map_err(|e| anyhow::anyhow!("Query order state failed: {e}"))?;
824            Ok(())
825        });
826
827        Ok(())
828    }
829
830    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
831        let order = self.core.cache().try_order_owned(&cmd.client_order_id)?;
832        self.submit_single_order(&order, "submit_order");
833        Ok(())
834    }
835
836    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
837        if cmd.order_list.client_order_ids.is_empty() {
838            log::debug!("submit_order_list called with empty order list");
839            return Ok(());
840        }
841
842        let orders = self.core.get_orders_for_list(&cmd.order_list)?;
843
844        log::debug!(
845            "Submitting order list {} with {} orders for instrument={}",
846            cmd.order_list.id,
847            orders.len(),
848            cmd.instrument_id
849        );
850
851        // Deribit doesn't have native batch order submission
852        // Loop through and submit each order individually using shared logic
853        for order in &orders {
854            self.submit_single_order(order, "submit_order_list_item");
855        }
856
857        Ok(())
858    }
859
860    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
861        let ws_client = self.ws_client.clone();
862
863        // Extract venue order ID (Deribit's order_id)
864        let order_id = match cmd.venue_order_id.as_ref() {
865            Some(venue_order_id) => venue_order_id.to_string(),
866            None => {
867                return reject_modify_command(
868                    &self.emitter,
869                    self.clock,
870                    &cmd,
871                    "venue_order_id required for modify_order",
872                );
873            }
874        };
875
876        // Extract quantity - if not provided, get from order in cache
877        let quantity = if let Some(qty) = cmd.quantity {
878            qty
879        } else {
880            // Get order from cache to use its current quantity
881            let cache = self.core.cache();
882            match cache.order(&cmd.client_order_id) {
883                Some(order) => order.quantity(),
884                None => {
885                    return reject_modify_command(
886                        &self.emitter,
887                        self.clock,
888                        &cmd,
889                        &format!("Order not found: {}", cmd.client_order_id),
890                    );
891                }
892            }
893        };
894
895        let price = match cmd.price {
896            Some(price) => price,
897            None => {
898                return reject_modify_command(
899                    &self.emitter,
900                    self.clock,
901                    &cmd,
902                    "price required for modify_order",
903                );
904            }
905        };
906
907        let client_order_id = cmd.client_order_id;
908        let trader_id = cmd.trader_id;
909        let strategy_id = cmd.strategy_id;
910        let instrument_id = cmd.instrument_id;
911
912        log::debug!(
913            "Modifying order: order_id={order_id}, quantity={quantity}, price={price}, client_order_id={client_order_id}"
914        );
915
916        let params = DeribitEditParams {
917            order_id: order_id.clone(),
918            amount: quantity.as_decimal(),
919            price: Some(price.as_decimal()),
920            post_only: None,
921            reject_post_only: None,
922            reduce_only: None,
923            trigger_price: None,
924        };
925
926        if let Err(e) = check_params_serialize(&params) {
927            return reject_modify_command(&self.emitter, self.clock, &cmd, &e.to_string());
928        }
929
930        // Spawn async task to send modify via WebSocket
931        self.spawn_task("modify_order", async move {
932            if let Err(e) = ws_client
933                .modify_order(
934                    params,
935                    client_order_id,
936                    trader_id,
937                    strategy_id,
938                    instrument_id,
939                )
940                .await
941            {
942                log::error!(
943                    "Modify order failed: order_id={order_id}, client_order_id={client_order_id}, error={e}"
944                );
945                anyhow::bail!("Modify order failed: {e}");
946            }
947            Ok(())
948        });
949
950        Ok(())
951    }
952
953    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
954        let ws_client = self.ws_client.clone();
955
956        // Extract venue order ID (Deribit's order_id)
957        let order_id = match cmd.venue_order_id.as_ref() {
958            Some(venue_order_id) => venue_order_id.to_string(),
959            None => {
960                log::warn!(
961                    "Cannot cancel order {} - no venue_order_id",
962                    cmd.client_order_id
963                );
964                return Ok(());
965            }
966        };
967
968        let client_order_id = cmd.client_order_id;
969        let trader_id = cmd.trader_id;
970        let strategy_id = cmd.strategy_id;
971        let instrument_id = cmd.instrument_id;
972
973        log::debug!("Canceling order: order_id={order_id}, client_order_id={client_order_id}");
974
975        // Spawn async task to send cancel via WebSocket
976        self.spawn_task("cancel_order", async move {
977            if let Err(e) = ws_client
978                .cancel_order(
979                    &order_id,
980                    client_order_id,
981                    trader_id,
982                    strategy_id,
983                    instrument_id,
984                )
985                .await
986            {
987                log::error!(
988                    "Cancel order failed: order_id={order_id}, client_order_id={client_order_id}, error={e}"
989                );
990                anyhow::bail!("Cancel order failed: {e}");
991            }
992            Ok(())
993        });
994
995        Ok(())
996    }
997
998    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
999        let instrument_id = cmd.instrument_id;
1000
1001        // Without a side filter, use efficient bulk cancel via Deribit API
1002        let Some(order_side) = cmd.order_side else {
1003            log::debug!(
1004                "Cancelling all orders: instrument={instrument_id}, order_side=None (bulk)"
1005            );
1006
1007            let ws_client = self.ws_client.clone();
1008            self.spawn_task("cancel_all_orders", async move {
1009                if let Err(e) = ws_client.cancel_all_orders(instrument_id, None).await {
1010                    log::error!("Cancel all orders failed for instrument {instrument_id}: {e}");
1011                    anyhow::bail!("Cancel all orders failed: {e}");
1012                }
1013                Ok(())
1014            });
1015
1016            return Ok(());
1017        };
1018
1019        // For specific side (Buy/Sell), filter from cache and cancel individually
1020        // Deribit API doesn't support side filtering, so we implement it locally
1021        log::debug!(
1022            "Cancelling orders by side: instrument={instrument_id}, order_side={order_side}"
1023        );
1024
1025        let orders_to_cancel: Vec<_> = {
1026            let cache = self.core.cache();
1027            let open_orders = cache.orders_open(None, Some(&instrument_id), None, None, None);
1028
1029            open_orders
1030                .into_iter()
1031                .filter(|order| order.order_side() == order_side)
1032                .filter_map(|order| {
1033                    let venue_order_id = order.venue_order_id()?;
1034                    Some((
1035                        venue_order_id.to_string(),
1036                        order.client_order_id(),
1037                        order.instrument_id(),
1038                        Some(venue_order_id),
1039                    ))
1040                })
1041                .collect()
1042        };
1043
1044        if orders_to_cancel.is_empty() {
1045            log::debug!("No open {order_side} orders to cancel for {instrument_id}");
1046            return Ok(());
1047        }
1048
1049        log::debug!(
1050            "Cancelling {} {order_side} orders for {instrument_id}",
1051            orders_to_cancel.len(),
1052        );
1053
1054        // Cancel each matching order individually
1055        for (venue_order_id_str, client_order_id, order_instrument_id, _venue_order_id) in
1056            orders_to_cancel
1057        {
1058            let ws_client = self.ws_client.clone();
1059            let trader_id = cmd.trader_id;
1060            let strategy_id = cmd.strategy_id;
1061
1062            self.spawn_task("cancel_order_by_side", async move {
1063                if let Err(e) = ws_client
1064                    .cancel_order(
1065                        &venue_order_id_str,
1066                        client_order_id,
1067                        trader_id,
1068                        strategy_id,
1069                        order_instrument_id,
1070                    )
1071                    .await
1072                {
1073                    log::error!(
1074                        "Cancel order failed: order_id={venue_order_id_str}, client_order_id={client_order_id}, error={e}"
1075                    );
1076                }
1077                Ok(())
1078            });
1079        }
1080
1081        Ok(())
1082    }
1083
1084    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
1085        if cmd.cancels.is_empty() {
1086            log::debug!("batch_cancel_orders called with empty cancels list");
1087            return Ok(());
1088        }
1089
1090        log::debug!(
1091            "Batch cancelling {} orders for instrument={}",
1092            cmd.cancels.len(),
1093            cmd.instrument_id
1094        );
1095
1096        // Deribit doesn't have native batch cancel by order ID
1097        // Loop through and cancel each order individually
1098        for cancel in &cmd.cancels {
1099            let order_id = match &cancel.venue_order_id {
1100                Some(id) => id.to_string(),
1101                None => {
1102                    log::warn!(
1103                        "Cannot cancel order {} - no venue_order_id",
1104                        cancel.client_order_id
1105                    );
1106                    continue;
1107                }
1108            };
1109
1110            let ws_client = self.ws_client.clone();
1111            let client_order_id = cancel.client_order_id;
1112            let trader_id = cancel.trader_id;
1113            let strategy_id = cancel.strategy_id;
1114            let instrument_id = cancel.instrument_id;
1115
1116            self.spawn_task("batch_cancel_order", async move {
1117                if let Err(e) = ws_client
1118                    .cancel_order(
1119                        &order_id,
1120                        client_order_id,
1121                        trader_id,
1122                        strategy_id,
1123                        instrument_id,
1124                    )
1125                    .await
1126                {
1127                    log::error!(
1128                        "Batch cancel order failed: order_id={order_id}, client_order_id={client_order_id}, error={e}"
1129                    );
1130                    anyhow::bail!("Batch cancel order failed: {e}");
1131                }
1132                Ok(())
1133            });
1134        }
1135
1136        Ok(())
1137    }
1138}
1139
1140/// Dispatches a WebSocket message using the event emitter.
1141fn dispatch_ws_message(message: NautilusWsMessage, emitter: &ExecutionEventEmitter) {
1142    match message {
1143        NautilusWsMessage::AccountState(state) => {
1144            emitter.send_account_state(state);
1145        }
1146        NautilusWsMessage::OrderStatusReports(reports) => {
1147            log::debug!("Processing {} order status report(s)", reports.len());
1148            for report in reports {
1149                emitter.send_order_status_report(report);
1150            }
1151        }
1152        NautilusWsMessage::FillReports(reports) => {
1153            log::debug!("Processing {} fill report(s)", reports.len());
1154            for report in reports {
1155                emitter.send_fill_report(report);
1156            }
1157        }
1158        NautilusWsMessage::OrderFilled(event) => {
1159            emitter.send_order_event(OrderEventAny::Filled(event));
1160        }
1161        NautilusWsMessage::OrderRejected(event) => {
1162            emitter.send_order_event(OrderEventAny::Rejected(event));
1163        }
1164        NautilusWsMessage::OrderAccepted(event) => {
1165            emitter.send_order_event(OrderEventAny::Accepted(event));
1166        }
1167        NautilusWsMessage::OrderCanceled(event) => {
1168            emitter.send_order_event(OrderEventAny::Canceled(event));
1169        }
1170        NautilusWsMessage::OrderExpired(event) => {
1171            emitter.send_order_event(OrderEventAny::Expired(event));
1172        }
1173        NautilusWsMessage::OrderUpdated(event) => {
1174            emitter.send_order_event(OrderEventAny::Updated(event));
1175        }
1176        NautilusWsMessage::OrderCancelRejected(event) => {
1177            emitter.send_order_event(OrderEventAny::CancelRejected(event));
1178        }
1179        NautilusWsMessage::OrderModifyRejected(event) => {
1180            emitter.send_order_event(OrderEventAny::ModifyRejected(event));
1181        }
1182        NautilusWsMessage::Error(e) => {
1183            log::warn!("WebSocket error: {e}");
1184        }
1185        NautilusWsMessage::Reconnected => {
1186            log::info!("WebSocket reconnected");
1187        }
1188        NautilusWsMessage::Authenticated(auth) => {
1189            log::debug!("WebSocket authenticated: scope={}", auth.scope);
1190        }
1191        NautilusWsMessage::AuthenticationFailed(reason) => {
1192            log::error!("Authentication failed in execution client: {reason}");
1193        }
1194        NautilusWsMessage::Data(_)
1195        | NautilusWsMessage::Deltas(_)
1196        | NautilusWsMessage::Instrument(_)
1197        | NautilusWsMessage::InstrumentStatus(_)
1198        | NautilusWsMessage::FundingRates(_)
1199        | NautilusWsMessage::OptionGreeks(_)
1200        | NautilusWsMessage::Raw(_) => {
1201            // Data messages are handled by the data client, not execution
1202            log::trace!("Ignoring data message in execution client");
1203        }
1204    }
1205}
1206
1207fn reject_modify_command(
1208    emitter: &ExecutionEventEmitter,
1209    clock: &AtomicTime,
1210    cmd: &ModifyOrder,
1211    reason: &str,
1212) -> anyhow::Result<()> {
1213    let ts_event = clock.get_time_ns();
1214    emitter.emit_order_modify_rejected_event(
1215        cmd.strategy_id,
1216        cmd.instrument_id,
1217        cmd.client_order_id,
1218        cmd.venue_order_id,
1219        reason,
1220        ts_event,
1221    );
1222    anyhow::bail!("{reason}");
1223}
1224
1225// Serializes the params as the handler will, so a decimal without an exact JSON number is rejected
1226// locally instead of failing after the order is submitted.
1227fn check_params_serialize<T: Serialize>(params: &T) -> anyhow::Result<()> {
1228    serde_json::to_writer(io::sink(), params)
1229        .map_err(|e| anyhow::anyhow!("Cannot serialize order params: {e}"))
1230}
1231
1232#[cfg(test)]
1233mod tests {
1234    use nautilus_common::messages::{ExecutionEvent, execution::ExecutionReport};
1235    use nautilus_core::UUID4;
1236    use nautilus_model::{
1237        enums::{LiquiditySide, OrderSide},
1238        events::OrderFilled,
1239        identifiers::{ClientOrderId, InstrumentId, StrategyId, TradeId, TraderId, VenueOrderId},
1240        types::{Currency, Money, Price, Quantity},
1241    };
1242    use rstest::rstest;
1243
1244    use super::*;
1245
1246    fn dispatch_test_rig() -> (
1247        ExecutionEventEmitter,
1248        tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
1249    ) {
1250        let trader_id = TraderId::from("TRADER-001");
1251        let account_id = AccountId::from("DERIBIT-001");
1252        let mut emitter = ExecutionEventEmitter::new(
1253            get_atomic_clock_realtime(),
1254            trader_id,
1255            account_id,
1256            AccountType::Margin,
1257            None,
1258        );
1259        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
1260        emitter.set_sender(tx);
1261        (emitter, rx)
1262    }
1263
1264    fn fill_report() -> FillReport {
1265        FillReport::new(
1266            AccountId::from("DERIBIT-001"),
1267            InstrumentId::from("BTC-PERPETUAL.DERIBIT"),
1268            VenueOrderId::from("ETH-584830574"),
1269            TradeId::from("ETH-2696068"),
1270            OrderSide::Buy,
1271            Quantity::from("1.000000"),
1272            Price::from("203.80"),
1273            Money::from("0.00036801 USDT"),
1274            LiquiditySide::Taker,
1275            Some(ClientOrderId::from("O-19700101-000000-001-001-1")),
1276            None,
1277            UnixNanos::from(2),
1278            UnixNanos::from(3),
1279            None,
1280        )
1281    }
1282
1283    #[rstest]
1284    fn dispatch_tracked_fill_uses_order_event_path() {
1285        let (emitter, mut rx) = dispatch_test_rig();
1286        let report = fill_report();
1287        let filled = OrderFilled::new(
1288            TraderId::from("TRADER-001"),
1289            StrategyId::from("S-001"),
1290            report.instrument_id,
1291            report.client_order_id.unwrap(),
1292            report.venue_order_id,
1293            report.account_id,
1294            report.trade_id,
1295            report.order_side,
1296            OrderType::Market,
1297            report.last_qty,
1298            report.last_px,
1299            Currency::USDT(),
1300            report.liquidity_side,
1301            UUID4::new(),
1302            report.ts_event,
1303            report.ts_init,
1304            false,
1305            None,
1306            Some(report.commission),
1307            None,
1308        );
1309
1310        dispatch_ws_message(NautilusWsMessage::OrderFilled(filled), &emitter);
1311
1312        assert!(matches!(
1313            rx.try_recv().unwrap(),
1314            ExecutionEvent::Order(OrderEventAny::Filled(event))
1315                if event.client_order_id == ClientOrderId::from("O-19700101-000000-001-001-1")
1316                    && event.trade_id == TradeId::from("ETH-2696068")
1317        ));
1318        assert!(rx.try_recv().is_err());
1319    }
1320
1321    #[rstest]
1322    fn dispatch_untracked_fill_keeps_report_path() {
1323        let (emitter, mut rx) = dispatch_test_rig();
1324        let report = fill_report();
1325
1326        dispatch_ws_message(NautilusWsMessage::FillReports(vec![report]), &emitter);
1327
1328        assert!(matches!(
1329            rx.try_recv().unwrap(),
1330            ExecutionEvent::Report(ExecutionReport::Fill(_))
1331        ));
1332        assert!(rx.try_recv().is_err());
1333    }
1334}
nautilus_deribit/execution.rs

nautilus_deribit/
execution.rs
