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

nautilus_deribit/
execution.rs
