nautilus_dydx/execution/
submitter.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//! Order submission utilities for dYdX v4.
17//!
18//! This module provides functions for building and submitting orders to the dYdX protocol,
19//! including conditional orders (stop-loss, take-profit) and market/limit orders.
20
21use chrono::{DateTime, Duration, Utc};
22use nautilus_model::{
23    enums::{OrderSide, TimeInForce},
24    identifiers::InstrumentId,
25    types::{Price, Quantity},
26};
27
28use crate::{
29    common::parse::{order_side_to_proto, time_in_force_to_proto_with_post_only},
30    error::DydxError,
31    grpc::{
32        DydxGrpcClient, OrderBuilder, OrderGoodUntil, OrderMarketParams,
33        SHORT_TERM_ORDER_MAXIMUM_LIFETIME, TxBuilder, Wallet, types::ChainId,
34    },
35    http::client::DydxHttpClient,
36    proto::{
37        ToAny,
38        dydxprotocol::clob::{MsgCancelOrder, MsgPlaceOrder},
39    },
40};
41
42/// Default expiration for GTC conditional orders (90 days).
43const GTC_CONDITIONAL_ORDER_EXPIRATION_DAYS: i64 = 90;
44
45/// Conditional order types supported by dYdX.
46#[derive(Debug, Clone, Copy)]
47pub enum ConditionalOrderType {
48    /// Triggers at trigger price, executes as market order.
49    StopMarket,
50    /// Triggers at trigger price, places limit order at limit price.
51    StopLimit,
52    /// Triggers at trigger price for profit taking, executes as market order.
53    TakeProfitMarket,
54    /// Triggers at trigger price for profit taking, places limit order at limit price.
55    TakeProfitLimit,
56}
57
58/// Calculates the expiration time for conditional orders based on TimeInForce.
59///
60/// - `GTD` with explicit `expire_time`: uses the provided timestamp.
61/// - `GTC` or no `expire_time`: defaults to 90 days from now.
62/// - `IOC`/`FOK`: uses 1 hour (these are unusual for conditional orders).
63///
64/// # Errors
65///
66/// Returns `DydxError::Parse` if the provided `expire_time` timestamp is invalid.
67fn calculate_conditional_order_expiration(
68    time_in_force: TimeInForce,
69    expire_time: Option<i64>,
70) -> Result<DateTime<Utc>, DydxError> {
71    if let Some(expire_ts) = expire_time {
72        DateTime::from_timestamp(expire_ts, 0)
73            .ok_or_else(|| DydxError::Parse(format!("Invalid expire timestamp: {expire_ts}")))
74    } else {
75        let expiration = match time_in_force {
76            TimeInForce::Gtc => Utc::now() + Duration::days(GTC_CONDITIONAL_ORDER_EXPIRATION_DAYS),
77            TimeInForce::Ioc | TimeInForce::Fok => {
78                // IOC/FOK don't typically apply to conditional orders, use short expiration
79                Utc::now() + Duration::hours(1)
80            }
81            // GTD without expire_time, or any other TIF - use long default
82            _ => Utc::now() + Duration::days(GTC_CONDITIONAL_ORDER_EXPIRATION_DAYS),
83        };
84        Ok(expiration)
85    }
86}
87
88#[derive(Debug)]
89pub struct OrderSubmitter {
90    grpc_client: DydxGrpcClient,
91    http_client: DydxHttpClient,
92    wallet_address: String,
93    subaccount_number: u32,
94    chain_id: ChainId,
95    authenticator_ids: Vec<u64>,
96}
97
98impl OrderSubmitter {
99    pub fn new(
100        grpc_client: DydxGrpcClient,
101        http_client: DydxHttpClient,
102        wallet_address: String,
103        subaccount_number: u32,
104        chain_id: ChainId,
105        authenticator_ids: Vec<u64>,
106    ) -> Self {
107        Self {
108            grpc_client,
109            http_client,
110            wallet_address,
111            subaccount_number,
112            chain_id,
113            authenticator_ids,
114        }
115    }
116
117    /// Submits a market order to dYdX via gRPC.
118    ///
119    /// Market orders execute immediately at the best available price.
120    ///
121    /// # Errors
122    ///
123    /// Returns `DydxError` if gRPC submission fails.
124    pub async fn submit_market_order(
125        &self,
126        wallet: &Wallet,
127        instrument_id: InstrumentId,
128        client_order_id: u32,
129        side: OrderSide,
130        quantity: Quantity,
131        block_height: u32,
132    ) -> Result<(), DydxError> {
133        log::info!(
134            "Submitting market order: client_id={client_order_id}, side={side:?}, quantity={quantity}"
135        );
136
137        // Get market params from instrument cache
138        let market_params = self.get_market_params(instrument_id)?;
139
140        // Build order using OrderBuilder
141        let mut builder = OrderBuilder::new(
142            market_params,
143            self.wallet_address.clone(),
144            self.subaccount_number,
145            client_order_id,
146        );
147
148        let proto_side = order_side_to_proto(side);
149        let size_decimal = quantity.as_decimal();
150
151        builder = builder.market(proto_side, size_decimal);
152        builder = builder.short_term(); // Market orders are short-term
153        builder = builder.until(OrderGoodUntil::Block(
154            block_height + SHORT_TERM_ORDER_MAXIMUM_LIFETIME,
155        ));
156
157        let order = builder
158            .build()
159            .map_err(|e| DydxError::Order(format!("Failed to build market order: {e}")))?;
160
161        // Create MsgPlaceOrder
162        let msg_place_order = MsgPlaceOrder { order: Some(order) };
163
164        // Broadcast transaction
165        self.broadcast_order_message(wallet, msg_place_order).await
166    }
167
168    /// Submits a limit order to dYdX via gRPC.
169    ///
170    /// Limit orders execute only at the specified price or better.
171    ///
172    /// # Errors
173    ///
174    /// Returns `DydxError` if gRPC submission fails.
175    #[allow(clippy::too_many_arguments)]
176    pub async fn submit_limit_order(
177        &self,
178        wallet: &Wallet,
179        instrument_id: InstrumentId,
180        client_order_id: u32,
181        side: OrderSide,
182        price: Price,
183        quantity: Quantity,
184        time_in_force: TimeInForce,
185        post_only: bool,
186        reduce_only: bool,
187        block_height: u32,
188        expire_time: Option<i64>,
189    ) -> Result<(), DydxError> {
190        log::info!(
191            "Submitting limit order: client_id={client_order_id}, side={side:?}, price={price}, quantity={quantity}, tif={time_in_force:?}, post_only={post_only}, reduce_only={reduce_only}"
192        );
193
194        // Get market params from instrument cache
195        let market_params = self.get_market_params(instrument_id)?;
196
197        // Build order using OrderBuilder
198        let mut builder = OrderBuilder::new(
199            market_params,
200            self.wallet_address.clone(),
201            self.subaccount_number,
202            client_order_id,
203        );
204
205        let proto_side = order_side_to_proto(side);
206        let price_decimal = price.as_decimal();
207        let size_decimal = quantity.as_decimal();
208
209        builder = builder.limit(proto_side, price_decimal, size_decimal);
210
211        // Set time in force (post_only orders use TimeInForce::PostOnly in dYdX)
212        let proto_tif = time_in_force_to_proto_with_post_only(time_in_force, post_only);
213        builder = builder.time_in_force(proto_tif);
214
215        // Set reduce_only flag
216        if reduce_only {
217            builder = builder.reduce_only(true);
218        }
219
220        // Determine if short-term or long-term based on TIF and expire_time
221        if let Some(expire_ts) = expire_time {
222            builder = builder.long_term();
223            builder = builder.until(OrderGoodUntil::Time(
224                DateTime::from_timestamp(expire_ts, 0)
225                    .ok_or_else(|| DydxError::Parse("Invalid expire timestamp".to_string()))?,
226            ));
227        } else {
228            builder = builder.short_term();
229            builder = builder.until(OrderGoodUntil::Block(
230                block_height + SHORT_TERM_ORDER_MAXIMUM_LIFETIME,
231            ));
232        }
233
234        let order = builder
235            .build()
236            .map_err(|e| DydxError::Order(format!("Failed to build limit order: {e}")))?;
237
238        // Create MsgPlaceOrder
239        let msg_place_order = MsgPlaceOrder { order: Some(order) };
240
241        // Broadcast transaction
242        self.broadcast_order_message(wallet, msg_place_order).await
243    }
244
245    /// Cancels an order on dYdX via gRPC.
246    ///
247    /// Requires instrument_id to retrieve correct clob_pair_id from market params.
248    /// For now, assumes short-term orders (order_flags=0). Future enhancement:
249    /// track order_flags when placing orders to handle long-term cancellations.
250    ///
251    /// # Errors
252    ///
253    /// Returns `DydxError` if gRPC cancellation fails or market params not found.
254    pub async fn cancel_order(
255        &self,
256        wallet: &Wallet,
257        instrument_id: InstrumentId,
258        client_order_id: u32,
259        block_height: u32,
260    ) -> Result<(), DydxError> {
261        log::info!("Cancelling order: client_id={client_order_id}, instrument={instrument_id}");
262
263        // Get market params to retrieve clob_pair_id
264        let market_params = self.get_market_params(instrument_id)?;
265
266        // Create MsgCancelOrder
267        let msg_cancel = MsgCancelOrder {
268            order_id: Some(crate::proto::dydxprotocol::clob::OrderId {
269                subaccount_id: Some(crate::proto::dydxprotocol::subaccounts::SubaccountId {
270                    owner: self.wallet_address.clone(),
271                    number: self.subaccount_number,
272                }),
273                client_id: client_order_id,
274                order_flags: 0, // Short-term orders (0), long-term (64), conditional (32)
275                clob_pair_id: market_params.clob_pair_id,
276            }),
277            good_til_oneof: Some(
278                crate::proto::dydxprotocol::clob::msg_cancel_order::GoodTilOneof::GoodTilBlock(
279                    block_height + SHORT_TERM_ORDER_MAXIMUM_LIFETIME,
280                ),
281            ),
282        };
283
284        // Broadcast transaction
285        self.broadcast_cancel_message(wallet, msg_cancel).await
286    }
287
288    /// Cancels multiple orders in a single blockchain transaction.
289    ///
290    /// Batches all cancellation messages into one transaction for efficiency.
291    /// This is more efficient than sequential cancellation as it requires only
292    /// one account lookup and one transaction broadcast.
293    ///
294    /// # Arguments
295    ///
296    /// * `wallet` - The wallet for signing transactions
297    /// * `orders` - Slice of (InstrumentId, client_order_id) tuples to cancel
298    /// * `block_height` - Current block height for order expiration
299    ///
300    /// # Errors
301    ///
302    /// Returns `DydxError` if transaction broadcast fails or market params not found.
303    pub async fn cancel_orders_batch(
304        &self,
305        wallet: &Wallet,
306        orders: &[(InstrumentId, u32)],
307        block_height: u32,
308    ) -> Result<(), DydxError> {
309        if orders.is_empty() {
310            return Ok(());
311        }
312
313        log::info!(
314            "Batch cancelling {} orders in single transaction",
315            orders.len()
316        );
317
318        // Build all cancel messages
319        let mut cancel_msgs = Vec::with_capacity(orders.len());
320        for (instrument_id, client_order_id) in orders {
321            let market_params = self.get_market_params(*instrument_id)?;
322
323            let msg_cancel = MsgCancelOrder {
324                order_id: Some(crate::proto::dydxprotocol::clob::OrderId {
325                    subaccount_id: Some(crate::proto::dydxprotocol::subaccounts::SubaccountId {
326                        owner: self.wallet_address.clone(),
327                        number: self.subaccount_number,
328                    }),
329                    client_id: *client_order_id,
330                    order_flags: 0,
331                    clob_pair_id: market_params.clob_pair_id,
332                }),
333                good_til_oneof: Some(
334                    crate::proto::dydxprotocol::clob::msg_cancel_order::GoodTilOneof::GoodTilBlock(
335                        block_height + SHORT_TERM_ORDER_MAXIMUM_LIFETIME,
336                    ),
337                ),
338            };
339            cancel_msgs.push(msg_cancel);
340        }
341
342        // Broadcast all cancellations in a single transaction
343        self.broadcast_cancel_messages_batch(wallet, cancel_msgs)
344            .await
345    }
346
347    /// Submits a conditional order (stop or take-profit) to dYdX via gRPC.
348    ///
349    /// This is the unified implementation for all conditional order types.
350    /// Market variants execute immediately when triggered; limit variants place
351    /// a limit order at the specified price.
352    ///
353    /// # Errors
354    ///
355    /// Returns `DydxError` if gRPC submission fails or limit_price missing for limit orders.
356    #[allow(clippy::too_many_arguments)]
357    pub async fn submit_conditional_order(
358        &self,
359        wallet: &Wallet,
360        instrument_id: InstrumentId,
361        client_order_id: u32,
362        order_type: ConditionalOrderType,
363        side: OrderSide,
364        trigger_price: Price,
365        limit_price: Option<Price>,
366        quantity: Quantity,
367        time_in_force: Option<TimeInForce>,
368        post_only: bool,
369        reduce_only: bool,
370        expire_time: Option<i64>,
371    ) -> Result<(), DydxError> {
372        let market_params = self.get_market_params(instrument_id)?;
373
374        let mut builder = OrderBuilder::new(
375            market_params,
376            self.wallet_address.clone(),
377            self.subaccount_number,
378            client_order_id,
379        );
380
381        let proto_side = order_side_to_proto(side);
382        let trigger_decimal = trigger_price.as_decimal();
383        let size_decimal = quantity.as_decimal();
384
385        // Apply order-type-specific builder method
386        builder = match order_type {
387            ConditionalOrderType::StopMarket => {
388                log::info!(
389                    "Submitting stop market order: client_id={client_order_id}, side={side:?}, trigger={trigger_price}, qty={quantity}"
390                );
391                builder.stop_market(proto_side, trigger_decimal, size_decimal)
392            }
393            ConditionalOrderType::StopLimit => {
394                let limit = limit_price.ok_or_else(|| {
395                    DydxError::Order("StopLimit requires limit_price".to_string())
396                })?;
397                log::info!(
398                    "Submitting stop limit order: client_id={client_order_id}, side={side:?}, trigger={trigger_price}, limit={limit}, qty={quantity}"
399                );
400                builder.stop_limit(
401                    proto_side,
402                    limit.as_decimal(),
403                    trigger_decimal,
404                    size_decimal,
405                )
406            }
407            ConditionalOrderType::TakeProfitMarket => {
408                log::info!(
409                    "Submitting take profit market order: client_id={client_order_id}, side={side:?}, trigger={trigger_price}, qty={quantity}"
410                );
411                builder.take_profit_market(proto_side, trigger_decimal, size_decimal)
412            }
413            ConditionalOrderType::TakeProfitLimit => {
414                let limit = limit_price.ok_or_else(|| {
415                    DydxError::Order("TakeProfitLimit requires limit_price".to_string())
416                })?;
417                log::info!(
418                    "Submitting take profit limit order: client_id={client_order_id}, side={side:?}, trigger={trigger_price}, limit={limit}, qty={quantity}"
419                );
420                builder.take_profit_limit(
421                    proto_side,
422                    limit.as_decimal(),
423                    trigger_decimal,
424                    size_decimal,
425                )
426            }
427        };
428
429        // Apply time-in-force for limit orders
430        let effective_tif = time_in_force.unwrap_or(TimeInForce::Gtc);
431        if matches!(
432            order_type,
433            ConditionalOrderType::StopLimit | ConditionalOrderType::TakeProfitLimit
434        ) {
435            let proto_tif = time_in_force_to_proto_with_post_only(effective_tif, post_only);
436            builder = builder.time_in_force(proto_tif);
437        }
438
439        if reduce_only {
440            builder = builder.reduce_only(true);
441        }
442
443        let expire = calculate_conditional_order_expiration(effective_tif, expire_time)?;
444        builder = builder.until(OrderGoodUntil::Time(expire));
445
446        let order = builder
447            .build()
448            .map_err(|e| DydxError::Order(format!("Failed to build {order_type:?} order: {e}")))?;
449
450        let msg_place_order = MsgPlaceOrder { order: Some(order) };
451        self.broadcast_order_message(wallet, msg_place_order).await
452    }
453
454    /// Submits a stop market order to dYdX via gRPC.
455    ///
456    /// Stop market orders are triggered when the price reaches `trigger_price`.
457    ///
458    /// # Errors
459    ///
460    /// Returns `DydxError` if gRPC submission fails.
461    #[allow(clippy::too_many_arguments)]
462    pub async fn submit_stop_market_order(
463        &self,
464        wallet: &Wallet,
465        instrument_id: InstrumentId,
466        client_order_id: u32,
467        side: OrderSide,
468        trigger_price: Price,
469        quantity: Quantity,
470        reduce_only: bool,
471        expire_time: Option<i64>,
472    ) -> Result<(), DydxError> {
473        self.submit_conditional_order(
474            wallet,
475            instrument_id,
476            client_order_id,
477            ConditionalOrderType::StopMarket,
478            side,
479            trigger_price,
480            None,
481            quantity,
482            None,
483            false,
484            reduce_only,
485            expire_time,
486        )
487        .await
488    }
489
490    /// Submits a stop limit order to dYdX via gRPC.
491    ///
492    /// Stop limit orders are triggered when the price reaches `trigger_price`,
493    /// then placed as a limit order at `limit_price`.
494    ///
495    /// # Errors
496    ///
497    /// Returns `DydxError` if gRPC submission fails.
498    #[allow(clippy::too_many_arguments)]
499    pub async fn submit_stop_limit_order(
500        &self,
501        wallet: &Wallet,
502        instrument_id: InstrumentId,
503        client_order_id: u32,
504        side: OrderSide,
505        trigger_price: Price,
506        limit_price: Price,
507        quantity: Quantity,
508        time_in_force: TimeInForce,
509        post_only: bool,
510        reduce_only: bool,
511        expire_time: Option<i64>,
512    ) -> Result<(), DydxError> {
513        self.submit_conditional_order(
514            wallet,
515            instrument_id,
516            client_order_id,
517            ConditionalOrderType::StopLimit,
518            side,
519            trigger_price,
520            Some(limit_price),
521            quantity,
522            Some(time_in_force),
523            post_only,
524            reduce_only,
525            expire_time,
526        )
527        .await
528    }
529
530    /// Submits a take profit market order to dYdX via gRPC.
531    ///
532    /// Take profit market orders are triggered when the price reaches `trigger_price`,
533    /// then executed as a market order.
534    ///
535    /// # Errors
536    ///
537    /// Returns `DydxError` if gRPC submission fails.
538    #[allow(clippy::too_many_arguments)]
539    pub async fn submit_take_profit_market_order(
540        &self,
541        wallet: &Wallet,
542        instrument_id: InstrumentId,
543        client_order_id: u32,
544        side: OrderSide,
545        trigger_price: Price,
546        quantity: Quantity,
547        reduce_only: bool,
548        expire_time: Option<i64>,
549    ) -> Result<(), DydxError> {
550        self.submit_conditional_order(
551            wallet,
552            instrument_id,
553            client_order_id,
554            ConditionalOrderType::TakeProfitMarket,
555            side,
556            trigger_price,
557            None,
558            quantity,
559            None,
560            false,
561            reduce_only,
562            expire_time,
563        )
564        .await
565    }
566
567    /// Submits a take profit limit order to dYdX via gRPC.
568    ///
569    /// Take profit limit orders are triggered when the price reaches `trigger_price`,
570    /// then placed as a limit order at `limit_price`.
571    ///
572    /// # Errors
573    ///
574    /// Returns `DydxError` if gRPC submission fails.
575    #[allow(clippy::too_many_arguments)]
576    pub async fn submit_take_profit_limit_order(
577        &self,
578        wallet: &Wallet,
579        instrument_id: InstrumentId,
580        client_order_id: u32,
581        side: OrderSide,
582        trigger_price: Price,
583        limit_price: Price,
584        quantity: Quantity,
585        time_in_force: TimeInForce,
586        post_only: bool,
587        reduce_only: bool,
588        expire_time: Option<i64>,
589    ) -> Result<(), DydxError> {
590        self.submit_conditional_order(
591            wallet,
592            instrument_id,
593            client_order_id,
594            ConditionalOrderType::TakeProfitLimit,
595            side,
596            trigger_price,
597            Some(limit_price),
598            quantity,
599            Some(time_in_force),
600            post_only,
601            reduce_only,
602            expire_time,
603        )
604        .await
605    }
606
607    /// Get market params from instrument cache.
608    ///
609    /// # Errors
610    ///
611    /// Returns an error if instrument is not found in cache or market params cannot be extracted.
612    fn get_market_params(
613        &self,
614        instrument_id: InstrumentId,
615    ) -> Result<OrderMarketParams, DydxError> {
616        // Look up market data from HTTP client cache
617        let market = self
618            .http_client
619            .get_market_params(&instrument_id)
620            .ok_or_else(|| {
621                DydxError::Order(format!(
622                    "Market params for instrument '{instrument_id}' not found in cache"
623                ))
624            })?;
625
626        Ok(OrderMarketParams {
627            atomic_resolution: market.atomic_resolution,
628            clob_pair_id: market.clob_pair_id,
629            oracle_price: None, // Oracle price is dynamic, updated separately
630            quantum_conversion_exponent: market.quantum_conversion_exponent,
631            step_base_quantums: market.step_base_quantums,
632            subticks_per_tick: market.subticks_per_tick,
633        })
634    }
635
636    /// Broadcasts a transaction message to dYdX via gRPC.
637    ///
638    /// Generic method for broadcasting any transaction type that implements `ToAny`.
639    /// Handles signing, serialization, and gRPC transmission.
640    async fn broadcast_tx_message<T: ToAny>(
641        &self,
642        wallet: &Wallet,
643        msg: T,
644        operation: &str,
645    ) -> Result<(), DydxError> {
646        // Derive account for signing (uses derivation index 0 for main account)
647        let mut account = wallet
648            .account_offline(0)
649            .map_err(|e| DydxError::Wallet(format!("Failed to derive account: {e}")))?;
650
651        // Fetch current account info from chain to get proper account_number and sequence
652        let mut grpc_client = self.grpc_client.clone();
653        let base_account = grpc_client
654            .get_account(&self.wallet_address)
655            .await
656            .map_err(|e| {
657                DydxError::Grpc(Box::new(tonic::Status::internal(format!(
658                    "Failed to fetch account info: {e}"
659                ))))
660            })?;
661
662        // Update account with on-chain values
663        account.set_account_info(base_account.account_number, base_account.sequence);
664
665        // Build transaction
666        let tx_builder =
667            TxBuilder::new(self.chain_id.clone(), "adydx".to_string()).map_err(|e| {
668                DydxError::Grpc(Box::new(tonic::Status::internal(format!(
669                    "TxBuilder init failed: {e}"
670                ))))
671            })?;
672
673        // Convert message to Any
674        let any_msg = msg.to_any();
675
676        // Build and sign transaction
677        let auth_ids = if self.authenticator_ids.is_empty() {
678            None
679        } else {
680            Some(self.authenticator_ids.as_slice())
681        };
682        let tx_raw = tx_builder
683            .build_transaction(&account, vec![any_msg], None, auth_ids)
684            .map_err(|e| {
685                DydxError::Grpc(Box::new(tonic::Status::internal(format!(
686                    "Failed to build tx: {e}"
687                ))))
688            })?;
689
690        // Broadcast transaction
691        let tx_bytes = tx_raw.to_bytes().map_err(|e| {
692            DydxError::Grpc(Box::new(tonic::Status::internal(format!(
693                "Failed to serialize tx: {e}"
694            ))))
695        })?;
696
697        log::debug!(
698            "Broadcasting {} with {} bytes, account_seq={}",
699            operation,
700            tx_bytes.len(),
701            account.sequence_number
702        );
703
704        let mut grpc_client = self.grpc_client.clone();
705        let tx_hash = grpc_client.broadcast_tx(tx_bytes).await.map_err(|e| {
706            log::error!("gRPC broadcast failed for {operation}: {e}");
707            DydxError::Grpc(Box::new(tonic::Status::internal(format!(
708                "Broadcast failed: {e}"
709            ))))
710        })?;
711
712        log::info!("{operation} successfully: tx_hash={tx_hash}");
713        Ok(())
714    }
715
716    /// Broadcast order placement message via gRPC.
717    async fn broadcast_order_message(
718        &self,
719        wallet: &Wallet,
720        msg: MsgPlaceOrder,
721    ) -> Result<(), DydxError> {
722        self.broadcast_tx_message(wallet, msg, "Order placed").await
723    }
724
725    /// Broadcast order cancellation message via gRPC.
726    async fn broadcast_cancel_message(
727        &self,
728        wallet: &Wallet,
729        msg: MsgCancelOrder,
730    ) -> Result<(), DydxError> {
731        self.broadcast_tx_message(wallet, msg, "Order cancelled")
732            .await
733    }
734
735    /// Broadcast multiple order cancellation messages in a single transaction.
736    async fn broadcast_cancel_messages_batch(
737        &self,
738        wallet: &Wallet,
739        msgs: Vec<MsgCancelOrder>,
740    ) -> Result<(), DydxError> {
741        let count = msgs.len();
742
743        // Derive account for signing
744        let mut account = wallet
745            .account_offline(0)
746            .map_err(|e| DydxError::Wallet(format!("Failed to derive account: {e}")))?;
747
748        // Fetch current account info
749        let mut grpc_client = self.grpc_client.clone();
750        let base_account = grpc_client
751            .get_account(&self.wallet_address)
752            .await
753            .map_err(|e| {
754                DydxError::Grpc(Box::new(tonic::Status::internal(format!(
755                    "Failed to fetch account info: {e}"
756                ))))
757            })?;
758
759        account.set_account_info(base_account.account_number, base_account.sequence);
760
761        // Build transaction with all messages
762        let tx_builder =
763            TxBuilder::new(self.chain_id.clone(), "adydx".to_string()).map_err(|e| {
764                DydxError::Grpc(Box::new(tonic::Status::internal(format!(
765                    "TxBuilder init failed: {e}"
766                ))))
767            })?;
768
769        // Convert all messages to Any
770        let any_msgs: Vec<_> = msgs.into_iter().map(|m| m.to_any()).collect();
771
772        let auth_ids = if self.authenticator_ids.is_empty() {
773            None
774        } else {
775            Some(self.authenticator_ids.as_slice())
776        };
777        let tx_raw = tx_builder
778            .build_transaction(&account, any_msgs, None, auth_ids)
779            .map_err(|e| {
780                DydxError::Grpc(Box::new(tonic::Status::internal(format!(
781                    "Failed to build tx: {e}"
782                ))))
783            })?;
784
785        let tx_bytes = tx_raw.to_bytes().map_err(|e| {
786            DydxError::Grpc(Box::new(tonic::Status::internal(format!(
787                "Failed to serialize tx: {e}"
788            ))))
789        })?;
790
791        let mut grpc_client = self.grpc_client.clone();
792        let tx_hash = grpc_client.broadcast_tx(tx_bytes).await.map_err(|e| {
793            DydxError::Grpc(Box::new(tonic::Status::internal(format!(
794                "Broadcast failed: {e}"
795            ))))
796        })?;
797
798        log::info!("Batch cancelled {count} orders: tx_hash={tx_hash}");
799        Ok(())
800    }
801}
802
803#[cfg(test)]
804mod tests {
805    use rstest::rstest;
806
807    use super::*;
808
809    #[rstest]
810    fn test_cancel_orders_batch_builds_multiple_messages() {
811        let btc_id = InstrumentId::from("BTC-USD-PERP.DYDX");
812        let eth_id = InstrumentId::from("ETH-USD-PERP.DYDX");
813        let orders = [(btc_id, 100u32), (btc_id, 101u32), (eth_id, 200u32)];
814
815        assert_eq!(orders.len(), 3);
816        assert_eq!(orders[0], (btc_id, 100));
817        assert_eq!(orders[1], (btc_id, 101));
818        assert_eq!(orders[2], (eth_id, 200));
819    }
820
821    #[rstest]
822    fn test_cancel_orders_batch_empty_returns_ok() {
823        let orders: [(InstrumentId, u32); 0] = [];
824        assert!(orders.is_empty());
825    }
826
827    #[rstest]
828    fn test_conditional_order_expiration_with_explicit_timestamp() {
829        let expire_ts = 1735689600i64; // 2025-01-01 00:00:00 UTC
830        let result =
831            calculate_conditional_order_expiration(TimeInForce::Gtd, Some(expire_ts)).unwrap();
832        assert_eq!(result.timestamp(), expire_ts);
833    }
834
835    #[rstest]
836    fn test_conditional_order_expiration_gtc_uses_90_days() {
837        let now = Utc::now();
838        let result = calculate_conditional_order_expiration(TimeInForce::Gtc, None).unwrap();
839
840        let expected_min = now + Duration::days(89);
841        let expected_max = now + Duration::days(91);
842
843        assert!(result > expected_min);
844        assert!(result < expected_max);
845    }
846
847    #[rstest]
848    fn test_conditional_order_expiration_gtd_without_timestamp_uses_90_days() {
849        let now = Utc::now();
850        let result = calculate_conditional_order_expiration(TimeInForce::Gtd, None).unwrap();
851
852        let expected_min = now + Duration::days(89);
853        let expected_max = now + Duration::days(91);
854
855        assert!(result > expected_min);
856        assert!(result < expected_max);
857    }
858
859    #[rstest]
860    fn test_conditional_order_expiration_ioc_uses_1_hour() {
861        let now = Utc::now();
862        let result = calculate_conditional_order_expiration(TimeInForce::Ioc, None).unwrap();
863
864        let expected_min = now + Duration::minutes(59);
865        let expected_max = now + Duration::minutes(61);
866
867        assert!(result > expected_min);
868        assert!(result < expected_max);
869    }
870
871    #[rstest]
872    fn test_conditional_order_expiration_fok_uses_1_hour() {
873        let now = Utc::now();
874        let result = calculate_conditional_order_expiration(TimeInForce::Fok, None).unwrap();
875
876        let expected_min = now + Duration::minutes(59);
877        let expected_max = now + Duration::minutes(61);
878
879        assert!(result > expected_min);
880        assert!(result < expected_max);
881    }
882
883    #[rstest]
884    fn test_conditional_order_expiration_day_uses_90_days() {
885        let now = Utc::now();
886        let result = calculate_conditional_order_expiration(TimeInForce::Day, None).unwrap();
887
888        let expected_min = now + Duration::days(89);
889        let expected_max = now + Duration::days(91);
890
891        assert!(result > expected_min);
892        assert!(result < expected_max);
893    }
894
895    #[rstest]
896    fn test_conditional_order_expiration_invalid_timestamp_returns_error() {
897        // i64::MAX is beyond the valid range for chrono timestamps
898        let result = calculate_conditional_order_expiration(TimeInForce::Gtd, Some(i64::MAX));
899        assert!(result.is_err());
900    }
901
902    #[rstest]
903    fn test_conditional_order_type_debug_format() {
904        assert_eq!(
905            format!("{:?}", ConditionalOrderType::StopMarket),
906            "StopMarket"
907        );
908        assert_eq!(
909            format!("{:?}", ConditionalOrderType::StopLimit),
910            "StopLimit"
911        );
912        assert_eq!(
913            format!("{:?}", ConditionalOrderType::TakeProfitMarket),
914            "TakeProfitMarket"
915        );
916        assert_eq!(
917            format!("{:?}", ConditionalOrderType::TakeProfitLimit),
918            "TakeProfitLimit"
919        );
920    }
921}
nautilus_dydx/execution/submitter.rs

nautilus_dydx/execution/
submitter.rs
