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, sync::Mutex};
19
20use anyhow::Context;
21use async_trait::async_trait;
22use futures_util::{StreamExt, pin_mut};
23use nautilus_common::{
24    clients::ExecutionClient,
25    live::{get_runtime, 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    MUTEX_POISONED, UnixNanos,
36    datetime::NANOSECONDS_IN_SECOND,
37    time::{AtomicTime, get_atomic_clock_realtime},
38};
39use nautilus_live::{ExecutionClientCore, ExecutionEventEmitter};
40use nautilus_model::{
41    accounts::AccountAny,
42    enums::{AccountType, OmsType, OrderSide, OrderType, TimeInForce},
43    events::OrderEventAny,
44    identifiers::{AccountId, ClientId, Venue},
45    orders::{Order, OrderAny},
46    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
47    types::{AccountBalance, MarginBalance},
48};
49use tokio::task::JoinHandle;
50
51use crate::{
52    common::{
53        consts::{DERIBIT_VENUE, DERIBIT_WS_HEARTBEAT_SECS},
54        enums::resolve_trigger_type,
55    },
56    config::DeribitExecClientConfig,
57    http::{client::DeribitHttpClient, models::DeribitCurrency, query::GetOrderStateParams},
58    websocket::{
59        auth::DERIBIT_EXECUTION_SESSION_NAME,
60        client::DeribitWebSocketClient,
61        messages::{DeribitOrderParams, NautilusWsMessage},
62        parse::parse_user_order_msg,
63    },
64};
65
66/// Deribit live execution client.
67#[derive(Debug)]
68pub struct DeribitExecutionClient {
69    core: ExecutionClientCore,
70    clock: &'static AtomicTime,
71    config: DeribitExecClientConfig,
72    emitter: ExecutionEventEmitter,
73    http_client: DeribitHttpClient,
74    ws_client: DeribitWebSocketClient,
75    ws_stream_handle: Option<JoinHandle<()>>,
76    pending_tasks: Mutex<Vec<JoinHandle<()>>>,
77}
78
79impl DeribitExecutionClient {
80    /// Creates a new [`DeribitExecutionClient`].
81    ///
82    /// # Errors
83    ///
84    /// Returns an error if the client fails to initialize.
85    pub fn new(core: ExecutionClientCore, config: DeribitExecClientConfig) -> anyhow::Result<Self> {
86        let http_client = if config.has_api_credentials() {
87            DeribitHttpClient::new_with_env(
88                config.api_key.clone(),
89                config.api_secret.clone(),
90                config.base_url_http.clone(),
91                config.environment,
92                config.http_timeout_secs,
93                config.max_retries,
94                config.retry_delay_initial_ms,
95                config.retry_delay_max_ms,
96                config.proxy_url.clone(),
97            )?
98        } else {
99            DeribitHttpClient::new(
100                config.base_url_http.clone(),
101                config.environment,
102                config.http_timeout_secs,
103                config.max_retries,
104                config.retry_delay_initial_ms,
105                config.retry_delay_max_ms,
106                config.proxy_url.clone(),
107            )?
108        };
109
110        let mut ws_client = DeribitWebSocketClient::new(
111            config.base_url_ws.clone(),
112            config.api_key.clone(),
113            config.api_secret.clone(),
114            DERIBIT_WS_HEARTBEAT_SECS,
115            config.environment,
116            config.transport_backend,
117            config.proxy_url.clone(),
118        )
119        .context("failed to create WebSocket client for execution")?;
120        // Set account ID for order/fill reports
121        ws_client.set_account_id(core.account_id);
122
123        let clock = get_atomic_clock_realtime();
124        let emitter = ExecutionEventEmitter::new(
125            clock,
126            core.trader_id,
127            core.account_id,
128            AccountType::Margin,
129            None,
130        );
131
132        Ok(Self {
133            core,
134            clock,
135            config,
136            emitter,
137            http_client,
138            ws_client,
139            ws_stream_handle: None,
140            pending_tasks: Mutex::new(Vec::new()),
141        })
142    }
143
144    /// Spawns an async task for execution operations.
145    fn spawn_task<F>(&self, description: &'static str, fut: F)
146    where
147        F: Future<Output = anyhow::Result<()>> + Send + 'static,
148    {
149        let runtime = get_runtime();
150        let handle = runtime.spawn(async move {
151            if let Err(e) = fut.await {
152                log::warn!("{description} failed: {e:?}");
153            }
154        });
155
156        let mut tasks = self.pending_tasks.lock().expect(MUTEX_POISONED);
157        tasks.retain(|handle| !handle.is_finished());
158        tasks.push(handle);
159    }
160
161    /// Aborts all pending async tasks.
162    fn abort_pending_tasks(&self) {
163        let mut tasks = self.pending_tasks.lock().expect(MUTEX_POISONED);
164        for handle in tasks.drain(..) {
165            handle.abort();
166        }
167    }
168
169    // Rejects unsupported order types and time-in-force values
170    fn build_order_params(order: &dyn Order) -> anyhow::Result<DeribitOrderParams> {
171        let order_type = match order.order_type() {
172            OrderType::Limit => "limit",
173            OrderType::Market => "market",
174            OrderType::StopLimit => "stop_limit",
175            OrderType::StopMarket => "stop_market",
176            OrderType::LimitIfTouched => "take_limit",
177            OrderType::MarketIfTouched => "take_market",
178            other => {
179                anyhow::bail!("Unsupported order type {other:?} for Deribit");
180            }
181        }
182        .to_string();
183
184        let time_in_force = if matches!(
185            order.order_type(),
186            OrderType::Market | OrderType::StopMarket | OrderType::MarketIfTouched
187        ) {
188            // Deribit rejects `time_in_force` on market-style order types
189            None
190        } else {
191            Some(
192                match order.time_in_force() {
193                    TimeInForce::Gtc => "good_til_cancelled",
194                    TimeInForce::Ioc => "immediate_or_cancel",
195                    TimeInForce::Fok => "fill_or_kill",
196                    TimeInForce::Gtd => {
197                        if order.expire_time().is_some() {
198                            log::warn!(
199                                "Deribit GTD orders expire at 8:00 UTC only - custom expire_time is ignored. \
200                                For custom expiry times, use managed GTD with emulation_trigger"
201                            );
202                        }
203                        "good_til_day"
204                    }
205                    other => {
206                        anyhow::bail!("Unsupported time_in_force {other:?} for Deribit");
207                    }
208                }
209                .to_string(),
210            )
211        };
212
213        // Deribit's `valid_until` is a REQUEST timeout, not order expiry.
214        // Deribit's `good_til_day` expires at end of trading session (8 UTC).
215        let valid_until = None;
216
217        let trigger = resolve_trigger_type(order.trigger_type());
218
219        Ok(DeribitOrderParams {
220            instrument_name: order.instrument_id().symbol.to_string(),
221            amount: order.quantity().as_decimal(),
222            order_type,
223            label: Some(order.client_order_id().to_string()),
224            price: order.price().map(|p| p.as_decimal()),
225            time_in_force,
226            post_only: if order.is_post_only() {
227                Some(true)
228            } else {
229                None
230            },
231            reject_post_only: if order.is_post_only() {
232                Some(true)
233            } else {
234                None
235            },
236            reduce_only: if order.is_reduce_only() {
237                Some(true)
238            } else {
239                None
240            },
241            trigger_price: order.trigger_price().map(|p| p.as_decimal()),
242            trigger,
243            max_show: None,
244            valid_until,
245        })
246    }
247
248    /// Submits a single order to Deribit.
249    ///
250    /// This is the core submission logic shared by `submit_order` and `submit_order_list`.
251    fn submit_single_order(&self, order: &OrderAny, task_name: &'static str) {
252        if order.is_closed() {
253            log::warn!("Cannot submit closed order {}", order.client_order_id());
254            return;
255        }
256
257        let params = match Self::build_order_params(order) {
258            Ok(params) => params,
259            Err(e) => {
260                let ts_event = self.clock.get_time_ns();
261                self.emitter.emit_order_rejected_event(
262                    order.strategy_id(),
263                    order.instrument_id(),
264                    order.client_order_id(),
265                    &format!("{e}"),
266                    ts_event,
267                    false,
268                );
269                return;
270            }
271        };
272        let client_order_id = order.client_order_id();
273        let trader_id = order.trader_id();
274        let strategy_id = order.strategy_id();
275        let instrument_id = order.instrument_id();
276        let order_side = order.order_side();
277
278        log::debug!("OrderSubmitted client_order_id={client_order_id}");
279        self.emitter.emit_order_submitted(order);
280
281        let ws_client = self.ws_client.clone();
282
283        self.spawn_task(task_name, async move {
284            let result = ws_client
285                .submit_order(
286                    order_side,
287                    params,
288                    client_order_id,
289                    trader_id,
290                    strategy_id,
291                    instrument_id,
292                )
293                .await;
294
295            if let Err(e) = result {
296                log::error!(
297                    "Submit order request failed: task={task_name}, client_order_id={client_order_id}, error={e}"
298                );
299                return Err(e.into());
300            }
301
302            Ok(())
303        });
304    }
305
306    /// Spawns a stream handler to dispatch WebSocket messages to the execution engine.
307    fn spawn_stream_handler(
308        &mut self,
309        stream: impl futures_util::Stream<Item = NautilusWsMessage> + Send + 'static,
310    ) {
311        if self.ws_stream_handle.is_some() {
312            return;
313        }
314
315        let emitter = self.emitter.clone();
316
317        let handle = get_runtime().spawn(async move {
318            pin_mut!(stream);
319            while let Some(message) = stream.next().await {
320                dispatch_ws_message(message, &emitter);
321            }
322        });
323
324        self.ws_stream_handle = Some(handle);
325        log::debug!("WebSocket stream handler started");
326    }
327}
328
329#[async_trait(?Send)]
330impl ExecutionClient for DeribitExecutionClient {
331    fn is_connected(&self) -> bool {
332        self.core.is_connected()
333    }
334
335    fn client_id(&self) -> ClientId {
336        self.core.client_id
337    }
338
339    fn account_id(&self) -> AccountId {
340        self.core.account_id
341    }
342
343    fn venue(&self) -> Venue {
344        *DERIBIT_VENUE
345    }
346
347    fn oms_type(&self) -> OmsType {
348        self.core.oms_type
349    }
350
351    fn get_account(&self) -> Option<AccountAny> {
352        self.core.cache().account_owned(&self.core.account_id)
353    }
354
355    fn generate_account_state(
356        &self,
357        balances: Vec<AccountBalance>,
358        margins: Vec<MarginBalance>,
359        reported: bool,
360        ts_event: UnixNanos,
361    ) -> anyhow::Result<()> {
362        self.emitter
363            .emit_account_state(balances, margins, reported, ts_event);
364        Ok(())
365    }
366
367    fn start(&mut self) -> anyhow::Result<()> {
368        if self.core.is_started() {
369            return Ok(());
370        }
371
372        let sender = get_exec_event_sender();
373        self.emitter.set_sender(sender);
374        self.core.set_started();
375
376        log::info!(
377            "Started: client_id={}, account_id={}, account_type={:?}, product_types={:?}, environment={}",
378            self.core.client_id,
379            self.core.account_id,
380            self.core.account_type,
381            self.config.product_types,
382            self.config.environment
383        );
384        Ok(())
385    }
386
387    fn stop(&mut self) -> anyhow::Result<()> {
388        if self.core.is_stopped() {
389            return Ok(());
390        }
391
392        self.core.set_stopped();
393        self.core.set_disconnected();
394        self.abort_pending_tasks();
395        log::info!("Stopped: client_id={}", self.core.client_id);
396        Ok(())
397    }
398
399    async fn connect(&mut self) -> anyhow::Result<()> {
400        if self.core.is_connected() {
401            return Ok(());
402        }
403
404        // Check if credentials are available before requesting account state
405        if !self.config.has_api_credentials() {
406            anyhow::bail!("Missing API credentials; set Deribit environment variables");
407        }
408
409        // Set account ID for order/fill reports
410        self.ws_client.set_account_id(self.core.account_id);
411
412        // Fetch and cache instruments in both HTTP client and WebSocket client
413        if !self.core.instruments_initialized() {
414            for product_type in &self.config.product_types {
415                let instruments = self
416                    .http_client
417                    .request_instruments(DeribitCurrency::ANY, Some(*product_type))
418                    .await
419                    .with_context(|| {
420                        format!("failed to request instruments for {product_type:?}")
421                    })?;
422
423                if instruments.is_empty() {
424                    log::warn!("No instruments returned for {product_type:?}");
425                    continue;
426                }
427
428                log::debug!("Fetched {} {product_type:?} instruments", instruments.len());
429                self.ws_client.cache_instruments(&instruments);
430                self.http_client.cache_instruments(&instruments);
431            }
432            self.core.set_instruments_initialized();
433        }
434
435        // Fetch initial account state
436        let account_state = self
437            .http_client
438            .request_account_state(self.core.account_id)
439            .await
440            .context("failed to request account state")?;
441
442        self.emitter.send_account_state(account_state);
443
444        self.ws_client
445            .connect()
446            .await
447            .context("failed to connect WebSocket client for execution")?;
448
449        self.ws_client
450            .authenticate_session(DERIBIT_EXECUTION_SESSION_NAME)
451            .await
452            .map_err(|e| anyhow::anyhow!("failed to authenticate WebSocket session: {e}"))?;
453
454        log::debug!("WebSocket client authenticated for execution");
455
456        // Subscribe to user order and trade updates for all instruments
457        self.ws_client
458            .subscribe_user_orders()
459            .await
460            .map_err(|e| anyhow::anyhow!("failed to subscribe to user orders: {e}"))?;
461        self.ws_client
462            .subscribe_user_trades()
463            .await
464            .map_err(|e| anyhow::anyhow!("failed to subscribe to user trades: {e}"))?;
465        self.ws_client
466            .subscribe_user_portfolio()
467            .await
468            .map_err(|e| anyhow::anyhow!("failed to subscribe to user portfolio: {e}"))?;
469
470        if let Err(e) = self.ws_client.wait_for_subscriptions_confirmed(30.0).await {
471            // Roll back subscription state so a retry re-sends subscribe requests
472            let _ = self.ws_client.unsubscribe_user_orders().await;
473            let _ = self.ws_client.unsubscribe_user_trades().await;
474            let _ = self.ws_client.unsubscribe_user_portfolio().await;
475            anyhow::bail!("subscription confirmation failed: {e}");
476        }
477
478        log::debug!("Subscribed to user order, trade, and portfolio updates");
479
480        // Spawn stream handler to dispatch WebSocket messages to the execution engine
481        let stream = self.ws_client.stream()?;
482        self.spawn_stream_handler(stream);
483
484        self.core.set_connected();
485        log::info!("Connected: client_id={}", self.core.client_id);
486        Ok(())
487    }
488
489    async fn disconnect(&mut self) -> anyhow::Result<()> {
490        if self.core.is_disconnected() {
491            return Ok(());
492        }
493
494        self.abort_pending_tasks();
495
496        // Abort stream handler
497        if let Some(handle) = self.ws_stream_handle.take() {
498            handle.abort();
499        }
500
501        // Close WebSocket client
502        if let Err(e) = self.ws_client.close().await {
503            log::warn!("Error closing WebSocket client: {e}");
504        }
505
506        self.core.set_disconnected();
507        log::info!("Disconnected: client_id={}", self.core.client_id);
508        Ok(())
509    }
510
511    async fn generate_order_status_report(
512        &self,
513        cmd: &GenerateOrderStatusReport,
514    ) -> anyhow::Result<Option<OrderStatusReport>> {
515        // If venue_order_id is provided, fetch the specific order by ID
516        if let Some(venue_order_id) = &cmd.venue_order_id {
517            let params = GetOrderStateParams {
518                order_id: venue_order_id.to_string(),
519            };
520            let ts_init = self.clock.get_time_ns();
521
522            match self.http_client.inner.get_order_state(params).await {
523                Ok(response) => {
524                    if let Some(order) = response.result {
525                        let symbol = ustr::Ustr::from(&order.instrument_name);
526                        if let Some(instrument) = self.http_client.get_instrument(&symbol) {
527                            let report = parse_user_order_msg(
528                                &order,
529                                &instrument,
530                                self.core.account_id,
531                                ts_init,
532                            )?;
533                            return Ok(Some(report));
534                        } else {
535                            log::warn!(
536                                "Instrument {} not in cache for order {}",
537                                order.instrument_name,
538                                order.order_id
539                            );
540                        }
541                    }
542                }
543                Err(e) => {
544                    log::warn!("Failed to get order state: {e}");
545                }
546            }
547            return Ok(None);
548        }
549
550        // If client_order_id is provided, search open then closed orders
551        if let Some(client_order_id) = &cmd.client_order_id {
552            let reports = self
553                .http_client
554                .request_order_status_reports(
555                    self.core.account_id,
556                    cmd.instrument_id,
557                    None,
558                    None,
559                    false, // search all orders, not just open
560                )
561                .await?;
562
563            // Filter by client_order_id
564            for report in reports {
565                if report.client_order_id == Some(*client_order_id) {
566                    return Ok(Some(report));
567                }
568            }
569        }
570
571        Ok(None)
572    }
573
574    async fn generate_order_status_reports(
575        &self,
576        cmd: &GenerateOrderStatusReports,
577    ) -> anyhow::Result<Vec<OrderStatusReport>> {
578        self.http_client
579            .request_order_status_reports(
580                self.core.account_id,
581                cmd.instrument_id,
582                cmd.start,
583                cmd.end,
584                cmd.open_only,
585            )
586            .await
587    }
588
589    async fn generate_fill_reports(
590        &self,
591        cmd: GenerateFillReports,
592    ) -> anyhow::Result<Vec<FillReport>> {
593        let mut reports = self
594            .http_client
595            .request_fill_reports(self.core.account_id, cmd.instrument_id, cmd.start, cmd.end)
596            .await?;
597
598        // Filter by venue_order_id if provided
599        if let Some(venue_order_id) = &cmd.venue_order_id {
600            reports.retain(|r| r.venue_order_id == *venue_order_id);
601        }
602
603        Ok(reports)
604    }
605
606    async fn generate_position_status_reports(
607        &self,
608        cmd: &GeneratePositionStatusReports,
609    ) -> anyhow::Result<Vec<PositionStatusReport>> {
610        self.http_client
611            .request_position_status_reports(self.core.account_id, cmd.instrument_id)
612            .await
613    }
614
615    async fn generate_mass_status(
616        &self,
617        lookback_mins: Option<u64>,
618    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
619        log::info!("Generating ExecutionMassStatus (lookback_mins={lookback_mins:?})");
620        let ts_now = self.clock.get_time_ns();
621        let start = lookback_mins.map(|mins| {
622            let lookback_ns = mins
623                .saturating_mul(60)
624                .saturating_mul(NANOSECONDS_IN_SECOND);
625            UnixNanos::from(ts_now.as_u64().saturating_sub(lookback_ns))
626        });
627
628        let order_cmd = GenerateOrderStatusReportsBuilder::default()
629            .ts_init(ts_now)
630            .open_only(false) // get all orders for mass status
631            .start(start)
632            .build()
633            .context("Failed to build GenerateOrderStatusReports")?;
634
635        let fill_cmd = GenerateFillReportsBuilder::default()
636            .ts_init(ts_now)
637            .start(start)
638            .build()
639            .context("Failed to build GenerateFillReports")?;
640
641        let position_cmd = GeneratePositionStatusReportsBuilder::default()
642            .ts_init(ts_now)
643            .start(start)
644            .build()
645            .context("Failed to build GeneratePositionStatusReports")?;
646
647        let (order_reports, fill_reports, position_reports) = tokio::try_join!(
648            self.generate_order_status_reports(&order_cmd),
649            self.generate_fill_reports(fill_cmd),
650            self.generate_position_status_reports(&position_cmd),
651        )?;
652
653        log::info!("Received {} OrderStatusReports", order_reports.len());
654        log::info!("Received {} FillReports", fill_reports.len());
655        log::info!("Received {} PositionReports", position_reports.len());
656
657        let mut mass_status = ExecutionMassStatus::new(
658            self.core.client_id,
659            self.core.account_id,
660            *DERIBIT_VENUE,
661            ts_now,
662            None,
663        );
664
665        mass_status.add_order_reports(order_reports);
666        mass_status.add_fill_reports(fill_reports);
667        mass_status.add_position_reports(position_reports);
668
669        Ok(Some(mass_status))
670    }
671
672    fn query_account(&self, _cmd: QueryAccount) -> anyhow::Result<()> {
673        let http_client = self.http_client.clone();
674        let account_id = self.core.account_id;
675        let emitter = self.emitter.clone();
676
677        self.spawn_task("query_account", async move {
678            let account_state = http_client
679                .request_account_state(account_id)
680                .await
681                .context("failed to query account state (check API credentials are valid)")?;
682
683            emitter.send_account_state(account_state);
684            Ok(())
685        });
686
687        Ok(())
688    }
689
690    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
691        let ws_client = self.ws_client.clone();
692
693        // Extract venue order ID (Deribit's order_id)
694        let order_id = cmd
695            .venue_order_id
696            .as_ref()
697            .ok_or_else(|| anyhow::anyhow!("venue_order_id required for query_order"))?
698            .to_string();
699
700        let client_order_id = cmd.client_order_id;
701        let trader_id = cmd.trader_id;
702        let strategy_id = cmd.strategy_id;
703        let instrument_id = cmd.instrument_id;
704
705        log::debug!("Querying order state: order_id={order_id}, client_order_id={client_order_id}");
706
707        // Spawn async task to query order state via WebSocket
708        // Response will be dispatched through the WebSocket stream handler as OrderStatusReport
709        self.spawn_task("query_order", async move {
710            ws_client
711                .query_order(
712                    &order_id,
713                    client_order_id,
714                    trader_id,
715                    strategy_id,
716                    instrument_id,
717                )
718                .await
719                .map_err(|e| anyhow::anyhow!("Query order state failed: {e}"))?;
720            Ok(())
721        });
722
723        Ok(())
724    }
725
726    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
727        let order = self.core.cache().try_order_owned(&cmd.client_order_id)?;
728        self.submit_single_order(&order, "submit_order");
729        Ok(())
730    }
731
732    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
733        if cmd.order_list.client_order_ids.is_empty() {
734            log::debug!("submit_order_list called with empty order list");
735            return Ok(());
736        }
737
738        let orders = self.core.get_orders_for_list(&cmd.order_list)?;
739
740        log::debug!(
741            "Submitting order list {} with {} orders for instrument={}",
742            cmd.order_list.id,
743            orders.len(),
744            cmd.instrument_id
745        );
746
747        // Deribit doesn't have native batch order submission
748        // Loop through and submit each order individually using shared logic
749        for order in &orders {
750            self.submit_single_order(order, "submit_order_list_item");
751        }
752
753        Ok(())
754    }
755
756    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
757        let ws_client = self.ws_client.clone();
758
759        // Extract venue order ID (Deribit's order_id)
760        let order_id = match cmd.venue_order_id.as_ref() {
761            Some(venue_order_id) => venue_order_id.to_string(),
762            None => {
763                return reject_modify_command(
764                    &self.emitter,
765                    self.clock,
766                    &cmd,
767                    "venue_order_id required for modify_order",
768                );
769            }
770        };
771
772        // Extract quantity - if not provided, get from order in cache
773        let quantity = if let Some(qty) = cmd.quantity {
774            qty
775        } else {
776            // Get order from cache to use its current quantity
777            let cache = self.core.cache();
778            match cache.order(&cmd.client_order_id) {
779                Some(order) => order.quantity(),
780                None => {
781                    return reject_modify_command(
782                        &self.emitter,
783                        self.clock,
784                        &cmd,
785                        &format!("Order not found: {}", cmd.client_order_id),
786                    );
787                }
788            }
789        };
790
791        let price = match cmd.price {
792            Some(price) => price,
793            None => {
794                return reject_modify_command(
795                    &self.emitter,
796                    self.clock,
797                    &cmd,
798                    "price required for modify_order",
799                );
800            }
801        };
802
803        let client_order_id = cmd.client_order_id;
804        let trader_id = cmd.trader_id;
805        let strategy_id = cmd.strategy_id;
806        let instrument_id = cmd.instrument_id;
807
808        log::debug!(
809            "Modifying order: order_id={order_id}, quantity={quantity}, price={price}, client_order_id={client_order_id}"
810        );
811
812        // Spawn async task to send modify via WebSocket
813        self.spawn_task("modify_order", async move {
814            if let Err(e) = ws_client
815                .modify_order(
816                    &order_id,
817                    quantity,
818                    price,
819                    client_order_id,
820                    trader_id,
821                    strategy_id,
822                    instrument_id,
823                )
824                .await
825            {
826                log::error!(
827                    "Modify order failed: order_id={order_id}, client_order_id={client_order_id}, error={e}"
828                );
829                anyhow::bail!("Modify order failed: {e}");
830            }
831            Ok(())
832        });
833
834        Ok(())
835    }
836
837    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
838        let ws_client = self.ws_client.clone();
839
840        // Extract venue order ID (Deribit's order_id)
841        let order_id = match cmd.venue_order_id.as_ref() {
842            Some(venue_order_id) => venue_order_id.to_string(),
843            None => {
844                log::warn!(
845                    "Cannot cancel order {} - no venue_order_id",
846                    cmd.client_order_id
847                );
848                return Ok(());
849            }
850        };
851
852        let client_order_id = cmd.client_order_id;
853        let trader_id = cmd.trader_id;
854        let strategy_id = cmd.strategy_id;
855        let instrument_id = cmd.instrument_id;
856
857        log::debug!("Canceling order: order_id={order_id}, client_order_id={client_order_id}");
858
859        // Spawn async task to send cancel via WebSocket
860        self.spawn_task("cancel_order", async move {
861            if let Err(e) = ws_client
862                .cancel_order(
863                    &order_id,
864                    client_order_id,
865                    trader_id,
866                    strategy_id,
867                    instrument_id,
868                )
869                .await
870            {
871                log::error!(
872                    "Cancel order failed: order_id={order_id}, client_order_id={client_order_id}, error={e}"
873                );
874                anyhow::bail!("Cancel order failed: {e}");
875            }
876            Ok(())
877        });
878
879        Ok(())
880    }
881
882    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
883        let instrument_id = cmd.instrument_id;
884
885        // If NoOrderSide, use efficient bulk cancel via Deribit API
886        if cmd.order_side == OrderSide::NoOrderSide {
887            log::debug!(
888                "Cancelling all orders: instrument={instrument_id}, order_side=NoOrderSide (bulk)"
889            );
890
891            let ws_client = self.ws_client.clone();
892            self.spawn_task("cancel_all_orders", async move {
893                if let Err(e) = ws_client.cancel_all_orders(instrument_id, None).await {
894                    log::error!("Cancel all orders failed for instrument {instrument_id}: {e}");
895                    anyhow::bail!("Cancel all orders failed: {e}");
896                }
897                Ok(())
898            });
899
900            return Ok(());
901        }
902
903        // For specific side (Buy/Sell), filter from cache and cancel individually
904        // Deribit API doesn't support side filtering, so we implement it locally
905        log::debug!(
906            "Cancelling orders by side: instrument={}, order_side={}",
907            instrument_id,
908            cmd.order_side
909        );
910
911        let orders_to_cancel: Vec<_> = {
912            let cache = self.core.cache();
913            let open_orders = cache.orders_open(None, Some(&instrument_id), None, None, None);
914
915            open_orders
916                .into_iter()
917                .filter(|order| order.order_side() == cmd.order_side)
918                .filter_map(|order| {
919                    let venue_order_id = order.venue_order_id()?;
920                    Some((
921                        venue_order_id.to_string(),
922                        order.client_order_id(),
923                        order.instrument_id(),
924                        Some(venue_order_id),
925                    ))
926                })
927                .collect()
928        };
929
930        if orders_to_cancel.is_empty() {
931            log::debug!(
932                "No open {} orders to cancel for {}",
933                cmd.order_side,
934                instrument_id
935            );
936            return Ok(());
937        }
938
939        log::debug!(
940            "Cancelling {} {} orders for {}",
941            orders_to_cancel.len(),
942            cmd.order_side,
943            instrument_id
944        );
945
946        // Cancel each matching order individually
947        for (venue_order_id_str, client_order_id, order_instrument_id, _venue_order_id) in
948            orders_to_cancel
949        {
950            let ws_client = self.ws_client.clone();
951            let trader_id = cmd.trader_id;
952            let strategy_id = cmd.strategy_id;
953
954            self.spawn_task("cancel_order_by_side", async move {
955                if let Err(e) = ws_client
956                    .cancel_order(
957                        &venue_order_id_str,
958                        client_order_id,
959                        trader_id,
960                        strategy_id,
961                        order_instrument_id,
962                    )
963                    .await
964                {
965                    log::error!(
966                        "Cancel order failed: order_id={venue_order_id_str}, client_order_id={client_order_id}, error={e}"
967                    );
968                }
969                Ok(())
970            });
971        }
972
973        Ok(())
974    }
975
976    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
977        if cmd.cancels.is_empty() {
978            log::debug!("batch_cancel_orders called with empty cancels list");
979            return Ok(());
980        }
981
982        log::debug!(
983            "Batch cancelling {} orders for instrument={}",
984            cmd.cancels.len(),
985            cmd.instrument_id
986        );
987
988        // Deribit doesn't have native batch cancel by order ID
989        // Loop through and cancel each order individually
990        for cancel in &cmd.cancels {
991            let order_id = match &cancel.venue_order_id {
992                Some(id) => id.to_string(),
993                None => {
994                    log::warn!(
995                        "Cannot cancel order {} - no venue_order_id",
996                        cancel.client_order_id
997                    );
998                    continue;
999                }
1000            };
1001
1002            let ws_client = self.ws_client.clone();
1003            let client_order_id = cancel.client_order_id;
1004            let trader_id = cancel.trader_id;
1005            let strategy_id = cancel.strategy_id;
1006            let instrument_id = cancel.instrument_id;
1007
1008            self.spawn_task("batch_cancel_order", async move {
1009                if let Err(e) = ws_client
1010                    .cancel_order(
1011                        &order_id,
1012                        client_order_id,
1013                        trader_id,
1014                        strategy_id,
1015                        instrument_id,
1016                    )
1017                    .await
1018                {
1019                    log::error!(
1020                        "Batch cancel order failed: order_id={order_id}, client_order_id={client_order_id}, error={e}"
1021                    );
1022                    anyhow::bail!("Batch cancel order failed: {e}");
1023                }
1024                Ok(())
1025            });
1026        }
1027
1028        Ok(())
1029    }
1030}
1031
1032/// Dispatches a WebSocket message using the event emitter.
1033fn dispatch_ws_message(message: NautilusWsMessage, emitter: &ExecutionEventEmitter) {
1034    match message {
1035        NautilusWsMessage::AccountState(state) => {
1036            emitter.send_account_state(state);
1037        }
1038        NautilusWsMessage::OrderStatusReports(reports) => {
1039            log::debug!("Processing {} order status report(s)", reports.len());
1040            for report in reports {
1041                emitter.send_order_status_report(report);
1042            }
1043        }
1044        NautilusWsMessage::FillReports(reports) => {
1045            log::debug!("Processing {} fill report(s)", reports.len());
1046            for report in reports {
1047                emitter.send_fill_report(report);
1048            }
1049        }
1050        NautilusWsMessage::OrderRejected(event) => {
1051            emitter.send_order_event(OrderEventAny::Rejected(event));
1052        }
1053        NautilusWsMessage::OrderAccepted(event) => {
1054            emitter.send_order_event(OrderEventAny::Accepted(event));
1055        }
1056        NautilusWsMessage::OrderCanceled(event) => {
1057            emitter.send_order_event(OrderEventAny::Canceled(event));
1058        }
1059        NautilusWsMessage::OrderExpired(event) => {
1060            emitter.send_order_event(OrderEventAny::Expired(event));
1061        }
1062        NautilusWsMessage::OrderUpdated(event) => {
1063            emitter.send_order_event(OrderEventAny::Updated(event));
1064        }
1065        NautilusWsMessage::OrderCancelRejected(event) => {
1066            emitter.send_order_event(OrderEventAny::CancelRejected(event));
1067        }
1068        NautilusWsMessage::OrderModifyRejected(event) => {
1069            emitter.send_order_event(OrderEventAny::ModifyRejected(event));
1070        }
1071        NautilusWsMessage::Error(e) => {
1072            log::warn!("WebSocket error: {e}");
1073        }
1074        NautilusWsMessage::Reconnected => {
1075            log::info!("WebSocket reconnected");
1076        }
1077        NautilusWsMessage::Authenticated(auth) => {
1078            log::debug!("WebSocket authenticated: scope={}", auth.scope);
1079        }
1080        NautilusWsMessage::AuthenticationFailed(reason) => {
1081            log::error!("Authentication failed in execution client: {reason}");
1082        }
1083        NautilusWsMessage::Data(_)
1084        | NautilusWsMessage::Deltas(_)
1085        | NautilusWsMessage::Instrument(_)
1086        | NautilusWsMessage::InstrumentStatus(_)
1087        | NautilusWsMessage::FundingRates(_)
1088        | NautilusWsMessage::OptionGreeks(_)
1089        | NautilusWsMessage::Raw(_) => {
1090            // Data messages are handled by the data client, not execution
1091            log::trace!("Ignoring data message in execution client");
1092        }
1093    }
1094}
1095
1096fn reject_modify_command(
1097    emitter: &ExecutionEventEmitter,
1098    clock: &AtomicTime,
1099    cmd: &ModifyOrder,
1100    reason: &str,
1101) -> anyhow::Result<()> {
1102    let ts_event = clock.get_time_ns();
1103    emitter.emit_order_modify_rejected_event(
1104        cmd.strategy_id,
1105        cmd.instrument_id,
1106        cmd.client_order_id,
1107        cmd.venue_order_id,
1108        reason,
1109        ts_event,
1110    );
1111    anyhow::bail!("{reason}");
1112}
nautilus_deribit/execution.rs

nautilus_deribit/
execution.rs
