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        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        tracing::debug!(
730            "Broadcasting {} with {} bytes, account_seq={}",
731            operation,
732            tx_bytes.len(),
733            account.sequence_number
734        );
735
736        let mut grpc_client = self.grpc_client.clone();
737        let tx_hash = grpc_client.broadcast_tx(tx_bytes).await.map_err(|e| {
738            tracing::error!("gRPC broadcast failed for {}: {}", operation, e);
739            DydxError::Grpc(Box::new(tonic::Status::internal(format!(
740                "Broadcast failed: {e}"
741            ))))
742        })?;
743
744        tracing::info!("{} successfully: tx_hash={}", operation, tx_hash);
745        Ok(())
746    }
747
748    /// Broadcast order placement message via gRPC.
749    async fn broadcast_order_message(
750        &self,
751        wallet: &Wallet,
752        msg: MsgPlaceOrder,
753    ) -> Result<(), DydxError> {
754        self.broadcast_tx_message(wallet, msg, "Order placed").await
755    }
756
757    /// Broadcast order cancellation message via gRPC.
758    async fn broadcast_cancel_message(
759        &self,
760        wallet: &Wallet,
761        msg: MsgCancelOrder,
762    ) -> Result<(), DydxError> {
763        self.broadcast_tx_message(wallet, msg, "Order cancelled")
764            .await
765    }
766
767    /// Broadcast multiple order cancellation messages in a single transaction.
768    async fn broadcast_cancel_messages_batch(
769        &self,
770        wallet: &Wallet,
771        msgs: Vec<MsgCancelOrder>,
772    ) -> Result<(), DydxError> {
773        let count = msgs.len();
774
775        // Derive account for signing
776        let mut account = wallet
777            .account_offline(0)
778            .map_err(|e| DydxError::Wallet(format!("Failed to derive account: {e}")))?;
779
780        // Fetch current account info
781        let mut grpc_client = self.grpc_client.clone();
782        let base_account = grpc_client
783            .get_account(&self.wallet_address)
784            .await
785            .map_err(|e| {
786                DydxError::Grpc(Box::new(tonic::Status::internal(format!(
787                    "Failed to fetch account info: {e}"
788                ))))
789            })?;
790
791        account.set_account_info(base_account.account_number, base_account.sequence);
792
793        // Build transaction with all messages
794        let tx_builder =
795            TxBuilder::new(self.chain_id.clone(), "adydx".to_string()).map_err(|e| {
796                DydxError::Grpc(Box::new(tonic::Status::internal(format!(
797                    "TxBuilder init failed: {e}"
798                ))))
799            })?;
800
801        // Convert all messages to Any
802        let any_msgs: Vec<_> = msgs.into_iter().map(|m| m.to_any()).collect();
803
804        let auth_ids = if self.authenticator_ids.is_empty() {
805            None
806        } else {
807            Some(self.authenticator_ids.as_slice())
808        };
809        let tx_raw = tx_builder
810            .build_transaction(&account, any_msgs, None, auth_ids)
811            .map_err(|e| {
812                DydxError::Grpc(Box::new(tonic::Status::internal(format!(
813                    "Failed to build tx: {e}"
814                ))))
815            })?;
816
817        let tx_bytes = tx_raw.to_bytes().map_err(|e| {
818            DydxError::Grpc(Box::new(tonic::Status::internal(format!(
819                "Failed to serialize tx: {e}"
820            ))))
821        })?;
822
823        let mut grpc_client = self.grpc_client.clone();
824        let tx_hash = grpc_client.broadcast_tx(tx_bytes).await.map_err(|e| {
825            DydxError::Grpc(Box::new(tonic::Status::internal(format!(
826                "Broadcast failed: {e}"
827            ))))
828        })?;
829
830        tracing::info!("Batch cancelled {} orders: tx_hash={}", count, tx_hash);
831        Ok(())
832    }
833}
834
835#[cfg(test)]
836mod tests {
837    use rstest::rstest;
838
839    use super::*;
840
841    #[rstest]
842    fn test_cancel_orders_batch_builds_multiple_messages() {
843        let btc_id = InstrumentId::from("BTC-USD-PERP.DYDX");
844        let eth_id = InstrumentId::from("ETH-USD-PERP.DYDX");
845        let orders = [(btc_id, 100u32), (btc_id, 101u32), (eth_id, 200u32)];
846
847        assert_eq!(orders.len(), 3);
848        assert_eq!(orders[0], (btc_id, 100));
849        assert_eq!(orders[1], (btc_id, 101));
850        assert_eq!(orders[2], (eth_id, 200));
851    }
852
853    #[rstest]
854    fn test_cancel_orders_batch_empty_returns_ok() {
855        let orders: [(InstrumentId, u32); 0] = [];
856        assert!(orders.is_empty());
857    }
858
859    #[rstest]
860    fn test_conditional_order_expiration_with_explicit_timestamp() {
861        let expire_ts = 1735689600i64; // 2025-01-01 00:00:00 UTC
862        let result =
863            calculate_conditional_order_expiration(TimeInForce::Gtd, Some(expire_ts)).unwrap();
864        assert_eq!(result.timestamp(), expire_ts);
865    }
866
867    #[rstest]
868    fn test_conditional_order_expiration_gtc_uses_90_days() {
869        let now = Utc::now();
870        let result = calculate_conditional_order_expiration(TimeInForce::Gtc, None).unwrap();
871
872        let expected_min = now + Duration::days(89);
873        let expected_max = now + Duration::days(91);
874
875        assert!(result > expected_min);
876        assert!(result < expected_max);
877    }
878
879    #[rstest]
880    fn test_conditional_order_expiration_gtd_without_timestamp_uses_90_days() {
881        let now = Utc::now();
882        let result = calculate_conditional_order_expiration(TimeInForce::Gtd, None).unwrap();
883
884        let expected_min = now + Duration::days(89);
885        let expected_max = now + Duration::days(91);
886
887        assert!(result > expected_min);
888        assert!(result < expected_max);
889    }
890
891    #[rstest]
892    fn test_conditional_order_expiration_ioc_uses_1_hour() {
893        let now = Utc::now();
894        let result = calculate_conditional_order_expiration(TimeInForce::Ioc, None).unwrap();
895
896        let expected_min = now + Duration::minutes(59);
897        let expected_max = now + Duration::minutes(61);
898
899        assert!(result > expected_min);
900        assert!(result < expected_max);
901    }
902
903    #[rstest]
904    fn test_conditional_order_expiration_fok_uses_1_hour() {
905        let now = Utc::now();
906        let result = calculate_conditional_order_expiration(TimeInForce::Fok, None).unwrap();
907
908        let expected_min = now + Duration::minutes(59);
909        let expected_max = now + Duration::minutes(61);
910
911        assert!(result > expected_min);
912        assert!(result < expected_max);
913    }
914
915    #[rstest]
916    fn test_conditional_order_expiration_day_uses_90_days() {
917        let now = Utc::now();
918        let result = calculate_conditional_order_expiration(TimeInForce::Day, None).unwrap();
919
920        let expected_min = now + Duration::days(89);
921        let expected_max = now + Duration::days(91);
922
923        assert!(result > expected_min);
924        assert!(result < expected_max);
925    }
926
927    #[rstest]
928    fn test_conditional_order_expiration_invalid_timestamp_returns_error() {
929        // i64::MAX is beyond the valid range for chrono timestamps
930        let result = calculate_conditional_order_expiration(TimeInForce::Gtd, Some(i64::MAX));
931        assert!(result.is_err());
932    }
933
934    #[rstest]
935    fn test_conditional_order_type_debug_format() {
936        assert_eq!(
937            format!("{:?}", ConditionalOrderType::StopMarket),
938            "StopMarket"
939        );
940        assert_eq!(
941            format!("{:?}", ConditionalOrderType::StopLimit),
942            "StopLimit"
943        );
944        assert_eq!(
945            format!("{:?}", ConditionalOrderType::TakeProfitMarket),
946            "TakeProfitMarket"
947        );
948        assert_eq!(
949            format!("{:?}", ConditionalOrderType::TakeProfitLimit),
950            "TakeProfitLimit"
951        );
952    }
953}
nautilus_dydx/execution/submitter.rs

nautilus_dydx/execution/
submitter.rs
