nautilus_dydx/execution/
submitter.rs

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

nautilus_dydx/execution/
submitter.rs
