nautilus_dydx/execution/
mod.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Live execution client implementation for the dYdX adapter.
17//!
18//! This module provides the execution client for submitting orders, cancellations,
19//! and managing positions on dYdX v4.
20//!
21//! # Order Types
22//!
23//! dYdX supports the following order types:
24//!
25//! - **Market**: Execute immediately at best available price.
26//! - **Limit**: Execute at specified price or better.
27//! - **Stop Market**: Triggered when price crosses stop price, then executes as market order.
28//! - **Stop Limit**: Triggered when price crosses stop price, then places limit order.
29//! - **Take Profit Market**: Close position at profit target, executes as market order.
30//! - **Take Profit Limit**: Close position at profit target, places limit order.
31//!
32//! See <https://docs.dydx.xyz/concepts/trading/orders#types> for details.
33//!
34//! # Order Lifetimes
35//!
36//! Orders can be short-term (expire by block height) or long-term/stateful (expire by timestamp).
37//! Conditional orders (Stop/TakeProfit) are always stateful.
38//!
39//! See <https://docs.dydx.xyz/concepts/trading/orders#short-term-vs-long-term> for details.
40
41use std::{
42    sync::{
43        Arc,
44        atomic::{AtomicU64, Ordering},
45    },
46    time::{Duration, Instant},
47};
48
49use ahash::AHashMap;
50use anyhow::Context;
51use async_trait::async_trait;
52use dashmap::DashMap;
53use futures_util::{Stream, StreamExt, pin_mut};
54use nautilus_common::{
55    clients::ExecutionClient,
56    live::runner::get_exec_event_sender,
57    messages::execution::{
58        BatchCancelOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
59        GenerateOrderStatusReport, GenerateOrderStatusReports, GeneratePositionStatusReports,
60        ModifyOrder, QueryAccount, QueryOrder, SubmitOrder, SubmitOrderList,
61    },
62};
63use nautilus_core::{
64    DurationNanos, Params, UUID4, UnixNanos,
65    time::{AtomicTime, get_atomic_clock_realtime},
66};
67use nautilus_live::{
68    ExecutionClientCore, ExecutionEventEmitter, SocketControlFactory,
69    execution::reports::retain_order_status_reports,
70    task::{TaskGroup, TaskGroupGuard},
71};
72use nautilus_model::{
73    accounts::AccountAny,
74    enums::{AccountType, OmsType, OrderStatus, OrderType, TimeInForce},
75    events::{
76        AccountState, OrderAccepted, OrderCanceled, OrderDeniedReason, OrderEventAny, OrderExpired,
77    },
78    identifiers::{
79        AccountId, ClientId, ClientOrderId, InstrumentId, StrategyId, Symbol, Venue, VenueOrderId,
80    },
81    instruments::{Instrument, InstrumentAny},
82    orders::{Order, OrderAny},
83    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
84    types::{AccountBalance, Currency, MarginBalance, Money},
85};
86use nautilus_network::retry::RetryConfig;
87use parking_lot::Mutex;
88use rust_decimal::Decimal;
89
90use crate::{
91    common::{
92        consts::DYDX_VENUE,
93        credential::{DydxCredential, credential_env_vars},
94        instrument_cache::InstrumentCache,
95        parse::nanos_to_secs_i64,
96    },
97    config::DydxAdapterConfig,
98    error::DydxError,
99    execution::{
100        broadcaster::TxBroadcaster,
101        encoder::ClientOrderIdEncoder,
102        order_builder::OrderMessageBuilder,
103        tx_manager::TransactionManager,
104        types::{LimitOrderParams, OrderContext},
105    },
106    grpc::{DydxGrpcClient, SHORT_TERM_ORDER_MAXIMUM_LIFETIME, types::ChainId},
107    http::{
108        client::DydxHttpClient,
109        parse::{
110            parse_account_state, parse_fill_report, parse_order_status_report,
111            parse_position_status_report,
112        },
113    },
114    websocket::{
115        DydxWsDispatchState, OrderIdentity,
116        client::DydxWebSocketClient,
117        enums::DydxWsOutputMessage,
118        fill_report_to_order_filled,
119        parse::{parse_ws_fill_report, parse_ws_order_report, parse_ws_position_report},
120    },
121};
122
123pub mod block_time;
124pub mod broadcaster;
125pub mod encoder;
126pub mod order_builder;
127pub mod submitter;
128pub mod tx_manager;
129pub mod types;
130pub mod wallet;
131
132use block_time::BlockTimeMonitor;
133
134const DYDX_INDEXER_REPORT_LIMIT: u32 = 1_000;
135
136/// Live execution client for the dYdX v4 exchange adapter.
137///
138/// Supports Market, Limit, Stop Market, Stop Limit, Take Profit Market (MarketIfTouched),
139/// and Take Profit Limit (LimitIfTouched) orders via gRPC. Trailing stops are NOT supported
140/// by the dYdX v4 protocol. dYdX requires u32 client IDs - strings are hashed to fit.
141///
142/// # Architecture
143///
144/// The client follows a two-layer execution model:
145/// 1. **Synchronous validation** - Immediate checks and event generation.
146/// 2. **Async submission** - Non-blocking gRPC calls via `TransactionManager`, `TxBroadcaster`, and `OrderMessageBuilder`.
147#[derive(Debug)]
148pub struct DydxExecutionClient {
149    core: ExecutionClientCore,
150    clock: &'static AtomicTime,
151    config: DydxAdapterConfig,
152    emitter: ExecutionEventEmitter,
153    http_client: DydxHttpClient,
154    ws_client: DydxWebSocketClient,
155    grpc_client: Arc<tokio::sync::RwLock<Option<DydxGrpcClient>>>,
156    instrument_cache: Arc<InstrumentCache>,
157    block_time_monitor: Arc<BlockTimeMonitor>,
158    oracle_prices: Arc<DashMap<InstrumentId, Decimal>>,
159    encoder: Arc<ClientOrderIdEncoder>,
160    dispatch_state: Arc<DydxWsDispatchState>,
161    order_contexts: Arc<DashMap<u32, OrderContext>>,
162    order_id_map: Arc<DashMap<String, (u32, u32)>>,
163    wallet_address: String,
164    subaccount_number: u32,
165    tx_manager: Option<Arc<TransactionManager>>,
166    broadcaster: Option<Arc<TxBroadcaster>>,
167    order_builder: Option<Arc<OrderMessageBuilder>>,
168    session_tasks: TaskGroup,
169    pending_tasks: TaskGroup,
170    shutdown_errors: Vec<String>,
171    pending_task_labels: Arc<Mutex<AHashMap<u64, &'static str>>>,
172    next_pending_task_id: AtomicU64,
173}
174
175impl DydxExecutionClient {
176    /// Creates a new [`DydxExecutionClient`].
177    ///
178    /// # Errors
179    ///
180    /// Returns an error if credentials are not found or client fails to construct.
181    pub fn new(
182        core: ExecutionClientCore,
183        config: DydxAdapterConfig,
184        wallet_address: String,
185        subaccount_number: u32,
186    ) -> anyhow::Result<Self> {
187        config.validate()?;
188
189        let trader_id = core.trader_id;
190        let account_id = core.account_id;
191        let clock = get_atomic_clock_realtime();
192        let emitter =
193            ExecutionEventEmitter::new(clock, trader_id, account_id, AccountType::Margin, None);
194
195        let retry_config = RetryConfig {
196            max_retries: config.max_retries,
197            initial_delay_ms: config.retry_delay_initial_ms,
198            max_delay_ms: config.retry_delay_max_ms,
199            ..Default::default()
200        };
201        let http_client = DydxHttpClient::new(
202            Some(config.base_url.clone()),
203            config.timeout_secs,
204            config
205                .proxy_url
206                .as_ref()
207                .map(|value| value.expose_secret().to_owned()),
208            config.network,
209            Some(retry_config),
210        )?;
211
212        let instrument_cache = http_client.instrument_cache().clone();
213
214        let credential = DydxCredential::resolve(
215            config
216                .private_key
217                .as_ref()
218                .map(|value| value.expose_secret()),
219            config.network,
220            config.authenticator_ids.clone(),
221        )?
222        .ok_or_else(|| anyhow::anyhow!("Credentials required for execution client"))?;
223
224        let ws_client = DydxWebSocketClient::new_private_with_cache(
225            config.ws_url.clone(),
226            credential,
227            core.account_id,
228            instrument_cache.clone(),
229            Some(20),
230            config.transport_backend,
231            config
232                .proxy_url
233                .as_ref()
234                .map(|value| value.expose_secret().to_owned()),
235        )
236        .with_socket_factory(SocketControlFactory::new(core.client_id, Some(*DYDX_VENUE)));
237
238        let grpc_client = Arc::new(tokio::sync::RwLock::new(None));
239
240        let session_tasks = TaskGroup::new();
241        let pending_tasks = TaskGroup::new();
242
243        Ok(Self {
244            core,
245            clock,
246            config,
247            emitter,
248            http_client,
249            ws_client,
250            grpc_client,
251            instrument_cache,
252            block_time_monitor: Arc::new(BlockTimeMonitor::new()),
253            oracle_prices: Arc::new(DashMap::new()),
254            encoder: Arc::new(ClientOrderIdEncoder::new()),
255            dispatch_state: Arc::new(DydxWsDispatchState::default()),
256            order_contexts: Arc::new(DashMap::new()),
257            order_id_map: Arc::new(DashMap::new()),
258            wallet_address,
259            subaccount_number,
260            tx_manager: None,
261            broadcaster: None,
262            order_builder: None,
263            session_tasks,
264            pending_tasks,
265            shutdown_errors: Vec::new(),
266            pending_task_labels: Arc::new(Mutex::new(AHashMap::new())),
267            next_pending_task_id: AtomicU64::new(0),
268        })
269    }
270
271    fn resolve_private_key(config: &DydxAdapterConfig) -> anyhow::Result<String> {
272        let (private_key_env, _) = credential_env_vars(config.network);
273
274        if let Some(ref pk) = config.private_key
275            && !pk.expose_secret().trim().is_empty()
276        {
277            return Ok(pk.expose_secret().to_owned());
278        }
279
280        if let Some(pk) = std::env::var(private_key_env)
281            .ok()
282            .filter(|s| !s.trim().is_empty())
283        {
284            return Ok(pk);
285        }
286
287        anyhow::bail!("{private_key_env} not found in config or environment")
288    }
289
290    fn register_order_context(&self, client_id_u32: u32, context: OrderContext) {
291        self.order_contexts.insert(client_id_u32, context);
292    }
293
294    fn get_order_context(&self, client_id_u32: u32) -> Option<OrderContext> {
295        self.order_contexts
296            .get(&client_id_u32)
297            .map(|r| r.value().clone())
298    }
299
300    fn get_chain_id(&self) -> ChainId {
301        self.config.get_chain_id()
302    }
303
304    fn spawn_ws_stream_handler(
305        &self,
306        stream: impl Stream<Item = DydxWsOutputMessage> + Send + 'static,
307    ) -> anyhow::Result<()> {
308        if !self.session_tasks.is_empty() {
309            anyhow::bail!("dYdX WebSocket stream task is already registered");
310        }
311
312        log::debug!("Starting execution WebSocket message processing task");
313
314        let trader_id = self.core.trader_id;
315        let account_id = self.core.account_id;
316        let instrument_cache = self.instrument_cache.clone();
317        let oracle_prices = self.oracle_prices.clone();
318        let encoder = self.encoder.clone();
319        let order_contexts = self.order_contexts.clone();
320        let order_id_map = self.order_id_map.clone();
321        let dispatch_state = self.dispatch_state.clone();
322        let block_time_monitor = self.block_time_monitor.clone();
323        let emitter = self.emitter.clone();
324        let clock = self.clock;
325
326        let future = async move {
327            log::debug!("Execution WebSocket message loop started");
328
329            let mut cum_fill_totals: AHashMap<VenueOrderId, (Decimal, Decimal)> = AHashMap::new();
330
331            pin_mut!(stream);
332            while let Some(msg) = stream.next().await {
333                match msg {
334                    DydxWsOutputMessage::SubaccountSubscribed(msg) => {
335                        log::debug!("Parsing subaccount subscription with full context");
336
337                        let inst_map = instrument_cache.to_instrument_id_map();
338
339                        let oracle_map: std::collections::HashMap<_, _> = oracle_prices
340                            .iter()
341                            .map(|entry| (*entry.key(), *entry.value()))
342                            .collect();
343
344                        let ts_init = clock.get_time_ns();
345                        let ts_event = ts_init;
346
347                        if let Some(ref subaccount) = msg.contents.subaccount {
348                            match parse_account_state(
349                                subaccount,
350                                account_id,
351                                &inst_map,
352                                &oracle_map,
353                                ts_event,
354                                ts_init,
355                            ) {
356                                Ok(account_state) => {
357                                    log::debug!(
358                                        "Parsed account state: {} balance(s), {} margin(s)",
359                                        account_state.balances.len(),
360                                        account_state.margins.len()
361                                    );
362                                    emitter.send_account_state(account_state);
363                                }
364                                Err(e) => {
365                                    log::error!("Failed to parse account state: {e}");
366                                }
367                            }
368
369                            if let Some(ref positions) = subaccount.open_perpetual_positions {
370                                log::debug!(
371                                    "Parsing {} position(s) from subscription",
372                                    positions.len()
373                                );
374
375                                for (market, ws_position) in positions {
376                                    match parse_ws_position_report(
377                                        ws_position,
378                                        &instrument_cache,
379                                        account_id,
380                                        ts_init,
381                                    ) {
382                                        Ok(report) => {
383                                            log::debug!(
384                                                "Parsed position report: {} {} {} {}",
385                                                report.instrument_id,
386                                                report.position_side,
387                                                report.quantity,
388                                                market
389                                            );
390                                            emitter.send_position_report(report);
391                                        }
392                                        Err(e) => {
393                                            log::error!(
394                                                "Failed to parse WebSocket position for {market}: {e}"
395                                            );
396                                        }
397                                    }
398                                }
399                            }
400                        } else {
401                            log::warn!(
402                                "Subaccount subscription without initial state (new/empty subaccount)"
403                            );
404
405                            let currency =
406                                Currency::get_or_create_crypto_with_context("USDC", None);
407                            let zero = Money::zero(currency);
408                            let balance = AccountBalance::new_checked(zero, zero, zero)
409                                .expect("zero balance should always be valid");
410                            let account_state = AccountState::new(
411                                account_id,
412                                AccountType::Margin,
413                                vec![balance],
414                                vec![],
415                                true,
416                                UUID4::new(),
417                                ts_init,
418                                ts_init,
419                                None,
420                            );
421                            emitter.send_account_state(account_state);
422                        }
423                    }
424                    DydxWsOutputMessage::SubaccountsChannelData(data) => {
425                        log::debug!(
426                            "Processing subaccounts channel data (orders={:?}, fills={:?})",
427                            data.contents.orders.as_ref().map(|o| o.len()),
428                            data.contents.fills.as_ref().map(|f| f.len())
429                        );
430                        let ts_init = clock.get_time_ns();
431
432                        let mut terminal_orders: Vec<(u32, u32, String)> = Vec::new();
433                        let mut pending_order_reports = Vec::new();
434
435                        if let Some(ref orders) = data.contents.orders {
436                            for ws_order in orders {
437                                log::debug!(
438                                    "Parsing WS order: clob_pair_id={}, status={:?}, client_id={}",
439                                    ws_order.clob_pair_id,
440                                    ws_order.status,
441                                    ws_order.client_id
442                                );
443
444                                if let Ok(client_id_u32) = ws_order.client_id.parse::<u32>() {
445                                    let client_meta = ws_order
446                                        .client_metadata
447                                        .as_ref()
448                                        .and_then(|s| s.parse::<u32>().ok())
449                                        .unwrap_or(crate::grpc::DEFAULT_RUST_CLIENT_METADATA);
450                                    order_id_map
451                                        .insert(ws_order.id.clone(), (client_id_u32, client_meta));
452                                }
453
454                                match parse_ws_order_report(
455                                    ws_order,
456                                    &instrument_cache,
457                                    &order_contexts,
458                                    &encoder,
459                                    account_id,
460                                    ts_init,
461                                ) {
462                                    Ok(report) => {
463                                        if report.order_status.is_closed()
464                                            && let Ok(cid) = ws_order.client_id.parse::<u32>()
465                                        {
466                                            let meta = ws_order
467                                                .client_metadata
468                                                .as_ref()
469                                                .and_then(|s| s.parse::<u32>().ok())
470                                                .unwrap_or(
471                                                    crate::grpc::DEFAULT_RUST_CLIENT_METADATA,
472                                                );
473                                            terminal_orders.push((cid, meta, ws_order.id.clone()));
474                                        }
475                                        let order_side = report
476                                            .order_side
477                                            .as_ref()
478                                            .map_or("NO_ORDER_SIDE", AsRef::as_ref);
479                                        log::debug!(
480                                            "Parsed order report: {} {} {:?} qty={} client_order_id={:?}",
481                                            report.instrument_id,
482                                            order_side,
483                                            report.order_status,
484                                            report.quantity,
485                                            report.client_order_id
486                                        );
487                                        pending_order_reports.push(report);
488                                    }
489                                    Err(e) => {
490                                        log::error!("Failed to parse WebSocket order: {e}");
491                                    }
492                                }
493                            }
494                        }
495
496                        // Emit fills before order status for correct reconciliation
497                        if let Some(ref fills) = data.contents.fills {
498                            for ws_fill in fills {
499                                match parse_ws_fill_report(
500                                    ws_fill,
501                                    &instrument_cache,
502                                    &order_id_map,
503                                    &order_contexts,
504                                    &encoder,
505                                    account_id,
506                                    ts_init,
507                                ) {
508                                    Ok(report) => {
509                                        log::debug!(
510                                            "Parsed fill report: {} {} {} @ {} client_order_id={:?}",
511                                            report.instrument_id,
512                                            report.venue_order_id,
513                                            report.last_qty,
514                                            report.last_px,
515                                            report.client_order_id
516                                        );
517
518                                        let identity = report.client_order_id.and_then(|cid| {
519                                            dispatch_state
520                                                .order_identities
521                                                .get(&cid)
522                                                .map(|r| (cid, r.clone()))
523                                        });
524
525                                        if let Some((cid, ident)) = identity {
526                                            if !dispatch_state.emitted_accepted.contains(&cid) {
527                                                dispatch_state.insert_accepted(cid);
528                                                let accepted = OrderAccepted::new(
529                                                    trader_id,
530                                                    ident.strategy_id,
531                                                    ident.instrument_id,
532                                                    cid,
533                                                    report.venue_order_id,
534                                                    account_id,
535                                                    UUID4::new(),
536                                                    ts_init,
537                                                    ts_init,
538                                                    false,
539                                                );
540                                                emitter.send_order_event(OrderEventAny::Accepted(
541                                                    accepted,
542                                                ));
543                                            }
544
545                                            dispatch_state.insert_filled(cid);
546                                            let instrument =
547                                                instrument_cache.get(&report.instrument_id);
548                                            let quote_currency = instrument
549                                                .map_or_else(Currency::USD, |i: InstrumentAny| {
550                                                    i.quote_currency()
551                                                });
552                                            let filled = fill_report_to_order_filled(
553                                                &report,
554                                                trader_id,
555                                                &ident,
556                                                quote_currency,
557                                            );
558                                            emitter.send_order_event(OrderEventAny::Filled(filled));
559                                        } else {
560                                            let entry = cum_fill_totals
561                                                .entry(report.venue_order_id)
562                                                .or_default();
563                                            let qty = report.last_qty.as_decimal();
564                                            entry.0 += report.last_px.as_decimal() * qty;
565                                            entry.1 += qty;
566                                            emitter.send_fill_report(report);
567                                        }
568                                    }
569                                    Err(e) => {
570                                        log::error!("Failed to parse WebSocket fill: {e}");
571                                    }
572                                }
573                            }
574                        }
575
576                        for report in &mut pending_order_reports {
577                            if let Some((notional, total_qty)) =
578                                cum_fill_totals.get(&report.venue_order_id)
579                                && !total_qty.is_zero()
580                            {
581                                report.avg_px = Some(notional / total_qty);
582                            }
583                        }
584
585                        for report in pending_order_reports {
586                            let identity = report.client_order_id.and_then(|cid| {
587                                dispatch_state
588                                    .order_identities
589                                    .get(&cid)
590                                    .map(|r| (cid, r.clone()))
591                            });
592
593                            if let Some((cid, ident)) = identity {
594                                match report.order_status {
595                                    OrderStatus::Accepted => {
596                                        if dispatch_state.emitted_accepted.contains(&cid)
597                                            || dispatch_state.filled_orders.contains(&cid)
598                                        {
599                                            log::debug!("Skipping duplicate Accepted for {cid}");
600                                            continue;
601                                        }
602                                        dispatch_state.insert_accepted(cid);
603                                        let accepted = OrderAccepted::new(
604                                            trader_id,
605                                            ident.strategy_id,
606                                            ident.instrument_id,
607                                            cid,
608                                            report.venue_order_id,
609                                            account_id,
610                                            UUID4::new(),
611                                            report.ts_last,
612                                            ts_init,
613                                            false,
614                                        );
615                                        emitter.send_order_event(OrderEventAny::Accepted(accepted));
616                                    }
617                                    OrderStatus::Canceled => {
618                                        if !dispatch_state.emitted_accepted.contains(&cid) {
619                                            dispatch_state.insert_accepted(cid);
620                                            let accepted = OrderAccepted::new(
621                                                trader_id,
622                                                ident.strategy_id,
623                                                ident.instrument_id,
624                                                cid,
625                                                report.venue_order_id,
626                                                account_id,
627                                                UUID4::new(),
628                                                ts_init,
629                                                ts_init,
630                                                false,
631                                            );
632                                            emitter.send_order_event(OrderEventAny::Accepted(
633                                                accepted,
634                                            ));
635                                        }
636                                        // Map venue cancel-on-expiry to OrderExpired
637                                        // (dYdX reports GTD expiry as a cancel event).
638                                        let is_expiry = report
639                                            .expire_time
640                                            .is_some_and(|exp| report.ts_last >= exp);
641
642                                        if is_expiry {
643                                            let expired = OrderExpired::new(
644                                                trader_id,
645                                                ident.strategy_id,
646                                                ident.instrument_id,
647                                                cid,
648                                                UUID4::new(),
649                                                report.ts_last,
650                                                ts_init,
651                                                false,
652                                                Some(report.venue_order_id),
653                                                Some(account_id),
654                                            );
655                                            emitter
656                                                .send_order_event(OrderEventAny::Expired(expired));
657                                        } else {
658                                            let canceled = OrderCanceled::new(
659                                                trader_id,
660                                                ident.strategy_id,
661                                                ident.instrument_id,
662                                                cid,
663                                                UUID4::new(),
664                                                report.ts_last,
665                                                ts_init,
666                                                false,
667                                                Some(report.venue_order_id),
668                                                Some(account_id),
669                                                None,
670                                            );
671                                            emitter.send_order_event(OrderEventAny::Canceled(
672                                                canceled,
673                                            ));
674                                        }
675                                        dispatch_state.cleanup_terminal(&cid);
676                                    }
677                                    OrderStatus::Filled => {
678                                        // Fills are emitted before the order status
679                                        dispatch_state.cleanup_terminal(&cid);
680                                    }
681                                    OrderStatus::Expired => {
682                                        // Reached when the parser reclassifies a venue
683                                        // cancel-on-expiry to `Expired` (HTTP path) or
684                                        // when the venue reports `Expired` directly. Mirror
685                                        // the Canceled arm's Accepted-synthesis so the
686                                        // strategy sees a complete lifecycle.
687                                        if !dispatch_state.emitted_accepted.contains(&cid) {
688                                            dispatch_state.insert_accepted(cid);
689                                            let accepted = OrderAccepted::new(
690                                                trader_id,
691                                                ident.strategy_id,
692                                                ident.instrument_id,
693                                                cid,
694                                                report.venue_order_id,
695                                                account_id,
696                                                UUID4::new(),
697                                                ts_init,
698                                                ts_init,
699                                                false,
700                                            );
701                                            emitter.send_order_event(OrderEventAny::Accepted(
702                                                accepted,
703                                            ));
704                                        }
705                                        let expired = OrderExpired::new(
706                                            trader_id,
707                                            ident.strategy_id,
708                                            ident.instrument_id,
709                                            cid,
710                                            UUID4::new(),
711                                            report.ts_last,
712                                            ts_init,
713                                            false,
714                                            Some(report.venue_order_id),
715                                            Some(account_id),
716                                        );
717                                        emitter.send_order_event(OrderEventAny::Expired(expired));
718                                        dispatch_state.cleanup_terminal(&cid);
719                                    }
720                                    _ => {
721                                        emitter.send_order_status_report(report);
722                                    }
723                                }
724                            } else {
725                                emitter.send_order_status_report(report);
726                            }
727                        }
728
729                        for (client_id, client_metadata, order_id) in terminal_orders {
730                            order_contexts.remove(&client_id);
731                            encoder.remove(client_id, client_metadata);
732                            order_id_map.remove(&order_id);
733                            cum_fill_totals.remove(&VenueOrderId::new(&order_id));
734                        }
735                    }
736                    DydxWsOutputMessage::Markets(contents) => {
737                        if let Some(ref oracle_map) = contents.oracle_prices {
738                            for (symbol_str, oracle_market) in oracle_map {
739                                let instrument_id = {
740                                    let symbol = format!("{symbol_str}-PERP");
741                                    InstrumentId::new(
742                                        Symbol::new(&symbol),
743                                        *crate::common::consts::DYDX_VENUE,
744                                    )
745                                };
746
747                                if instrument_cache.get(&instrument_id).is_some()
748                                    && let Ok(price_dec) =
749                                        oracle_market.oracle_price.parse::<Decimal>()
750                                {
751                                    oracle_prices.insert(instrument_id, price_dec);
752                                    log::trace!(
753                                        "Updated oracle price for {instrument_id}: {price_dec}"
754                                    );
755                                }
756                            }
757                        }
758
759                        if let Some(ref markets) = contents.markets {
760                            for (symbol_str, market_data) in markets {
761                                if let Some(oracle_price_str) = &market_data.oracle_price {
762                                    let instrument_id = {
763                                        let symbol = format!("{symbol_str}-PERP");
764                                        InstrumentId::new(
765                                            Symbol::new(&symbol),
766                                            *crate::common::consts::DYDX_VENUE,
767                                        )
768                                    };
769
770                                    if instrument_cache.get(&instrument_id).is_some()
771                                        && let Ok(price_dec) = oracle_price_str.parse::<Decimal>()
772                                    {
773                                        oracle_prices.insert(instrument_id, price_dec);
774                                    }
775                                }
776                            }
777                        }
778                    }
779                    DydxWsOutputMessage::BlockHeight { height, time } => {
780                        log::debug!("Block height update: {height} at {time}");
781                        block_time_monitor.record_block(height, time);
782                    }
783                    DydxWsOutputMessage::Error(err) => {
784                        log::warn!("WebSocket error: {err:?}");
785                    }
786                    DydxWsOutputMessage::Reconnected { .. } => {
787                        log::info!("WebSocket reconnected");
788                    }
789                    _ => {}
790                }
791            }
792            log::debug!("WebSocket message processing task ended");
793        };
794
795        self.session_tasks
796            .spawn(future)
797            .context("failed to register dYdX execution WebSocket stream task")?;
798        log::debug!("WebSocket stream handler started");
799        Ok(())
800    }
801
802    fn mark_instruments_initialized(&self) {
803        let count = self.instrument_cache.len();
804        self.core.set_instruments_initialized();
805        log::debug!("Instruments initialized: {count} instruments in shared cache");
806    }
807
808    fn get_instrument_by_market(&self, market: &str) -> Option<InstrumentAny> {
809        self.instrument_cache.get_by_market(market)
810    }
811
812    fn get_instrument_by_clob_pair_id(&self, clob_pair_id: u32) -> Option<InstrumentAny> {
813        let instrument = self.instrument_cache.get_by_clob_id(clob_pair_id);
814
815        if instrument.is_none() {
816            self.instrument_cache.log_missing_clob_pair_id(clob_pair_id);
817        }
818
819        instrument
820    }
821
822    fn get_execution_components(
823        &self,
824    ) -> anyhow::Result<(
825        Arc<TransactionManager>,
826        Arc<TxBroadcaster>,
827        Arc<OrderMessageBuilder>,
828    )> {
829        let tx_manager = self
830            .tx_manager
831            .as_ref()
832            .ok_or_else(|| {
833                anyhow::anyhow!("TransactionManager not initialized - call connect() first")
834            })?
835            .clone();
836        let broadcaster = self
837            .broadcaster
838            .as_ref()
839            .ok_or_else(|| anyhow::anyhow!("TxBroadcaster not initialized - call connect() first"))?
840            .clone();
841        let order_builder = self
842            .order_builder
843            .as_ref()
844            .ok_or_else(|| {
845                anyhow::anyhow!("OrderMessageBuilder not initialized - call connect() first")
846            })?
847            .clone();
848        Ok((tx_manager, broadcaster, order_builder))
849    }
850
851    fn spawn_task<F>(&self, label: &'static str, fut: F)
852    where
853        F: Future<Output = anyhow::Result<()>> + Send + 'static,
854    {
855        let future = async move {
856            if let Err(e) = fut.await {
857                log::error!("{label}: {e:?}");
858            }
859        };
860
861        self.spawn_labeled(label, future);
862    }
863
864    // Generates an `OrderRejected` only for a definitive CheckTx rejection; other
865    // failures remain in flight for reconciliation
866    fn spawn_order_task<F>(
867        &self,
868        label: &'static str,
869        strategy_id: StrategyId,
870        instrument_id: InstrumentId,
871        client_order_id: ClientOrderId,
872        fut: F,
873    ) where
874        F: Future<Output = anyhow::Result<()>> + Send + 'static,
875    {
876        let emitter = self.emitter.clone();
877        let clock = self.clock;
878
879        let future = async move {
880            if let Err(e) = fut.await {
881                if is_definitive_broadcast_rejection(&e) {
882                    let error_msg = format!("{label} failed: {e:?}");
883                    log::warn!("{error_msg}");
884
885                    let ts_event = clock.get_time_ns();
886                    emitter.emit_order_rejected_event(
887                        strategy_id,
888                        instrument_id,
889                        client_order_id,
890                        &error_msg,
891                        ts_event,
892                        false,
893                    );
894                } else {
895                    log::warn!(
896                        "Ambiguous dYdX {label} failure for {client_order_id}, awaiting reconciliation: {e:?}"
897                    );
898                }
899            }
900        };
901
902        self.spawn_labeled(label, future);
903    }
904
905    fn spawn_labeled<F>(&self, label: &'static str, future: F)
906    where
907        F: Future<Output = ()> + Send + 'static,
908    {
909        let id = self.next_pending_task_id.fetch_add(1, Ordering::Relaxed);
910        self.pending_task_labels.lock().insert(id, label);
911
912        let label_guard = PendingTaskLabel {
913            id,
914            labels: Arc::clone(&self.pending_task_labels),
915        };
916        let wrapped = async move {
917            let _label = label_guard;
918            future.await;
919        };
920
921        if let Err(e) = self.pending_tasks.spawn(wrapped) {
922            self.pending_task_labels.lock().remove(&id);
923            log::warn!("Skipping dYdX {label} after shutdown began: {e}");
924        }
925    }
926
927    fn begin_pending_shutdown(&self) {
928        let labels = self.pending_task_labels.lock();
929        let pending_cancels = labels
930            .values()
931            .filter(|label| label.contains("cancel"))
932            .count();
933        let pending_other = labels.len() - pending_cancels;
934
935        if pending_cancels > 0 {
936            log::warn!(
937                "Waiting for {pending_cancels} in-flight cancel task(s) before disconnect; orders may still be open on the venue"
938            );
939        }
940
941        if pending_other > 0 {
942            log::debug!("Waiting for {pending_other} other in-flight task(s) before disconnect");
943        }
944        drop(labels);
945
946        self.pending_tasks.begin_shutdown();
947    }
948
949    async fn finish_tasks(&self) -> anyhow::Result<()> {
950        let (session_result, pending_result) = tokio::join!(
951            self.session_tasks
952                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
953            self.pending_tasks
954                .finish_shutdown(Duration::from_secs(2), Duration::from_secs(2)),
955        );
956        session_result.context("failed to finish dYdX execution session tasks")?;
957        pending_result.context("failed to finish dYdX execution command tasks")?;
958        Ok(())
959    }
960
961    async fn prepare_task_groups(&mut self) -> anyhow::Result<()> {
962        if !self.session_tasks.is_open() || !self.pending_tasks.is_open() {
963            self.session_tasks.begin_shutdown();
964            self.begin_pending_shutdown();
965            self.ws_client.begin_shutdown();
966            self.finish_shutdown().await?;
967            self.session_tasks
968                .start_generation()
969                .context("failed to start dYdX execution session task generation")?;
970            self.pending_tasks
971                .start_generation()
972                .context("failed to start dYdX execution command task generation")?;
973        }
974        Ok(())
975    }
976
977    async fn finish_shutdown(&mut self) -> anyhow::Result<()> {
978        if let Err(e) = self
979            .ws_client
980            .disconnect()
981            .await
982            .context("failed to disconnect dYdX websocket")
983        {
984            self.shutdown_errors.push(e.to_string());
985        }
986
987        if let Err(e) = self.finish_tasks().await {
988            self.shutdown_errors.push(e.to_string());
989        }
990
991        if !self.shutdown_errors.is_empty() {
992            anyhow::bail!(std::mem::take(&mut self.shutdown_errors).join("; "));
993        }
994        Ok(())
995    }
996
997    async fn teardown_partial_connect(&mut self) -> anyhow::Result<()> {
998        self.begin_pending_shutdown();
999        self.session_tasks.begin_shutdown();
1000        self.ws_client.begin_shutdown();
1001        let shutdown_result = self.finish_shutdown().await;
1002        self.core.set_disconnected();
1003        shutdown_result
1004    }
1005
1006    fn send_modify_rejected(
1007        &self,
1008        strategy_id: StrategyId,
1009        instrument_id: InstrumentId,
1010        client_order_id: ClientOrderId,
1011        venue_order_id: Option<VenueOrderId>,
1012        reason: &str,
1013    ) {
1014        let ts_event = self.clock.get_time_ns();
1015        self.emitter.emit_order_modify_rejected_event(
1016            strategy_id,
1017            instrument_id,
1018            client_order_id,
1019            venue_order_id,
1020            reason,
1021            ts_event,
1022        );
1023    }
1024
1025    async fn await_account_registered(&self, timeout_secs: f64) -> anyhow::Result<()> {
1026        let account_id = self.core.account_id;
1027
1028        if self.core.cache().account(&account_id).is_some() {
1029            log::info!("Account {account_id} registered");
1030            return Ok(());
1031        }
1032
1033        let start = Instant::now();
1034        let timeout = Duration::from_secs_f64(timeout_secs);
1035        let interval = Duration::from_millis(10);
1036
1037        loop {
1038            tokio::time::sleep(interval).await;
1039
1040            if self.core.cache().account(&account_id).is_some() {
1041                log::info!("Account {account_id} registered");
1042                return Ok(());
1043            }
1044
1045            if start.elapsed() >= timeout {
1046                anyhow::bail!(
1047                    "Timeout waiting for account {account_id} to be registered after {timeout_secs}s"
1048                );
1049            }
1050        }
1051    }
1052}
1053
1054/// Broadcasts cancel orders with optimal partitioned strategy.
1055///
1056/// Partitions orders into short-term and long-term/conditional groups:
1057/// - Short-term -> single `MsgBatchCancel` via `broadcast_short_term()`
1058/// - Long-term/conditional -> batched `MsgCancelOrder` via `broadcast_with_retry()`
1059///
1060/// At most 2 gRPC calls regardless of order count or mix.
1061async fn broadcast_partitioned_cancels(
1062    orders: Vec<DydxCancelOrderRequest>,
1063    block_height: u32,
1064    tx_manager: Arc<TransactionManager>,
1065    broadcaster: Arc<TxBroadcaster>,
1066    order_builder: Arc<OrderMessageBuilder>,
1067    emitter: ExecutionEventEmitter,
1068    clock: &'static AtomicTime,
1069) -> anyhow::Result<()> {
1070    if orders.is_empty() {
1071        return Ok(());
1072    }
1073
1074    let (short_term_orders, long_term_orders): (Vec<_>, Vec<_>) = orders
1075        .into_iter()
1076        .partition(|order| order.order_flags == types::ORDER_FLAG_SHORT_TERM);
1077
1078    let mut errors = Vec::new();
1079
1080    if !short_term_orders.is_empty() {
1081        let st_pairs: Vec<_> = short_term_orders
1082            .iter()
1083            .map(|order| (order.instrument_id, order.client_id))
1084            .collect();
1085
1086        log::debug!(
1087            "Batch cancelling {} short-term orders with MsgBatchCancel",
1088            st_pairs.len()
1089        );
1090
1091        match order_builder.build_batch_cancel_short_term(&st_pairs, block_height) {
1092            Ok(msg) => {
1093                let operation = format!("BatchCancel {} short-term orders", st_pairs.len());
1094                match broadcaster
1095                    .broadcast_short_term(&tx_manager, vec![msg], &operation)
1096                    .await
1097                {
1098                    Ok(tx_hash) => {
1099                        log::debug!(
1100                            "Successfully batch cancelled {} short-term orders, tx_hash: {}",
1101                            st_pairs.len(),
1102                            tx_hash
1103                        );
1104                    }
1105                    Err(e) => {
1106                        let msg = format!("Short-term batch cancel failed: {e:?}");
1107                        log::warn!("{msg}");
1108
1109                        // A CheckTx rejection refuses the whole transaction
1110                        // atomically: a venue result for every cancel in it.
1111                        if e.is_definitive_broadcast_rejection() {
1112                            emit_partitioned_cancel_rejections(
1113                                &short_term_orders,
1114                                &emitter,
1115                                clock,
1116                                &msg,
1117                            );
1118                        }
1119                        errors.push(msg);
1120                    }
1121                }
1122            }
1123            Err(e) => {
1124                let msg = format!("Failed to build MsgBatchCancel: {e:?}");
1125                log::error!("{msg}");
1126                errors.push(msg);
1127            }
1128        }
1129    }
1130
1131    if !long_term_orders.is_empty() {
1132        let lt_tuples: Vec<_> = long_term_orders
1133            .iter()
1134            .map(|order| (order.instrument_id, order.client_id, order.order_flags))
1135            .collect();
1136
1137        log::debug!(
1138            "Batch cancelling {} long-term orders",
1139            long_term_orders.len(),
1140        );
1141
1142        match order_builder.build_cancel_orders_batch_with_flags(&lt_tuples, block_height) {
1143            Ok(cancel_msgs) => {
1144                let operation = format!("BatchCancel {} long-term orders", long_term_orders.len());
1145                match broadcaster
1146                    .broadcast_with_retry(&tx_manager, cancel_msgs, &operation)
1147                    .await
1148                {
1149                    Ok(tx_hash) => {
1150                        log::debug!(
1151                            "Successfully batch cancelled {} long-term orders, tx_hash: {}",
1152                            long_term_orders.len(),
1153                            tx_hash
1154                        );
1155                    }
1156                    Err(e) => {
1157                        let msg = format!("Long-term batch cancel failed: {e:?}");
1158                        log::warn!("{msg}");
1159
1160                        if e.is_definitive_broadcast_rejection() {
1161                            emit_partitioned_cancel_rejections(
1162                                &long_term_orders,
1163                                &emitter,
1164                                clock,
1165                                &msg,
1166                            );
1167                        }
1168                        errors.push(msg);
1169                    }
1170                }
1171            }
1172            Err(e) => {
1173                let msg = format!("Failed to build long-term cancel messages: {e:?}");
1174                log::error!("{msg}");
1175                errors.push(msg);
1176            }
1177        }
1178    }
1179
1180    if !errors.is_empty() {
1181        anyhow::bail!("partitioned cancel failed: {}", errors.join("; "));
1182    }
1183
1184    Ok(())
1185}
1186
1187fn emit_partitioned_cancel_rejections(
1188    orders: &[DydxCancelOrderRequest],
1189    emitter: &ExecutionEventEmitter,
1190    clock: &AtomicTime,
1191    reason: &str,
1192) {
1193    for order in orders {
1194        emitter.emit_order_cancel_rejected_event(
1195            order.strategy_id,
1196            order.instrument_id,
1197            order.client_order_id,
1198            order.venue_order_id,
1199            reason,
1200            clock.get_time_ns(),
1201        );
1202    }
1203}
1204
1205fn is_definitive_broadcast_rejection(err: &anyhow::Error) -> bool {
1206    err.downcast_ref::<DydxError>()
1207        .is_some_and(DydxError::is_definitive_broadcast_rejection)
1208}
1209
1210#[async_trait(?Send)]
1211impl ExecutionClient for DydxExecutionClient {
1212    fn is_connected(&self) -> bool {
1213        self.core.is_connected()
1214    }
1215
1216    fn client_id(&self) -> ClientId {
1217        self.core.client_id
1218    }
1219
1220    fn account_id(&self) -> AccountId {
1221        self.core.account_id
1222    }
1223
1224    fn venue(&self) -> Venue {
1225        *DYDX_VENUE
1226    }
1227
1228    fn oms_type(&self) -> OmsType {
1229        self.core.oms_type
1230    }
1231
1232    fn get_account(&self) -> Option<AccountAny> {
1233        self.core.cache().account_owned(&self.core.account_id)
1234    }
1235
1236    fn generate_account_state(
1237        &self,
1238        balances: Vec<AccountBalance>,
1239        margins: Vec<MarginBalance>,
1240        reported: bool,
1241        ts_event: UnixNanos,
1242        info: Option<Params>,
1243    ) -> anyhow::Result<()> {
1244        self.emitter
1245            .emit_account_state(balances, margins, reported, ts_event, info);
1246        Ok(())
1247    }
1248
1249    fn start(&mut self) -> anyhow::Result<()> {
1250        if self.core.is_started() {
1251            log::warn!("dYdX execution client already started");
1252            return Ok(());
1253        }
1254
1255        let sender = get_exec_event_sender();
1256        self.emitter.set_sender(sender);
1257        log::info!("Starting dYdX execution client");
1258        self.core.set_started();
1259        Ok(())
1260    }
1261
1262    fn stop(&mut self) -> anyhow::Result<()> {
1263        if self.core.is_stopped() {
1264            log::warn!("dYdX execution client not started");
1265            return Ok(());
1266        }
1267
1268        log::info!("Stopping dYdX execution client");
1269        self.session_tasks.begin_shutdown();
1270        self.begin_pending_shutdown();
1271        self.ws_client.begin_shutdown();
1272        self.core.set_stopped();
1273        self.core.set_disconnected();
1274        Ok(())
1275    }
1276
1277    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
1278        if !self.is_connected() {
1279            let reason = "Cannot submit order: execution client not connected";
1280            log::error!("{reason}");
1281            anyhow::bail!(reason);
1282        }
1283
1284        let current_block = self.block_time_monitor.current_block_height();
1285        let order = self.core.cache().try_order_owned(&cmd.client_order_id)?;
1286
1287        let client_order_id = order.client_order_id();
1288        let instrument_id = order.instrument_id();
1289        let strategy_id = order.strategy_id();
1290
1291        if current_block == 0 {
1292            let reason = "Block height not initialized";
1293            log::warn!("Cannot submit order {client_order_id}: {reason}");
1294            self.emitter.emit_order_denied(&order, reason);
1295            return Ok(());
1296        }
1297
1298        if order.is_closed() {
1299            log::warn!("Cannot submit closed order {client_order_id}");
1300            return Ok(());
1301        }
1302
1303        if order.is_quote_quantity() {
1304            let reason = OrderDeniedReason::unsupported_quote_quantity().to_string();
1305            log::warn!("{reason}");
1306            self.emitter.emit_order_denied(&order, &reason);
1307            return Ok(());
1308        }
1309
1310        // Deny unsupported time-in-force values up front so the strategy gets
1311        // an immediate `OrderDenied` rather than a venue-side `code=48` (FOK
1312        // deprecated) or a silent translation to GTC (DAY).
1313        let unsupported_tif_reason = match order.time_in_force() {
1314            TimeInForce::Fok => Some(
1315                "Fill-or-kill (FOK) orders are deprecated by dYdX v4 (chain rejects with code=48)",
1316            ),
1317            TimeInForce::Day => Some("DAY time-in-force is not supported by dYdX v4"),
1318            _ => None,
1319        };
1320
1321        if let Some(reason) = unsupported_tif_reason {
1322            log::warn!("{reason}");
1323            self.emitter.emit_order_denied(&order, reason);
1324            return Ok(());
1325        }
1326
1327        match order.order_type() {
1328            OrderType::Market
1329            | OrderType::Limit
1330            | OrderType::StopMarket
1331            | OrderType::StopLimit
1332            | OrderType::MarketIfTouched
1333            | OrderType::LimitIfTouched => {}
1334            OrderType::TrailingStopMarket | OrderType::TrailingStopLimit => {
1335                let reason = "Trailing stop orders not supported by dYdX v4 protocol";
1336                log::warn!("{reason}");
1337                self.emitter.emit_order_denied(&order, reason);
1338                return Ok(());
1339            }
1340            order_type => {
1341                let reason = format!("Order type {order_type:?} not supported by dYdX");
1342                log::warn!("{reason}");
1343                self.emitter.emit_order_denied(&order, &reason);
1344                return Ok(());
1345            }
1346        }
1347
1348        let (tx_manager, broadcaster, order_builder) = match self.get_execution_components() {
1349            Ok(components) => components,
1350            Err(e) => {
1351                log::error!("Failed to get execution components: {e}");
1352                self.emitter.emit_order_denied(&order, &e.to_string());
1353                return Ok(());
1354            }
1355        };
1356
1357        let block_height = self.block_time_monitor.current_block_height() as u32;
1358
1359        let encoded = match self.encoder.encode(client_order_id) {
1360            Ok(enc) => enc,
1361            Err(e) => {
1362                log::error!("Failed to generate client order ID: {e}");
1363                self.emitter.emit_order_denied(&order, &e.to_string());
1364                return Ok(());
1365            }
1366        };
1367
1368        self.emitter.emit_order_submitted(&order);
1369        let client_id_u32 = encoded.client_id;
1370        let client_metadata = encoded.client_metadata;
1371
1372        log::debug!(
1373            "[SUBMIT_ORDER] Nautilus '{}' -> dYdX u32={} meta={:#x} | instrument={} side={:?} qty={} type={:?}",
1374            client_order_id,
1375            client_id_u32,
1376            client_metadata,
1377            instrument_id,
1378            order.order_side(),
1379            order.quantity(),
1380            order.order_type()
1381        );
1382
1383        let expire_time = order.expire_time().map(nanos_to_secs_i64);
1384
1385        let order_flags = match order.order_type() {
1386            OrderType::StopMarket
1387            | OrderType::StopLimit
1388            | OrderType::MarketIfTouched
1389            | OrderType::LimitIfTouched => types::ORDER_FLAG_CONDITIONAL,
1390            OrderType::Market => types::ORDER_FLAG_SHORT_TERM,
1391            OrderType::Limit => {
1392                let lifetime = types::OrderLifetime::from_time_in_force(
1393                    order.time_in_force(),
1394                    expire_time,
1395                    false,
1396                    order_builder.max_short_term_secs(),
1397                );
1398                lifetime.order_flags()
1399            }
1400            _ => types::ORDER_FLAG_LONG_TERM,
1401        };
1402
1403        let ts_submitted = self.clock.get_time_ns();
1404        let trader_id = order.trader_id();
1405        self.register_order_context(
1406            client_id_u32,
1407            OrderContext {
1408                client_order_id,
1409                trader_id,
1410                strategy_id,
1411                instrument_id,
1412                submitted_at: ts_submitted,
1413                order_flags,
1414            },
1415        );
1416
1417        // Register dispatch identity so the WS handler emits proper order
1418        // events (OrderAccepted, OrderFilled, OrderCanceled) instead of reports
1419        self.dispatch_state.order_identities.insert(
1420            client_order_id,
1421            OrderIdentity {
1422                instrument_id,
1423                strategy_id,
1424                order_side: order.order_side(),
1425                order_type: order.order_type(),
1426            },
1427        );
1428
1429        self.spawn_order_task(
1430            "submit_order",
1431            strategy_id,
1432            instrument_id,
1433            client_order_id,
1434            async move {
1435                let (msg, order_type_str) = match order.order_type() {
1436                    OrderType::Market => {
1437                        let msg = order_builder.build_market_order_with_reduce_only(
1438                            instrument_id,
1439                            client_id_u32,
1440                            client_metadata,
1441                            order.order_side(),
1442                            order.quantity(),
1443                            order.is_reduce_only(),
1444                            block_height,
1445                        )?;
1446                        (msg, "market")
1447                    }
1448                    OrderType::Limit => {
1449                        let msg = order_builder.build_limit_order(
1450                            instrument_id,
1451                            client_id_u32,
1452                            client_metadata,
1453                            order.order_side(),
1454                            order
1455                                .price()
1456                                .ok_or_else(|| anyhow::anyhow!("Limit order missing price"))?,
1457                            order.quantity(),
1458                            order.time_in_force(),
1459                            order.is_post_only(),
1460                            order.is_reduce_only(),
1461                            block_height,
1462                            expire_time, // Uses default_short_term_expiry if configured
1463                        )?;
1464                        (msg, "limit")
1465                    }
1466                    // Conditional orders use their own expiration logic (not affected by default_short_term_expiry)
1467                    // They are always stored on-chain with long-term semantics
1468                    OrderType::StopMarket => {
1469                        let trigger_price = order.trigger_price().ok_or_else(|| {
1470                            anyhow::anyhow!("Stop market order missing trigger_price")
1471                        })?;
1472                        let cond_expire = order.expire_time().map(nanos_to_secs_i64);
1473                        let msg = order_builder.build_stop_market_order(
1474                            instrument_id,
1475                            client_id_u32,
1476                            client_metadata,
1477                            order.order_side(),
1478                            trigger_price,
1479                            order.quantity(),
1480                            order.is_reduce_only(),
1481                            cond_expire,
1482                        )?;
1483                        (msg, "stop_market")
1484                    }
1485                    OrderType::StopLimit => {
1486                        let trigger_price = order.trigger_price().ok_or_else(|| {
1487                            anyhow::anyhow!("Stop limit order missing trigger_price")
1488                        })?;
1489                        let limit_price = order.price().ok_or_else(|| {
1490                            anyhow::anyhow!("Stop limit order missing limit price")
1491                        })?;
1492                        let cond_expire = order.expire_time().map(nanos_to_secs_i64);
1493                        let msg = order_builder.build_stop_limit_order(
1494                            instrument_id,
1495                            client_id_u32,
1496                            client_metadata,
1497                            order.order_side(),
1498                            trigger_price,
1499                            limit_price,
1500                            order.quantity(),
1501                            order.time_in_force(),
1502                            order.is_post_only(),
1503                            order.is_reduce_only(),
1504                            cond_expire,
1505                        )?;
1506                        (msg, "stop_limit")
1507                    }
1508                    // dYdX TakeProfitMarket maps to Nautilus MarketIfTouched
1509                    OrderType::MarketIfTouched => {
1510                        let trigger_price = order.trigger_price().ok_or_else(|| {
1511                            anyhow::anyhow!("Take profit market order missing trigger_price")
1512                        })?;
1513                        let cond_expire = order.expire_time().map(nanos_to_secs_i64);
1514                        let msg = order_builder.build_take_profit_market_order(
1515                            instrument_id,
1516                            client_id_u32,
1517                            client_metadata,
1518                            order.order_side(),
1519                            trigger_price,
1520                            order.quantity(),
1521                            order.is_reduce_only(),
1522                            cond_expire,
1523                        )?;
1524                        (msg, "take_profit_market")
1525                    }
1526                    // dYdX TakeProfitLimit maps to Nautilus LimitIfTouched
1527                    OrderType::LimitIfTouched => {
1528                        let trigger_price = order.trigger_price().ok_or_else(|| {
1529                            anyhow::anyhow!("Take profit limit order missing trigger_price")
1530                        })?;
1531                        let limit_price = order.price().ok_or_else(|| {
1532                            anyhow::anyhow!("Take profit limit order missing limit price")
1533                        })?;
1534                        let cond_expire = order.expire_time().map(nanos_to_secs_i64);
1535                        let msg = order_builder.build_take_profit_limit_order(
1536                            instrument_id,
1537                            client_id_u32,
1538                            client_metadata,
1539                            order.order_side(),
1540                            trigger_price,
1541                            limit_price,
1542                            order.quantity(),
1543                            order.time_in_force(),
1544                            order.is_post_only(),
1545                            order.is_reduce_only(),
1546                            cond_expire,
1547                        )?;
1548                        (msg, "take_profit_limit")
1549                    }
1550                    _ => unreachable!("Order type already validated"),
1551                };
1552
1553                // Broadcast: short-term orders use cached sequence (no increment),
1554                // stateful orders use broadcast_with_retry (proper sequence management)
1555                let operation = format!("Submit {order_type_str} order {client_order_id}");
1556
1557                if order_flags == types::ORDER_FLAG_SHORT_TERM {
1558                    broadcaster
1559                        .broadcast_short_term(&tx_manager, vec![msg], &operation)
1560                        .await?;
1561                } else {
1562                    broadcaster
1563                        .broadcast_with_retry(&tx_manager, vec![msg], &operation)
1564                        .await?;
1565                }
1566                log::debug!("Successfully submitted {order_type_str} order: {client_order_id}");
1567
1568                Ok(())
1569            },
1570        );
1571
1572        Ok(())
1573    }
1574
1575    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
1576        let orders = self.core.get_orders_for_list(&cmd.order_list)?;
1577        let order_count = orders.len();
1578
1579        if !self.is_connected() {
1580            let reason = "Cannot submit order list: execution client not connected";
1581            log::error!("{reason}");
1582            anyhow::bail!(reason);
1583        }
1584
1585        // Pre-submission TIF gate: deny FOK and DAY before doing any further
1586        // work, mirroring `submit_order`'s gate. Done up front so the strategy
1587        // sees an immediate `OrderDenied` even when the rest of the pipeline
1588        // (block height, TX manager) is not yet ready.
1589        let unsupported_tif_reason = |order: &OrderAny| match order.time_in_force() {
1590            TimeInForce::Fok => Some(
1591                "Fill-or-kill (FOK) orders are deprecated by dYdX v4 (chain rejects with code=48)",
1592            ),
1593            TimeInForce::Day => Some("DAY time-in-force is not supported by dYdX v4"),
1594            _ => None,
1595        };
1596
1597        if orders
1598            .iter()
1599            .any(|order| unsupported_tif_reason(order).is_some())
1600        {
1601            // Deny the whole list: dYdX has no atomic batch semantics worth
1602            // preserving, and siblings must not be left unresolved.
1603            for order in &orders {
1604                let reason = unsupported_tif_reason(order)
1605                    .unwrap_or("Order list denied: sibling order has unsupported time in force");
1606                self.emitter.emit_order_denied(order, reason);
1607            }
1608            return Ok(());
1609        }
1610
1611        let current_block = self.block_time_monitor.current_block_height();
1612        if current_block == 0 {
1613            let reason = "Block height not initialized";
1614            log::warn!("Cannot submit order list: {reason}");
1615            for order in &orders {
1616                self.emitter.emit_order_denied(order, reason);
1617            }
1618            return Ok(());
1619        }
1620
1621        let (tx_manager, broadcaster, order_builder) = match self.get_execution_components() {
1622            Ok(components) => components,
1623            Err(e) => {
1624                log::error!("Failed to get execution components for batch: {e}");
1625                for order in &orders {
1626                    self.emitter.emit_order_denied(order, &e.to_string());
1627                }
1628                return Ok(());
1629            }
1630        };
1631
1632        let mut order_params: Vec<LimitOrderParams> = Vec::with_capacity(order_count);
1633        let mut order_info: Vec<(ClientOrderId, InstrumentId, StrategyId)> =
1634            Vec::with_capacity(order_count);
1635
1636        for order in &orders {
1637            if order.order_type() != OrderType::Limit {
1638                log::debug!(
1639                    "Order {} has type {:?}, falling back to individual submission",
1640                    order.client_order_id(),
1641                    order.order_type()
1642                );
1643                let submit_cmd = SubmitOrder::new(
1644                    cmd.trader_id,
1645                    cmd.client_id,
1646                    cmd.strategy_id,
1647                    order.instrument_id(),
1648                    order.client_order_id(),
1649                    order.init_event().clone(),
1650                    cmd.exec_algorithm_id,
1651                    cmd.position_id,
1652                    cmd.params.clone(),
1653                    UUID4::new(),
1654                    cmd.ts_init,
1655                    cmd.correlation_id,
1656                );
1657
1658                if let Err(e) = self.submit_order(submit_cmd) {
1659                    log::error!(
1660                        "Failed to submit order {} from order list: {e}",
1661                        order.client_order_id()
1662                    );
1663                }
1664                continue;
1665            }
1666
1667            if order.is_quote_quantity() {
1668                self.emitter.emit_order_denied(
1669                    order,
1670                    &OrderDeniedReason::unsupported_quote_quantity().to_string(),
1671                );
1672                continue;
1673            }
1674
1675            let Some(price) = order.price() else {
1676                self.emitter
1677                    .emit_order_denied(order, "Limit order missing price");
1678                continue;
1679            };
1680
1681            let encoded = match self.encoder.encode(order.client_order_id()) {
1682                Ok(enc) => enc,
1683                Err(e) => {
1684                    log::error!("Failed to generate client order ID: {e}");
1685                    self.emitter.emit_order_denied(order, &e.to_string());
1686                    continue;
1687                }
1688            };
1689            let client_id_u32 = encoded.client_id;
1690            let client_metadata = encoded.client_metadata;
1691
1692            self.emitter.emit_order_submitted(order);
1693
1694            let expire_time_secs = order.expire_time().map(nanos_to_secs_i64);
1695            let lifetime = types::OrderLifetime::from_time_in_force(
1696                order.time_in_force(),
1697                expire_time_secs,
1698                false,
1699                order_builder.max_short_term_secs(),
1700            );
1701
1702            let ts_submitted = self.clock.get_time_ns();
1703            self.register_order_context(
1704                client_id_u32,
1705                OrderContext {
1706                    client_order_id: order.client_order_id(),
1707                    trader_id: order.trader_id(),
1708                    strategy_id: order.strategy_id(),
1709                    instrument_id: order.instrument_id(),
1710                    submitted_at: ts_submitted,
1711                    order_flags: lifetime.order_flags(),
1712                },
1713            );
1714
1715            self.dispatch_state.order_identities.insert(
1716                order.client_order_id(),
1717                OrderIdentity {
1718                    instrument_id: order.instrument_id(),
1719                    strategy_id: order.strategy_id(),
1720                    order_side: order.order_side(),
1721                    order_type: order.order_type(),
1722                },
1723            );
1724
1725            order_params.push(LimitOrderParams {
1726                instrument_id: order.instrument_id(),
1727                client_order_id: client_id_u32,
1728                client_metadata,
1729                side: order.order_side(),
1730                price,
1731                quantity: order.quantity(),
1732                time_in_force: order.time_in_force(),
1733                post_only: order.is_post_only(),
1734                reduce_only: order.is_reduce_only(),
1735                expire_time_ns: order.expire_time(),
1736            });
1737            order_info.push((
1738                order.client_order_id(),
1739                order.instrument_id(),
1740                order.strategy_id(),
1741            ));
1742        }
1743
1744        if order_params.is_empty() {
1745            return Ok(());
1746        }
1747
1748        // dYdX requires each short-term order in its own transaction
1749        let has_short_term = order_params
1750            .iter()
1751            .any(|params| order_builder.is_short_term_order(params));
1752
1753        let block_height = current_block as u32;
1754        let emitter = self.emitter.clone();
1755        let clock = self.clock;
1756
1757        if has_short_term {
1758            log::debug!(
1759                "Submitting {} short-term limit orders concurrently (sequence not consumed)",
1760                order_params.len()
1761            );
1762
1763            self.spawn_labeled("batch_submit_short_term", async move {
1764                // Build and broadcast all orders concurrently -- no sequence coordination needed.
1765                // Short-term orders use cached sequence (not incremented) via broadcast_short_term.
1766                let mut handles = tokio::task::JoinSet::new();
1767
1768                for (params, (client_order_id, instrument_id, strategy_id)) in
1769                    order_params.into_iter().zip(order_info)
1770                {
1771                    let tx_manager = tx_manager.clone();
1772                    let broadcaster = broadcaster.clone();
1773                    let order_builder = order_builder.clone();
1774                    let emitter = emitter.clone();
1775
1776                    handles.spawn(async move {
1777                        let msg = match order_builder
1778                            .build_limit_order_from_params(&params, block_height)
1779                        {
1780                            Ok(m) => m,
1781                            Err(e) => {
1782                                // Local failure after OrderSubmitted: leave the
1783                                // order for in-flight resolution.
1784                                log::error!(
1785                                    "Failed to build order message for {client_order_id}: {e:?}"
1786                                );
1787                                return;
1788                            }
1789                        };
1790
1791                        // Broadcast with cached sequence (short-term orders don't consume sequences)
1792                        let operation = format!("Submit short-term order {client_order_id}");
1793
1794                        if let Err(e) = broadcaster
1795                            .broadcast_short_term(&tx_manager, vec![msg], &operation)
1796                            .await
1797                        {
1798                            if e.is_definitive_broadcast_rejection() {
1799                                let error_msg = format!("Order submission failed: {e:?}");
1800                                log::warn!("{error_msg}");
1801                                let ts_event = clock.get_time_ns();
1802                                emitter.emit_order_rejected_event(
1803                                    strategy_id,
1804                                    instrument_id,
1805                                    client_order_id,
1806                                    &error_msg,
1807                                    ts_event,
1808                                    false,
1809                                );
1810                            } else {
1811                                log::warn!(
1812                                    "Ambiguous dYdX submit failure for {client_order_id}, awaiting reconciliation: {e:?}"
1813                                );
1814                            }
1815                        }
1816                    });
1817                }
1818
1819                while let Some(result) = handles.join_next().await {
1820                    if let Err(e) = result
1821                        && !e.is_cancelled()
1822                    {
1823                        log::warn!("dYdX short-term order task failed: {e}");
1824                    }
1825                }
1826            });
1827        } else {
1828            log::debug!(
1829                "Batch submitting {} long-term limit orders in single transaction",
1830                order_params.len()
1831            );
1832
1833            self.spawn_labeled("batch_submit_long_term", async move {
1834                let msgs: Result<Vec<_>, _> = order_params
1835                    .iter()
1836                    .map(|params| order_builder.build_limit_order_from_params(params, block_height))
1837                    .collect();
1838
1839                let msgs = match msgs {
1840                    Ok(m) => m,
1841                    Err(e) => {
1842                        // Local failure after OrderSubmitted: leave the orders
1843                        // for in-flight resolution.
1844                        log::error!("Failed to build batch order messages: {e:?}");
1845                        return;
1846                    }
1847                };
1848
1849                let operation = format!("Submit batch of {} limit orders", msgs.len());
1850
1851                if let Err(e) = broadcaster
1852                    .broadcast_with_retry(&tx_manager, msgs, &operation)
1853                    .await
1854                {
1855                    if e.is_definitive_broadcast_rejection() {
1856                        // A CheckTx rejection refuses the whole transaction
1857                        // atomically: a venue result for every order in it.
1858                        let error_msg = format!("Batch order submission failed: {e:?}");
1859                        log::warn!("{error_msg}");
1860                        let ts_event = clock.get_time_ns();
1861
1862                        for (client_order_id, instrument_id, strategy_id) in order_info {
1863                            emitter.emit_order_rejected_event(
1864                                strategy_id,
1865                                instrument_id,
1866                                client_order_id,
1867                                &error_msg,
1868                                ts_event,
1869                                false,
1870                            );
1871                        }
1872                    } else {
1873                        log::warn!(
1874                            "Ambiguous dYdX batch submit failure, awaiting reconciliation: {e:?}"
1875                        );
1876                    }
1877                }
1878            });
1879        }
1880
1881        Ok(())
1882    }
1883
1884    /// dYdX does not support native order modification.
1885    ///
1886    /// Strategies should handle `OrderModifyRejected` by canceling and resubmitting.
1887    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
1888        let reason = "dYdX does not support order modification. Use cancel and resubmit instead.";
1889        log::warn!("{reason}");
1890
1891        self.send_modify_rejected(
1892            cmd.strategy_id,
1893            cmd.instrument_id,
1894            cmd.client_order_id,
1895            cmd.venue_order_id,
1896            reason,
1897        );
1898        Ok(())
1899    }
1900
1901    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
1902        if !self.is_connected() {
1903            anyhow::bail!("Cannot cancel order: not connected");
1904        }
1905
1906        let client_order_id = cmd.client_order_id;
1907        let instrument_id = cmd.instrument_id;
1908        let strategy_id = cmd.strategy_id;
1909        let venue_order_id = cmd.venue_order_id;
1910
1911        let (order_time_in_force, order_expire_time) = {
1912            let cache = self.core.cache();
1913
1914            let order = match cache.order(&client_order_id) {
1915                Some(order) => order,
1916                None => {
1917                    log::error!("Cannot cancel order {client_order_id}: not found in cache");
1918                    return Ok(()); // Not an error - order may have been filled/canceled already
1919                }
1920            };
1921
1922            if order.is_closed() {
1923                log::warn!(
1924                    "CancelOrder command for {} when order already {} (will not send to exchange)",
1925                    client_order_id,
1926                    order.status()
1927                );
1928                return Ok(());
1929            }
1930
1931            if cache.instrument(&instrument_id).is_none() {
1932                log::error!(
1933                    "Cannot cancel order {client_order_id}: instrument {instrument_id} not found in cache"
1934                );
1935                return Ok(()); // Not an error - missing instrument is a cache issue
1936            }
1937
1938            (
1939                order.time_in_force(),
1940                order.expire_time().map(nanos_to_secs_i64),
1941            )
1942        }; // Cache borrow released here
1943
1944        log::debug!("Cancelling order {client_order_id} for instrument {instrument_id}");
1945
1946        let (tx_manager, broadcaster, order_builder) = match self.get_execution_components() {
1947            Ok(components) => components,
1948            Err(e) => {
1949                log::error!("Failed to get execution components for cancel: {e}");
1950                return Ok(());
1951            }
1952        };
1953
1954        let block_height = self.block_time_monitor.current_block_height() as u32;
1955
1956        let encoded = match self.encoder.get(&client_order_id) {
1957            Some(enc) => enc,
1958            None => {
1959                log::error!("Client order ID {client_order_id} not found in cache");
1960                anyhow::bail!("Client order ID not found in cache")
1961            }
1962        };
1963        let client_id_u32 = encoded.client_id;
1964
1965        log::debug!(
1966            "[CANCEL_ORDER] Nautilus '{client_order_id}' -> dYdX u32={client_id_u32} | instrument={instrument_id}"
1967        );
1968
1969        // Stored flags remain authoritative after the order expires
1970        let order_flags = self.get_order_context(client_id_u32).map_or_else(
1971            || {
1972                log::debug!(
1973                    "Order context not found for {client_order_id}, deriving flags from order"
1974                );
1975                types::OrderLifetime::from_time_in_force(
1976                    order_time_in_force, // Using extracted value
1977                    order_expire_time,   // Using extracted value
1978                    false,
1979                    order_builder.max_short_term_secs(),
1980                )
1981                .order_flags()
1982            },
1983            |ctx| ctx.order_flags,
1984        );
1985
1986        let clock = self.clock;
1987        let emitter = self.emitter.clone();
1988
1989        self.spawn_task("cancel_order", async move {
1990            let cancel_msg = match order_builder.build_cancel_order_with_flags(
1991                instrument_id,
1992                client_id_u32,
1993                order_flags,
1994                block_height,
1995            ) {
1996                Ok(msg) => msg,
1997                Err(e) => {
1998                    // Local validation failure: leave the cancel outcome for
1999                    // in-flight resolution.
2000                    log::warn!(
2001                        "Cancel command failed local validation for {client_order_id}: {e:?}"
2002                    );
2003                    return Ok(());
2004                }
2005            };
2006
2007            // Broadcast cancel: short-term uses cached sequence, stateful uses retry
2008            let cancel_op = format!("Cancel order {client_order_id}");
2009            let result = if order_flags == types::ORDER_FLAG_SHORT_TERM {
2010                broadcaster
2011                    .broadcast_short_term(&tx_manager, vec![cancel_msg], &cancel_op)
2012                    .await
2013            } else {
2014                broadcaster
2015                    .broadcast_with_retry(&tx_manager, vec![cancel_msg], &cancel_op)
2016                    .await
2017            };
2018
2019            match result {
2020                Ok(_) => {
2021                    log::debug!("Successfully cancelled order: {client_order_id}");
2022                }
2023                Err(e) if e.is_definitive_broadcast_rejection() => {
2024                    log::warn!("Failed to cancel order {client_order_id}: {e:?}");
2025
2026                    let ts_event = clock.get_time_ns();
2027                    emitter.emit_order_cancel_rejected_event(
2028                        strategy_id,
2029                        instrument_id,
2030                        client_order_id,
2031                        venue_order_id,
2032                        &format!("Cancel order failed: {e:?}"),
2033                        ts_event,
2034                    );
2035                }
2036                Err(e) => {
2037                    log::warn!(
2038                        "Ambiguous dYdX cancel failure for {client_order_id}, awaiting reconciliation: {e:?}"
2039                    );
2040                }
2041            }
2042
2043            Ok(())
2044        });
2045
2046        Ok(())
2047    }
2048
2049    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
2050        if !self.is_connected() {
2051            anyhow::bail!("Cannot cancel orders: not connected");
2052        }
2053
2054        let instrument_id = cmd.instrument_id;
2055        let order_side_filter = cmd.order_side;
2056
2057        let order_data: Vec<CancelAllOrderData> = {
2058            let cache = self.core.cache();
2059            let side_filter = order_side_filter;
2060            cache
2061                .orders_open(None, Some(&instrument_id), None, None, side_filter)
2062                .into_iter()
2063                .map(|order| {
2064                    (
2065                        order.strategy_id(),
2066                        order.client_order_id(),
2067                        order.venue_order_id(),
2068                        order.time_in_force(),
2069                        order.expire_time(),
2070                    )
2071                })
2072                .collect()
2073        }; // Cache borrow released here
2074
2075        let short_term_count = order_data
2076            .iter()
2077            .filter(|(_, _, _, tif, _)| matches!(tif, TimeInForce::Ioc | TimeInForce::Fok))
2078            .count();
2079        let long_term_count = order_data.len() - short_term_count;
2080
2081        log::debug!(
2082            "Cancel all orders: total={}, short_term={}, long_term={}, instrument_id={instrument_id}, order_side={order_side_filter:?}",
2083            order_data.len(),
2084            short_term_count,
2085            long_term_count
2086        );
2087
2088        let (tx_manager, broadcaster, order_builder) = match self.get_execution_components() {
2089            Ok(components) => components,
2090            Err(e) => {
2091                log::error!("Failed to get execution components for cancel_all: {e}");
2092                return Ok(());
2093            }
2094        };
2095
2096        let block_height = self.block_time_monitor.current_block_height() as u32;
2097
2098        let mut orders_to_cancel = Vec::new();
2099
2100        for (strategy_id, client_order_id, venue_order_id, _time_in_force, _expire_time) in
2101            &order_data
2102        {
2103            let Some(encoded) = self.encoder.get(client_order_id) else {
2104                log::warn!("Cannot cancel order {client_order_id}: not found in encoder");
2105                continue;
2106            };
2107            let client_id_u32 = encoded.client_id;
2108
2109            let Some(ctx) = self.get_order_context(client_id_u32) else {
2110                log::debug!(
2111                    "Skipping cancel for {client_order_id}: order context already cleaned up (terminal)"
2112                );
2113                continue;
2114            };
2115            orders_to_cancel.push(DydxCancelOrderRequest {
2116                instrument_id,
2117                client_id: client_id_u32,
2118                order_flags: ctx.order_flags,
2119                strategy_id: *strategy_id,
2120                client_order_id: *client_order_id,
2121                venue_order_id: *venue_order_id,
2122            });
2123        }
2124
2125        if orders_to_cancel.is_empty() {
2126            return Ok(());
2127        }
2128
2129        log::debug!(
2130            "Cancel all: {} orders (short_term={}, long_term={}), instrument_id={instrument_id}, order_side={order_side_filter:?}",
2131            orders_to_cancel.len(),
2132            orders_to_cancel
2133                .iter()
2134                .filter(|order| order.order_flags == types::ORDER_FLAG_SHORT_TERM)
2135                .count(),
2136            orders_to_cancel
2137                .iter()
2138                .filter(|order| order.order_flags != types::ORDER_FLAG_SHORT_TERM)
2139                .count(),
2140        );
2141
2142        let clock = self.clock;
2143        let emitter = self.emitter.clone();
2144
2145        self.spawn_task("cancel_all_orders", async move {
2146            broadcast_partitioned_cancels(
2147                orders_to_cancel,
2148                block_height,
2149                tx_manager,
2150                broadcaster,
2151                order_builder,
2152                emitter,
2153                clock,
2154            )
2155            .await
2156        });
2157
2158        Ok(())
2159    }
2160
2161    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
2162        if cmd.cancels.is_empty() {
2163            return Ok(());
2164        }
2165
2166        if !self.is_connected() {
2167            anyhow::bail!("Cannot cancel orders: not connected");
2168        }
2169
2170        let (tx_manager, broadcaster, order_builder) = match self.get_execution_components() {
2171            Ok(components) => components,
2172            Err(e) => {
2173                log::error!("Failed to get execution components for batch cancel: {e}");
2174                return Ok(());
2175            }
2176        };
2177
2178        let mut orders_to_cancel = Vec::with_capacity(cmd.cancels.len());
2179        for cancel in &cmd.cancels {
2180            let client_order_id = cancel.client_order_id;
2181            let encoded = match self.encoder.get(&client_order_id) {
2182                Some(enc) => enc,
2183                None => {
2184                    log::warn!(
2185                        "No u32 mapping found for client_order_id={client_order_id}, skipping cancel"
2186                    );
2187                    continue;
2188                }
2189            };
2190            let client_id_u32 = encoded.client_id;
2191
2192            let Some(ctx) = self.get_order_context(client_id_u32) else {
2193                log::debug!(
2194                    "Skipping cancel for {client_order_id}: order context already cleaned up (terminal)"
2195                );
2196                continue;
2197            };
2198
2199            orders_to_cancel.push(DydxCancelOrderRequest {
2200                instrument_id: cancel.instrument_id,
2201                client_id: client_id_u32,
2202                order_flags: ctx.order_flags,
2203                strategy_id: cancel.strategy_id,
2204                client_order_id,
2205                venue_order_id: cancel.venue_order_id,
2206            });
2207        }
2208
2209        if orders_to_cancel.is_empty() {
2210            log::warn!("No valid orders to cancel in batch");
2211            return Ok(());
2212        }
2213
2214        let block_height = self.block_time_monitor.current_block_height() as u32;
2215        let clock = self.clock;
2216        let emitter = self.emitter.clone();
2217
2218        log::debug!(
2219            "Batch cancelling {} orders via partitioned strategy",
2220            orders_to_cancel.len(),
2221        );
2222
2223        self.spawn_task("batch_cancel_orders", async move {
2224            broadcast_partitioned_cancels(
2225                orders_to_cancel,
2226                block_height,
2227                tx_manager,
2228                broadcaster,
2229                order_builder,
2230                emitter,
2231                clock,
2232            )
2233            .await
2234        });
2235
2236        Ok(())
2237    }
2238
2239    fn query_account(&self, _cmd: QueryAccount) -> anyhow::Result<()> {
2240        let http_client = self.http_client.clone();
2241        let wallet_address = self.wallet_address.clone();
2242        let subaccount_number = self.subaccount_number;
2243        let account_id = self.core.account_id;
2244        let emitter = self.emitter.clone();
2245
2246        self.spawn_task("query_account", async move {
2247            let account_state = http_client
2248                .request_account_state(&wallet_address, subaccount_number, account_id)
2249                .await
2250                .context("failed to query account state")?;
2251
2252            emitter.emit_account_state(
2253                account_state.balances.clone(),
2254                account_state.margins.clone(),
2255                account_state.is_reported,
2256                account_state.ts_event,
2257                account_state.info,
2258            );
2259            Ok(())
2260        });
2261
2262        Ok(())
2263    }
2264
2265    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
2266        log::debug!("Querying order: client_order_id={}", cmd.client_order_id);
2267
2268        let http_client = self.http_client.clone();
2269        let wallet_address = self.wallet_address.clone();
2270        let subaccount_number = self.subaccount_number;
2271        let account_id = self.core.account_id;
2272        let emitter = self.emitter.clone();
2273        let client_order_id = cmd.client_order_id;
2274        let venue_order_id = cmd.venue_order_id;
2275        let instrument_id = cmd.instrument_id;
2276
2277        self.spawn_task("query_order", async move {
2278            let reports = http_client
2279                .request_order_status_reports(
2280                    &wallet_address,
2281                    subaccount_number,
2282                    account_id,
2283                    Some(instrument_id),
2284                )
2285                .await
2286                .context("failed to query order status")?;
2287
2288            let report = reports.into_iter().find(|r| {
2289                if venue_order_id.is_some_and(|vid| r.venue_order_id == vid) {
2290                    return true;
2291                }
2292                r.client_order_id.is_some_and(|cid| cid == client_order_id)
2293            });
2294
2295            if let Some(report) = report {
2296                emitter.send_order_status_report(report);
2297            } else {
2298                log::warn!(
2299                    "No order found for client_order_id={client_order_id}, venue_order_id={venue_order_id:?}"
2300                );
2301            }
2302
2303            Ok(())
2304        });
2305
2306        Ok(())
2307    }
2308
2309    async fn connect(&mut self) -> anyhow::Result<()> {
2310        if self.core.is_connected() && self.session_tasks.is_open() && self.pending_tasks.is_open()
2311        {
2312            log::warn!("dYdX execution client already connected");
2313            return Ok(());
2314        }
2315
2316        log::info!("Connecting to dYdX");
2317
2318        self.prepare_task_groups().await?;
2319        let ws_client = self.ws_client.clone();
2320        let setup_guard =
2321            TaskGroupGuard::new(&[&self.session_tasks, &self.pending_tasks], move || {
2322                ws_client.begin_shutdown();
2323            });
2324
2325        log::debug!("Loading instruments from HTTP API");
2326        self.http_client.fetch_and_cache_instruments().await?;
2327        log::debug!(
2328            "Loaded {} instruments from HTTP into shared cache",
2329            self.http_client.cached_instruments_count()
2330        );
2331        self.mark_instruments_initialized();
2332
2333        // Initialize gRPC client (deferred from constructor to avoid blocking)
2334        let grpc_urls = self.config.get_grpc_urls();
2335        let mut grpc_client = DydxGrpcClient::new_with_fallback(&grpc_urls)
2336            .await
2337            .context("failed to construct dYdX gRPC client")?;
2338        log::debug!("gRPC client initialized");
2339
2340        // Fetch the initial height synchronously so orders can be submitted immediately
2341        let initial_height = grpc_client
2342            .latest_block_height()
2343            .await
2344            .context("failed to fetch initial block height")?;
2345        // Use current time as approximation; actual timestamps will come from WebSocket updates
2346        self.block_time_monitor
2347            .record_block(initial_height.0 as u64, jiff::Timestamp::now());
2348        log::debug!("Initial block height: {}", initial_height.0);
2349
2350        *self.grpc_client.write().await = Some(grpc_client.clone());
2351
2352        let private_key =
2353            Self::resolve_private_key(&self.config).context("failed to resolve private key")?;
2354        let tx_manager = Arc::new(
2355            TransactionManager::new(
2356                grpc_client.clone(),
2357                &private_key,
2358                self.wallet_address.clone(),
2359                self.get_chain_id(),
2360            )
2361            .context("failed to create TransactionManager")?,
2362        );
2363
2364        tx_manager
2365            .resolve_authenticators()
2366            .await
2367            .context("failed to resolve authenticators")?;
2368
2369        // Proactively initialize sequence from chain so orders can be submitted
2370        // immediately after connect() without first-transaction latency penalty.
2371        tx_manager
2372            .initialize_sequence()
2373            .await
2374            .context("failed to initialize sequence")?;
2375
2376        self.tx_manager = Some(tx_manager);
2377        self.broadcaster = Some(Arc::new(TxBroadcaster::new(
2378            grpc_client,
2379            self.config.grpc_quota(),
2380        )));
2381        self.order_builder = Some(Arc::new(OrderMessageBuilder::new(
2382            self.http_client.clone(),
2383            self.wallet_address.clone(),
2384            self.subaccount_number,
2385            self.block_time_monitor.clone(),
2386        )));
2387        log::debug!(
2388            "OrderMessageBuilder initialized (block_time_monitor ready: {}, max_short_term: {:.1}s)",
2389            self.block_time_monitor.is_ready(),
2390            SHORT_TERM_ORDER_MAXIMUM_LIFETIME as f64
2391                * self.block_time_monitor.seconds_per_block_or_default()
2392        );
2393
2394        let session_result = async {
2395            self.ws_client.connect().await?;
2396            log::debug!("WebSocket connected");
2397
2398            self.ws_client.subscribe_block_height().await?;
2399            log::debug!("Subscribed to block height updates");
2400
2401            self.ws_client.subscribe_markets().await?;
2402            log::debug!("Subscribed to markets");
2403
2404            log::debug!(
2405                "Using wallet address for queries: {} (subaccount {})",
2406                self.wallet_address,
2407                self.subaccount_number
2408            );
2409            self.ws_client
2410                .subscribe_subaccount(&self.wallet_address, self.subaccount_number)
2411                .await?;
2412            log::debug!(
2413                "Subscribed to subaccount updates: {}/{}",
2414                self.wallet_address,
2415                self.subaccount_number
2416            );
2417
2418            let stream = self.ws_client.stream();
2419            self.spawn_ws_stream_handler(stream)?;
2420
2421            // Fills require a registered account for portfolio updates
2422            self.await_account_registered(30.0).await?;
2423
2424            Ok::<(), anyhow::Error>(())
2425        }
2426        .await;
2427
2428        if let Err(e) = session_result {
2429            if let Err(teardown_error) = self.teardown_partial_connect().await {
2430                return Err(e.context(format!(
2431                    "dYdX execution startup teardown failed: {teardown_error}"
2432                )));
2433            }
2434            return Err(e);
2435        }
2436
2437        self.core.set_connected();
2438        setup_guard.disarm();
2439        log::info!("Connected: client_id={}", self.core.client_id);
2440        Ok(())
2441    }
2442
2443    async fn disconnect(&mut self) -> anyhow::Result<()> {
2444        log::info!("Disconnecting from dYdX");
2445
2446        self.begin_pending_shutdown();
2447        let ws_client = self.ws_client.clone();
2448        let disconnect_guard = TaskGroupGuard::new(&[&self.session_tasks], move || {
2449            ws_client.begin_shutdown();
2450        });
2451
2452        let _ = self
2453            .ws_client
2454            .unsubscribe_subaccount(&self.wallet_address, self.subaccount_number)
2455            .await
2456            .map_err(|e| log::warn!("Failed to unsubscribe from subaccount: {e}"));
2457
2458        let _ = self
2459            .ws_client
2460            .unsubscribe_markets()
2461            .await
2462            .map_err(|e| log::warn!("Failed to unsubscribe from markets: {e}"));
2463
2464        let _ = self
2465            .ws_client
2466            .unsubscribe_block_height()
2467            .await
2468            .map_err(|e| log::warn!("Failed to unsubscribe from block height: {e}"));
2469
2470        let shutdown_result = self.teardown_partial_connect().await;
2471        disconnect_guard.disarm();
2472        shutdown_result?;
2473        log::info!("Disconnected: client_id={}", self.core.client_id);
2474        Ok(())
2475    }
2476
2477    async fn generate_order_status_report(
2478        &self,
2479        cmd: &GenerateOrderStatusReport,
2480    ) -> anyhow::Result<Option<OrderStatusReport>> {
2481        // dYdX Indexer `/v4/orders` caps at `limit` and has no offset cursor, so we
2482        // request the maximum page to maximize the chance of finding a match
2483        // on active subaccounts. Callers looking for older orders should prefer
2484        // `generate_mass_status` or narrow via `instrument_id`.
2485        let market = cmd
2486            .instrument_id
2487            .map(|id| id.symbol.as_str().trim_end_matches("-PERP").to_string());
2488
2489        let response = self
2490            .http_client
2491            .inner
2492            .get_orders(
2493                &self.wallet_address,
2494                self.subaccount_number,
2495                market.as_deref(),
2496                Some(DYDX_INDEXER_REPORT_LIMIT),
2497            )
2498            .await
2499            .context("failed to fetch order from dYdX API")?;
2500
2501        if response.is_empty() {
2502            log::debug!(
2503                "No orders returned for {}/subaccount={} (market_filter={:?})",
2504                self.wallet_address,
2505                self.subaccount_number,
2506                market,
2507            );
2508            return Ok(None);
2509        }
2510
2511        let ts_init = UnixNanos::default();
2512        let scanned_count = response.len();
2513
2514        let report = find_matching_order_report(
2515            &response,
2516            cmd.instrument_id,
2517            cmd.client_order_id,
2518            cmd.venue_order_id,
2519            |clob_pair_id| self.get_instrument_by_clob_pair_id(clob_pair_id),
2520            &self.encoder,
2521            self.core.account_id,
2522            ts_init,
2523        )?;
2524
2525        if report.is_none() {
2526            // The target order was not in the fetched page. Surface the scope so
2527            // callers can tell whether the order is older than the page or the
2528            // filters simply didn't match any returned order.
2529            let page_full = scanned_count == DYDX_INDEXER_REPORT_LIMIT as usize;
2530            log::debug!(
2531                "No order matched filters for {}/subaccount={} \
2532                 (client_order_id={:?}, venue_order_id={:?}, instrument_id={:?}, \
2533                 scanned={scanned_count}, page_full={page_full}, limit={DYDX_INDEXER_REPORT_LIMIT})",
2534                self.wallet_address,
2535                self.subaccount_number,
2536                cmd.client_order_id,
2537                cmd.venue_order_id,
2538                cmd.instrument_id,
2539            );
2540        }
2541
2542        Ok(report)
2543    }
2544
2545    async fn generate_order_status_reports(
2546        &self,
2547        cmd: &GenerateOrderStatusReports,
2548    ) -> anyhow::Result<Vec<OrderStatusReport>> {
2549        let response = self
2550            .http_client
2551            .inner
2552            .get_orders(
2553                &self.wallet_address,
2554                self.subaccount_number,
2555                None, // market filter
2556                Some(DYDX_INDEXER_REPORT_LIMIT),
2557            )
2558            .await
2559            .context("failed to fetch orders from dYdX API")?;
2560
2561        let mut reports = Vec::new();
2562        let ts_init = UnixNanos::default();
2563
2564        for order in response {
2565            let instrument = match self.get_instrument_by_clob_pair_id(order.clob_pair_id) {
2566                Some(inst) => inst,
2567                None => continue,
2568            };
2569
2570            if let Some(filter_id) = cmd.instrument_id
2571                && instrument.id() != filter_id
2572            {
2573                continue;
2574            }
2575
2576            match parse_order_status_report(&order, &instrument, self.core.account_id, ts_init) {
2577                Ok(mut r) => {
2578                    if !order.client_id.is_empty()
2579                        && let Ok(client_id_u32) = order.client_id.parse::<u32>()
2580                    {
2581                        self.encoder.register_known_client_id(client_id_u32);
2582
2583                        if let Some(decoded) = self
2584                            .encoder
2585                            .decode_if_known(client_id_u32, order.client_metadata)
2586                        {
2587                            log::debug!(
2588                                "Decoded order: dYdX client_id={} meta={:#x} -> '{}'",
2589                                client_id_u32,
2590                                order.client_metadata,
2591                                decoded,
2592                            );
2593                            r.client_order_id = Some(decoded);
2594                        }
2595                    }
2596                    reports.push(r);
2597                }
2598                Err(e) => {
2599                    log::warn!("Failed to parse order status report: {e}");
2600                }
2601            }
2602        }
2603
2604        retain_order_status_reports(&mut reports, cmd);
2605
2606        // Drop reports that conflict with the local cache: if we already have
2607        // the order in a terminal status (FILLED/CANCELED/EXPIRED/REJECTED/DENIED),
2608        // replaying an `Accepted` from open-check would error in the ExecEngine.
2609        // The venue may legitimately still consider the order open due to clock
2610        // skew between our inferred fill and the next venue update; skipping
2611        // here keeps the engine quiet while reconciliation eventually catches up.
2612        {
2613            let cache = self.core.cache();
2614            reports.retain(|r| {
2615                let Some(cid) = r.client_order_id else {
2616                    return true;
2617                };
2618
2619                match cache.order(&cid) {
2620                    Some(order) if order.status().is_closed() => {
2621                        log::debug!(
2622                            "Skipping reconciliation report for terminal order {cid} \
2623                             (cache status={:?}, venue status={:?})",
2624                            order.status(),
2625                            r.order_status,
2626                        );
2627                        false
2628                    }
2629                    _ => true,
2630                }
2631            });
2632        }
2633
2634        Ok(reports)
2635    }
2636
2637    async fn generate_fill_reports(
2638        &self,
2639        cmd: GenerateFillReports,
2640    ) -> anyhow::Result<Vec<FillReport>> {
2641        let response = self
2642            .http_client
2643            .inner
2644            .get_fills(
2645                &self.wallet_address,
2646                self.subaccount_number,
2647                None, // market filter
2648                Some(DYDX_INDEXER_REPORT_LIMIT),
2649            )
2650            .await
2651            .context("failed to fetch fills from dYdX API")?;
2652
2653        let mut reports = Vec::new();
2654        let ts_init = UnixNanos::default();
2655
2656        for fill in response.fills {
2657            let instrument = match self.get_instrument_by_market(&fill.market) {
2658                Some(inst) => inst,
2659                None => {
2660                    log::warn!("Unknown market in fill: {}", fill.market);
2661                    continue;
2662                }
2663            };
2664
2665            if let Some(filter_id) = cmd.instrument_id
2666                && instrument.id() != filter_id
2667            {
2668                continue;
2669            }
2670
2671            let report = match parse_fill_report(&fill, &instrument, self.core.account_id, ts_init)
2672            {
2673                Ok(r) => r,
2674                Err(e) => {
2675                    log::warn!("Failed to parse fill report: {e}");
2676                    continue;
2677                }
2678            };
2679
2680            reports.push(report);
2681        }
2682
2683        if let Some(venue_order_id) = cmd.venue_order_id {
2684            reports.retain(|r| r.venue_order_id.as_str() == venue_order_id.as_str());
2685        }
2686
2687        Ok(reports)
2688    }
2689
2690    async fn generate_position_status_reports(
2691        &self,
2692        cmd: &GeneratePositionStatusReports,
2693    ) -> anyhow::Result<Vec<PositionStatusReport>> {
2694        let response = self
2695            .http_client
2696            .inner
2697            .get_subaccount(&self.wallet_address, self.subaccount_number)
2698            .await
2699            .context("failed to fetch subaccount from dYdX API")?;
2700
2701        let mut reports = Vec::new();
2702        let ts_init = UnixNanos::default();
2703
2704        for (market_ticker, perp_position) in &response.subaccount.open_perpetual_positions {
2705            let instrument = match self.get_instrument_by_market(market_ticker) {
2706                Some(inst) => inst,
2707                None => {
2708                    log::warn!("Unknown market in position: {market_ticker}");
2709                    continue;
2710                }
2711            };
2712
2713            if let Some(filter_id) = cmd.instrument_id
2714                && instrument.id() != filter_id
2715            {
2716                continue;
2717            }
2718
2719            let report = match parse_position_status_report(
2720                perp_position,
2721                &instrument,
2722                self.core.account_id,
2723                ts_init,
2724            ) {
2725                Ok(r) => r,
2726                Err(e) => {
2727                    log::warn!("Failed to parse position status report: {e}");
2728                    continue;
2729                }
2730            };
2731
2732            reports.push(report);
2733        }
2734
2735        Ok(reports)
2736    }
2737
2738    async fn generate_mass_status(
2739        &self,
2740        lookback_mins: Option<u64>,
2741    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
2742        let ts_init = self.clock.get_time_ns();
2743
2744        let orders_response = self
2745            .http_client
2746            .inner
2747            .get_orders(
2748                &self.wallet_address,
2749                self.subaccount_number,
2750                None,
2751                Some(DYDX_INDEXER_REPORT_LIMIT),
2752            )
2753            .await
2754            .context("failed to fetch orders for mass status")?;
2755
2756        let subaccount_response = self
2757            .http_client
2758            .inner
2759            .get_subaccount(&self.wallet_address, self.subaccount_number)
2760            .await
2761            .context("failed to fetch subaccount for mass status")?;
2762
2763        let fills_response = self
2764            .http_client
2765            .inner
2766            .get_fills(
2767                &self.wallet_address,
2768                self.subaccount_number,
2769                None,
2770                Some(DYDX_INDEXER_REPORT_LIMIT),
2771            )
2772            .await
2773            .context("failed to fetch fills for mass status")?;
2774
2775        let mut order_reports = Vec::new();
2776        let mut orders_filtered = 0usize;
2777
2778        for order in orders_response {
2779            let instrument = match self.get_instrument_by_clob_pair_id(order.clob_pair_id) {
2780                Some(inst) => inst,
2781                None => {
2782                    orders_filtered += 1;
2783                    continue;
2784                }
2785            };
2786
2787            match parse_order_status_report(&order, &instrument, self.core.account_id, ts_init) {
2788                Ok(mut r) => {
2789                    if !order.client_id.is_empty()
2790                        && let Ok(client_id_u32) = order.client_id.parse::<u32>()
2791                    {
2792                        self.encoder.register_known_client_id(client_id_u32);
2793
2794                        if let Some(decoded) = self
2795                            .encoder
2796                            .decode_if_known(client_id_u32, order.client_metadata)
2797                        {
2798                            log::debug!(
2799                                "Decoded reconciliation order: dYdX client_id={} meta={:#x} -> '{}'",
2800                                client_id_u32,
2801                                order.client_metadata,
2802                                decoded,
2803                            );
2804                            r.client_order_id = Some(decoded);
2805                        }
2806                    }
2807                    order_reports.push(r);
2808                }
2809                Err(e) => {
2810                    log::warn!("Failed to parse order status report: {e}");
2811                    orders_filtered += 1;
2812                }
2813            }
2814        }
2815
2816        let mut position_reports = Vec::new();
2817
2818        for (market_ticker, perp_position) in
2819            &subaccount_response.subaccount.open_perpetual_positions
2820        {
2821            let instrument = match self.get_instrument_by_market(market_ticker) {
2822                Some(inst) => inst,
2823                None => continue,
2824            };
2825
2826            match parse_position_status_report(
2827                perp_position,
2828                &instrument,
2829                self.core.account_id,
2830                ts_init,
2831            ) {
2832                Ok(r) => position_reports.push(r),
2833                Err(e) => {
2834                    log::warn!("Failed to parse position status report: {e}");
2835                }
2836            }
2837        }
2838
2839        let mut fill_reports = Vec::new();
2840        let mut fills_filtered = 0usize;
2841
2842        for fill in fills_response.fills {
2843            let instrument = match self.get_instrument_by_market(&fill.market) {
2844                Some(inst) => inst,
2845                None => {
2846                    fills_filtered += 1;
2847                    continue;
2848                }
2849            };
2850
2851            match parse_fill_report(&fill, &instrument, self.core.account_id, ts_init) {
2852                Ok(r) => fill_reports.push(r),
2853                Err(e) => {
2854                    log::warn!("Failed to parse fill report: {e}");
2855                    fills_filtered += 1;
2856                }
2857            }
2858        }
2859
2860        apply_avg_px_from_fills(&mut order_reports, &fill_reports);
2861
2862        // Drop reports that conflict with the local cache: same rationale as in
2863        // `generate_order_status_reports`. On a cold start the cache is empty so
2864        // this is a no-op; on a hot reconciliation pass it prevents replaying
2865        // `Accepted` events for orders the engine already considers terminal.
2866        {
2867            let cache = self.core.cache();
2868            order_reports.retain(|r| {
2869                let Some(cid) = r.client_order_id else {
2870                    return true;
2871                };
2872
2873                match cache.order(&cid) {
2874                    Some(order) if order.status().is_closed() => {
2875                        log::debug!(
2876                            "Skipping reconciliation report for terminal order {cid} \
2877                             (cache status={:?}, venue status={:?})",
2878                            order.status(),
2879                            r.order_status,
2880                        );
2881                        false
2882                    }
2883                    _ => true,
2884                }
2885            });
2886        }
2887
2888        if let Some(mins) = lookback_mins {
2889            let now_ns = self.clock.get_time_ns();
2890            let cutoff = now_ns.saturating_sub(DurationNanos::try_from_mins(mins)?);
2891
2892            let orders_before = order_reports.len();
2893            order_reports.retain(|r| r.ts_last >= cutoff);
2894            let orders_removed = orders_before - order_reports.len();
2895
2896            let fills_before = fill_reports.len();
2897            fill_reports.retain(|r| r.ts_event >= cutoff);
2898            let fills_removed = fills_before - fill_reports.len();
2899
2900            log::debug!(
2901                "Lookback filter ({}min): orders {}->{} (removed {}), fills {}->{} (removed {}), positions {} (unfiltered)",
2902                mins,
2903                orders_before,
2904                order_reports.len(),
2905                orders_removed,
2906                fills_before,
2907                fill_reports.len(),
2908                fills_removed,
2909                position_reports.len(),
2910            );
2911        } else {
2912            log::debug!(
2913                "Generated mass status: {} orders ({} filtered), {} positions, {} fills ({} filtered)",
2914                order_reports.len(),
2915                orders_filtered,
2916                position_reports.len(),
2917                fill_reports.len(),
2918                fills_filtered,
2919            );
2920        }
2921
2922        let mut mass_status = ExecutionMassStatus::new(
2923            self.core.client_id,
2924            self.core.account_id,
2925            self.core.venue,
2926            ts_init,
2927            None, // report_id will be auto-generated
2928        );
2929
2930        mass_status.add_order_reports(order_reports);
2931        mass_status.add_position_reports(position_reports);
2932        mass_status.add_fill_reports(fill_reports);
2933
2934        Ok(Some(mass_status))
2935    }
2936}
2937
2938type CancelAllOrderData = (
2939    StrategyId,
2940    ClientOrderId,
2941    Option<VenueOrderId>,
2942    TimeInForce,
2943    Option<UnixNanos>,
2944);
2945
2946#[derive(Clone, Copy)]
2947struct DydxCancelOrderRequest {
2948    instrument_id: InstrumentId,
2949    client_id: u32,
2950    order_flags: u32,
2951    strategy_id: StrategyId,
2952    client_order_id: ClientOrderId,
2953    venue_order_id: Option<VenueOrderId>,
2954}
2955
2956fn apply_avg_px_from_fills(order_reports: &mut [OrderStatusReport], fill_reports: &[FillReport]) {
2957    let mut totals: AHashMap<VenueOrderId, (Decimal, Decimal)> = AHashMap::new();
2958    for fill in fill_reports {
2959        let entry = totals.entry(fill.venue_order_id).or_default();
2960        let qty = fill.last_qty.as_decimal();
2961        entry.0 += fill.last_px.as_decimal() * qty;
2962        entry.1 += qty;
2963    }
2964
2965    for report in order_reports {
2966        if let Some((notional, total_qty)) = totals.get(&report.venue_order_id)
2967            && !total_qty.is_zero()
2968        {
2969            report.avg_px = Some(notional / total_qty);
2970        }
2971    }
2972}
2973
2974struct PendingTaskLabel {
2975    id: u64,
2976    labels: Arc<Mutex<AHashMap<u64, &'static str>>>,
2977}
2978
2979impl Drop for PendingTaskLabel {
2980    fn drop(&mut self) {
2981        self.labels.lock().remove(&self.id);
2982    }
2983}
2984
2985#[allow(clippy::too_many_arguments)]
2986fn find_matching_order_report<F>(
2987    orders: &[crate::http::models::Order],
2988    instrument_filter: Option<InstrumentId>,
2989    client_order_id_filter: Option<ClientOrderId>,
2990    venue_order_id_filter: Option<VenueOrderId>,
2991    lookup_instrument: F,
2992    encoder: &ClientOrderIdEncoder,
2993    account_id: AccountId,
2994    ts_init: UnixNanos,
2995) -> anyhow::Result<Option<OrderStatusReport>>
2996where
2997    F: Fn(u32) -> Option<InstrumentAny>,
2998{
2999    for order in orders {
3000        let instrument = match lookup_instrument(order.clob_pair_id) {
3001            Some(inst) => inst,
3002            None => continue,
3003        };
3004
3005        if let Some(filter_id) = instrument_filter
3006            && instrument.id() != filter_id
3007        {
3008            continue;
3009        }
3010
3011        let mut report = parse_order_status_report(order, &instrument, account_id, ts_init)
3012            .context("failed to parse order status report")?;
3013
3014        if !order.client_id.is_empty()
3015            && let Ok(client_id_u32) = order.client_id.parse::<u32>()
3016        {
3017            encoder.register_known_client_id(client_id_u32);
3018
3019            if let Some(decoded) = encoder.decode_if_known(client_id_u32, order.client_metadata) {
3020                log::debug!(
3021                    "Decoded order: dYdX client_id={} meta={:#x} -> '{}'",
3022                    client_id_u32,
3023                    order.client_metadata,
3024                    decoded,
3025                );
3026                report.client_order_id = Some(decoded);
3027            }
3028        }
3029
3030        if let Some(client_order_id) = client_order_id_filter
3031            && report.client_order_id != Some(client_order_id)
3032        {
3033            continue;
3034        }
3035
3036        if let Some(venue_order_id) = venue_order_id_filter
3037            && report.venue_order_id.as_str() != venue_order_id.as_str()
3038        {
3039            continue;
3040        }
3041
3042        return Ok(Some(report));
3043    }
3044
3045    Ok(None)
3046}
3047
3048#[cfg(test)]
3049mod tests {
3050    use std::{cell::RefCell, rc::Rc};
3051
3052    use axum::{Json, Router, http::Uri, routing::get};
3053    use jiff::Timestamp;
3054    use nautilus_common::{
3055        cache::Cache, clock::VirtualClock, factories::OrderFactory, messages::ExecutionEvent,
3056    };
3057    use nautilus_model::{
3058        enums::OrderSide,
3059        identifiers::{Symbol, TraderId},
3060        instruments::{CryptoPerpetual, InstrumentAny},
3061        orders::{Order as _, OrderAny},
3062        types::{Currency, Price, Quantity},
3063    };
3064    use rstest::rstest;
3065    use rust_decimal_macros::dec;
3066    use serde_json::Value;
3067
3068    use super::*;
3069    use crate::{
3070        common::{
3071            consts::DYDX_CLIENT_ID,
3072            enums::{DydxOrderStatus, DydxOrderType, DydxTimeInForce},
3073        },
3074        http::models::Order,
3075    };
3076
3077    fn test_instrument(symbol: &str, venue: &str) -> InstrumentAny {
3078        let instrument_id = InstrumentId::new(Symbol::new(symbol), Venue::new(venue));
3079        InstrumentAny::CryptoPerpetual(
3080            CryptoPerpetual::builder()
3081                .instrument_id(instrument_id)
3082                .raw_symbol(instrument_id.symbol)
3083                .base_currency(Currency::BTC())
3084                .quote_currency(Currency::USD())
3085                .settlement_currency(Currency::USD())
3086                .is_inverse(false)
3087                .price_precision(2)
3088                .size_precision(3)
3089                .price_increment(Price::new(0.01, 2))
3090                .size_increment(Quantity::new(0.001, 3))
3091                .ts_event(UnixNanos::default())
3092                .ts_init(UnixNanos::default())
3093                .build()
3094                .unwrap(),
3095        )
3096    }
3097
3098    fn test_order(id: &str, clob_pair_id: u32, client_id: &str) -> Order {
3099        Order {
3100            id: id.to_string(),
3101            subaccount_id: "sub-1".to_string(),
3102            client_id: client_id.to_string(),
3103            clob_pair_id,
3104            side: OrderSide::Buy,
3105            size: dec!(1.0),
3106            total_filled: dec!(0),
3107            price: dec!(50000),
3108            status: DydxOrderStatus::Open,
3109            order_type: DydxOrderType::Limit,
3110            time_in_force: DydxTimeInForce::Gtt,
3111            reduce_only: false,
3112            post_only: false,
3113            order_flags: 64,
3114            good_til_block: None,
3115            good_til_block_time: None,
3116            created_at_height: Some(100),
3117            client_metadata: 4,
3118            trigger_price: None,
3119            condition_type: None,
3120            conditional_order_trigger_subticks: None,
3121            execution: None,
3122            updated_at: None,
3123            updated_at_height: None,
3124            ticker: None,
3125            subaccount_number: 0,
3126            order_router_address: None,
3127        }
3128    }
3129
3130    fn test_order_factory() -> OrderFactory {
3131        let clock = Rc::new(RefCell::new(VirtualClock::new()));
3132        OrderFactory::new(
3133            TraderId::from("TRADER-001"),
3134            StrategyId::from("S-001"),
3135            Some(0),
3136            Some(0),
3137            clock,
3138            false,
3139            false,
3140        )
3141    }
3142
3143    fn create_execution_client() -> (
3144        DydxExecutionClient,
3145        Rc<RefCell<Cache>>,
3146        tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
3147    ) {
3148        const TEST_PRIVATE_KEY: &str =
3149            "0000000000000000000000000000000000000000000000000000000000000001";
3150
3151        let cache = Rc::new(RefCell::new(Cache::default()));
3152        let core = ExecutionClientCore::new(
3153            TraderId::from("TRADER-001"),
3154            *DYDX_CLIENT_ID,
3155            *DYDX_VENUE,
3156            OmsType::Netting,
3157            AccountId::from("DYDX-001"),
3158            AccountType::Margin,
3159            None,
3160            cache.clone(),
3161        );
3162        core.set_connected();
3163
3164        let config = DydxAdapterConfig {
3165            base_url: "http://127.0.0.1:1".to_string(),
3166            ws_url: "ws://127.0.0.1:1".to_string(),
3167            grpc_url: "http://127.0.0.1:1".to_string(),
3168            grpc_urls: vec!["http://127.0.0.1:1".to_string()],
3169            wallet_address: Some("dydx1test".to_string()),
3170            private_key: Some(TEST_PRIVATE_KEY.into()),
3171            ..Default::default()
3172        };
3173
3174        let mut client =
3175            DydxExecutionClient::new(core, config, "dydx1test".to_string(), 0).unwrap();
3176        client
3177            .block_time_monitor
3178            .record_block(100, Timestamp::now());
3179
3180        let (sender, receiver) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
3181        client.emitter.set_sender(sender);
3182
3183        (client, cache, receiver)
3184    }
3185
3186    #[rstest]
3187    #[tokio::test]
3188    async fn test_mass_status_captures_collection_start() {
3189        let clock: &'static AtomicTime =
3190            Box::leak(Box::new(AtomicTime::new(false, UnixNanos::from(100))));
3191
3192        let router = Router::new().fallback(get(move |uri: Uri| async move {
3193            clock.set_time(UnixNanos::from(200));
3194
3195            let fixture = if uri.path() == "/v4/orders" {
3196                include_str!("../../test_data/http_get_orders.json")
3197            } else if uri.path() == "/v4/fills" {
3198                include_str!("../../test_data/http_get_fills.json")
3199            } else {
3200                include_str!("../../test_data/http_get_subaccount.json")
3201            };
3202
3203            let value: Value = serde_json::from_str(fixture).unwrap();
3204            Json(value["result"].clone())
3205        }));
3206
3207        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
3208        let addr = listener.local_addr().unwrap();
3209
3210        let server = tokio::spawn(async move {
3211            axum::serve(listener, router).await.unwrap();
3212        });
3213
3214        let (mut client, _cache, _rx) = create_execution_client();
3215        client.clock = clock;
3216        client.http_client = DydxHttpClient::new(
3217            Some(format!("http://{addr}")),
3218            5,
3219            None,
3220            client.config.network,
3221            None,
3222        )
3223        .unwrap();
3224
3225        let snapshot = client.generate_mass_status(None).await.unwrap().unwrap();
3226        server.abort();
3227
3228        assert_eq!(snapshot.ts_init, UnixNanos::from(100));
3229        assert_eq!(clock.get_time_ns(), UnixNanos::from(200));
3230    }
3231
3232    fn cache_order(cache: &Rc<RefCell<Cache>>, order: OrderAny) {
3233        cache
3234            .borrow_mut()
3235            .add_order(order, None, Some(*DYDX_CLIENT_ID), false)
3236            .unwrap();
3237    }
3238
3239    fn recv_order_event(
3240        rx: &mut tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
3241    ) -> OrderEventAny {
3242        match rx.try_recv().expect("expected execution event") {
3243            ExecutionEvent::Order(event) => event,
3244            event => panic!("Expected order event, was {event:?}"),
3245        }
3246    }
3247
3248    #[rstest]
3249    fn test_submit_order_denies_quote_quantity() {
3250        let (client, cache, mut rx) = create_execution_client();
3251        let mut factory = test_order_factory();
3252        let instrument_id = InstrumentId::from("BTC-USD-PERP.DYDX");
3253        let order = factory.market(
3254            instrument_id,
3255            OrderSide::Buy,
3256            Quantity::from("100"),
3257            Some(TimeInForce::Ioc),
3258            None,
3259            Some(true),
3260            None,
3261            None,
3262            None,
3263            Some(ClientOrderId::from("O-QUOTE-QTY")),
3264        );
3265        cache_order(&cache, order.clone());
3266
3267        let command = SubmitOrder::from_order(
3268            &order,
3269            order.trader_id(),
3270            Some(*DYDX_CLIENT_ID),
3271            None,
3272            UUID4::new(),
3273            UnixNanos::default(),
3274        );
3275        client.submit_order(command).unwrap();
3276
3277        match recv_order_event(&mut rx) {
3278            OrderEventAny::Denied(event) => {
3279                assert_eq!(event.client_order_id, order.client_order_id());
3280                assert_eq!(event.instrument_id, instrument_id);
3281                assert_eq!(event.reason, "UNSUPPORTED_QUOTE_QUANTITY");
3282            }
3283            event => panic!("Expected order denied event, was {event:?}"),
3284        }
3285    }
3286
3287    #[rstest]
3288    fn test_submit_order_denies_market_fok() {
3289        let (client, cache, mut rx) = create_execution_client();
3290        let mut factory = test_order_factory();
3291        let instrument_id = InstrumentId::from("BTC-USD-PERP.DYDX");
3292        let order = factory.market(
3293            instrument_id,
3294            OrderSide::Buy,
3295            Quantity::from("0.1"),
3296            Some(TimeInForce::Fok),
3297            None,
3298            Some(false),
3299            None,
3300            None,
3301            None,
3302            Some(ClientOrderId::from("O-MARKET-FOK")),
3303        );
3304        cache_order(&cache, order.clone());
3305
3306        let command = SubmitOrder::from_order(
3307            &order,
3308            order.trader_id(),
3309            Some(*DYDX_CLIENT_ID),
3310            None,
3311            UUID4::new(),
3312            UnixNanos::default(),
3313        );
3314        client.submit_order(command).unwrap();
3315
3316        match recv_order_event(&mut rx) {
3317            OrderEventAny::Denied(event) => {
3318                assert_eq!(event.client_order_id, order.client_order_id());
3319                assert_eq!(event.instrument_id, instrument_id);
3320                assert!(
3321                    event.reason.contains("Fill-or-kill") && event.reason.contains("deprecated"),
3322                    "expected FOK-deprecation reason, was: {}",
3323                    event.reason,
3324                );
3325            }
3326            event => panic!("Expected order denied event, was {event:?}"),
3327        }
3328    }
3329
3330    #[rstest]
3331    fn test_submit_order_denies_limit_fok() {
3332        // dYdX v4 deprecated FOK at the protocol level (chain returns code=48).
3333        // The adapter must deny FOK pre-submission so the strategy gets an
3334        // immediate `OrderDenied` instead of a venue-side broadcast failure.
3335        let (client, cache, mut rx) = create_execution_client();
3336        let mut factory = test_order_factory();
3337        let instrument_id = InstrumentId::from("BTC-USD-PERP.DYDX");
3338        let order = factory.limit(
3339            instrument_id,
3340            OrderSide::Buy,
3341            Quantity::from("0.1"),
3342            Price::from("50000.0"),
3343            Some(TimeInForce::Fok),
3344            None,
3345            Some(false),
3346            None,
3347            Some(false),
3348            None,
3349            None,
3350            None,
3351            None,
3352            None,
3353            None,
3354            Some(ClientOrderId::from("O-LIMIT-FOK")),
3355        );
3356        cache_order(&cache, order.clone());
3357
3358        let command = SubmitOrder::from_order(
3359            &order,
3360            order.trader_id(),
3361            Some(*DYDX_CLIENT_ID),
3362            None,
3363            UUID4::new(),
3364            UnixNanos::default(),
3365        );
3366        client.submit_order(command).unwrap();
3367
3368        match recv_order_event(&mut rx) {
3369            OrderEventAny::Denied(event) => {
3370                assert_eq!(event.client_order_id, order.client_order_id());
3371                assert!(
3372                    event.reason.contains("Fill-or-kill") && event.reason.contains("deprecated"),
3373                    "expected FOK-deprecation reason, was: {}",
3374                    event.reason,
3375                );
3376            }
3377            event => panic!("Expected order denied event, was {event:?}"),
3378        }
3379    }
3380
3381    #[rstest]
3382    fn test_submit_order_denies_day_tif() {
3383        // DAY TIF is not supported by dYdX v4; the adapter must deny pre-submission
3384        // rather than silently translate to GTC.
3385        let (client, cache, mut rx) = create_execution_client();
3386        let mut factory = test_order_factory();
3387        let instrument_id = InstrumentId::from("BTC-USD-PERP.DYDX");
3388        let order = factory.limit(
3389            instrument_id,
3390            OrderSide::Buy,
3391            Quantity::from("0.1"),
3392            Price::from("50000.0"),
3393            Some(TimeInForce::Day),
3394            None,
3395            Some(false),
3396            None,
3397            Some(false),
3398            None,
3399            None,
3400            None,
3401            None,
3402            None,
3403            None,
3404            Some(ClientOrderId::from("O-LIMIT-DAY")),
3405        );
3406        cache_order(&cache, order.clone());
3407
3408        let command = SubmitOrder::from_order(
3409            &order,
3410            order.trader_id(),
3411            Some(*DYDX_CLIENT_ID),
3412            None,
3413            UUID4::new(),
3414            UnixNanos::default(),
3415        );
3416        client.submit_order(command).unwrap();
3417
3418        match recv_order_event(&mut rx) {
3419            OrderEventAny::Denied(event) => {
3420                assert_eq!(event.client_order_id, order.client_order_id());
3421                assert!(
3422                    event.reason.contains("DAY"),
3423                    "expected DAY-unsupported reason, was: {}",
3424                    event.reason,
3425                );
3426            }
3427            event => panic!("Expected order denied event, was {event:?}"),
3428        }
3429    }
3430
3431    #[rstest]
3432    #[tokio::test]
3433    async fn test_submit_order_list_denies_quote_quantity() {
3434        use nautilus_model::{
3435            identifiers::OrderListId,
3436            orders::{OrderList, builder::OrderTestBuilder},
3437        };
3438
3439        let requests = Arc::new(std::sync::atomic::AtomicUsize::new(0));
3440        let observed_requests = Arc::clone(&requests);
3441        let router = Router::new().fallback(move || {
3442            observed_requests.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
3443            async { axum::http::StatusCode::NOT_FOUND }
3444        });
3445
3446        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
3447        let addr = listener.local_addr().unwrap();
3448        let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
3449        let grpc = DydxGrpcClient::new(format!("http://{addr}")).await.unwrap();
3450        let (mut client, cache, mut rx) = create_execution_client();
3451        client.tx_manager = Some(Arc::new(
3452            TransactionManager::new(
3453                grpc.clone(),
3454                &DydxExecutionClient::resolve_private_key(&client.config).unwrap(),
3455                client.wallet_address.clone(),
3456                client.get_chain_id(),
3457            )
3458            .unwrap(),
3459        ));
3460        client.broadcaster = Some(Arc::new(TxBroadcaster::new(
3461            grpc,
3462            client.config.grpc_quota(),
3463        )));
3464        client.order_builder = Some(Arc::new(OrderMessageBuilder::new(
3465            client.http_client.clone(),
3466            client.wallet_address.clone(),
3467            client.subaccount_number,
3468            client.block_time_monitor.clone(),
3469        )));
3470        let order = OrderTestBuilder::new(OrderType::Limit)
3471            .instrument_id(InstrumentId::from("BTC-USD-PERP.DYDX"))
3472            .quantity(Quantity::from("100"))
3473            .price(Price::from("50000"))
3474            .quote_quantity(true)
3475            .build();
3476        cache_order(&cache, order.clone());
3477
3478        let order_list = OrderList::new(
3479            OrderListId::from("OL-QUOTE"),
3480            order.instrument_id(),
3481            order.strategy_id(),
3482            vec![order.client_order_id()],
3483            UnixNanos::default(),
3484        );
3485
3486        let command = SubmitOrderList::new(
3487            order.trader_id(),
3488            Some(*DYDX_CLIENT_ID),
3489            order.strategy_id(),
3490            order_list,
3491            vec![order.init_event().clone()],
3492            None,
3493            None,
3494            None,
3495            UUID4::new(),
3496            UnixNanos::default(),
3497            None,
3498        );
3499
3500        client.submit_order_list(command).unwrap();
3501
3502        let OrderEventAny::Denied(denied) = recv_order_event(&mut rx) else {
3503            panic!("Expected order denial");
3504        };
3505
3506        assert_eq!(denied.client_order_id, order.client_order_id());
3507        assert_eq!(denied.instrument_id, order.instrument_id());
3508        assert_eq!(denied.reason.as_str(), "UNSUPPORTED_QUOTE_QUANTITY");
3509        assert_eq!(
3510            rx.try_recv().unwrap_err(),
3511            tokio::sync::mpsc::error::TryRecvError::Empty
3512        );
3513        assert_eq!(requests.load(std::sync::atomic::Ordering::Relaxed), 0);
3514        server.abort();
3515    }
3516
3517    // `submit_order_list` mirrors the single-submit TIF gate; an order in the
3518    // list with FOK or DAY must be denied pre-submission rather than
3519    // forwarded to the gRPC builder.
3520    #[rstest]
3521    #[case(TimeInForce::Fok, "Fill-or-kill")]
3522    #[case(TimeInForce::Day, "DAY")]
3523    fn test_submit_order_list_denies_unsupported_tif(
3524        #[case] tif: TimeInForce,
3525        #[case] reason_substring: &str,
3526    ) {
3527        use nautilus_common::messages::execution::SubmitOrderList;
3528        use nautilus_model::{identifiers::OrderListId, orders::OrderList};
3529
3530        let (client, cache, mut rx) = create_execution_client();
3531        let mut factory = test_order_factory();
3532        let instrument_id = InstrumentId::from("BTC-USD-PERP.DYDX");
3533        let order = factory.limit(
3534            instrument_id,
3535            OrderSide::Buy,
3536            Quantity::from("0.1"),
3537            Price::from("50000.0"),
3538            Some(tif),
3539            None,
3540            Some(false),
3541            None,
3542            Some(false),
3543            None,
3544            None,
3545            None,
3546            None,
3547            None,
3548            None,
3549            Some(ClientOrderId::from("O-LIST-1")),
3550        );
3551        cache_order(&cache, order.clone());
3552
3553        let order_list = OrderList::new(
3554            OrderListId::from("OL-1"),
3555            instrument_id,
3556            order.strategy_id(),
3557            vec![order.client_order_id()],
3558            UnixNanos::default(),
3559        );
3560        let init = order.init_event().clone();
3561
3562        let cmd = SubmitOrderList::new(
3563            order.trader_id(),
3564            Some(*DYDX_CLIENT_ID),
3565            order.strategy_id(),
3566            order_list,
3567            vec![init],
3568            None,
3569            None,
3570            None,
3571            UUID4::new(),
3572            UnixNanos::default(),
3573            None, // correlation_id
3574        );
3575        client.submit_order_list(cmd).unwrap();
3576
3577        match recv_order_event(&mut rx) {
3578            OrderEventAny::Denied(event) => {
3579                assert_eq!(event.client_order_id, order.client_order_id());
3580                assert!(
3581                    event.reason.contains(reason_substring),
3582                    "expected reason containing {reason_substring:?}, was: {}",
3583                    event.reason,
3584                );
3585            }
3586            event => panic!("Expected order denied event, was {event:?}"),
3587        }
3588    }
3589
3590    // A mixed list must not leave supported sibling orders unresolved: when
3591    // the list is aborted for an unsupported TIF, every order is denied.
3592    #[rstest]
3593    fn test_submit_order_list_denies_supported_siblings_on_abort() {
3594        use nautilus_common::messages::execution::SubmitOrderList;
3595        use nautilus_model::{identifiers::OrderListId, orders::OrderList};
3596
3597        let (client, cache, mut rx) = create_execution_client();
3598        let mut factory = test_order_factory();
3599        let instrument_id = InstrumentId::from("BTC-USD-PERP.DYDX");
3600        let order_gtc = factory.limit(
3601            instrument_id,
3602            OrderSide::Buy,
3603            Quantity::from("0.1"),
3604            Price::from("50000.0"),
3605            Some(TimeInForce::Gtc),
3606            None,
3607            Some(false),
3608            None,
3609            Some(false),
3610            None,
3611            None,
3612            None,
3613            None,
3614            None,
3615            None,
3616            Some(ClientOrderId::from("O-MIXED-GTC")),
3617        );
3618        let order_fok = factory.limit(
3619            instrument_id,
3620            OrderSide::Sell,
3621            Quantity::from("0.1"),
3622            Price::from("51000.0"),
3623            Some(TimeInForce::Fok),
3624            None,
3625            Some(false),
3626            None,
3627            Some(false),
3628            None,
3629            None,
3630            None,
3631            None,
3632            None,
3633            None,
3634            Some(ClientOrderId::from("O-MIXED-FOK")),
3635        );
3636        cache_order(&cache, order_gtc.clone());
3637        cache_order(&cache, order_fok.clone());
3638
3639        let order_list = OrderList::new(
3640            OrderListId::from("OL-MIXED"),
3641            instrument_id,
3642            order_gtc.strategy_id(),
3643            vec![order_gtc.client_order_id(), order_fok.client_order_id()],
3644            UnixNanos::default(),
3645        );
3646
3647        let cmd = SubmitOrderList::new(
3648            order_gtc.trader_id(),
3649            Some(*DYDX_CLIENT_ID),
3650            order_gtc.strategy_id(),
3651            order_list,
3652            vec![
3653                order_gtc.init_event().clone(),
3654                order_fok.init_event().clone(),
3655            ],
3656            None,
3657            None,
3658            None,
3659            UUID4::new(),
3660            UnixNanos::default(),
3661            None, // correlation_id
3662        );
3663        client.submit_order_list(cmd).unwrap();
3664
3665        let mut denied_reasons = std::collections::HashMap::new();
3666
3667        for _ in 0..2 {
3668            match recv_order_event(&mut rx) {
3669                OrderEventAny::Denied(event) => {
3670                    denied_reasons.insert(event.client_order_id, event.reason);
3671                }
3672                event => panic!("Expected order denied event, was {event:?}"),
3673            }
3674        }
3675
3676        assert!(
3677            denied_reasons[&order_gtc.client_order_id()].contains("sibling order"),
3678            "GTC sibling should be denied with list-abort reason",
3679        );
3680        assert!(
3681            denied_reasons[&order_fok.client_order_id()].contains("Fill-or-kill"),
3682            "FOK order should be denied with TIF reason",
3683        );
3684    }
3685
3686    // Shutdown should retain cancel-related labels until the registered task
3687    // completes or its join is observed after forced abort.
3688    #[tokio::test]
3689    async fn test_pending_task_labels_drain_with_scope() {
3690        let (client, _cache, _rx) = create_execution_client();
3691
3692        client.spawn_labeled("cancel_all_orders", async {
3693            tokio::time::sleep(std::time::Duration::from_secs(60)).await;
3694        });
3695        client.spawn_labeled("submit_order", async {
3696            tokio::time::sleep(std::time::Duration::from_secs(60)).await;
3697        });
3698
3699        {
3700            let labels = client.pending_task_labels.lock();
3701            assert_eq!(labels.len(), 2);
3702        }
3703
3704        client.begin_pending_shutdown();
3705        client
3706            .pending_tasks
3707            .finish_shutdown(Duration::ZERO, Duration::from_secs(1))
3708            .await
3709            .unwrap();
3710
3711        assert!(client.pending_tasks.is_empty());
3712        assert!(client.pending_task_labels.lock().is_empty());
3713    }
3714
3715    // The label substring used by `begin_pending_shutdown` to recognize cancel
3716    // tasks must match the labels actually used at spawn sites. Pin those
3717    // sites here so a rename in only one place is caught.
3718    #[rstest]
3719    #[case("cancel_all_orders", true)]
3720    #[case("batch_cancel_orders", true)]
3721    #[case("Submit order", false)]
3722    #[case("batch_submit_short_term", false)]
3723    #[case("batch_submit_long_term", false)]
3724    fn test_pending_task_label_classification(
3725        #[case] label: &'static str,
3726        #[case] is_cancel: bool,
3727    ) {
3728        // The classification branch is `label.contains("cancel")` (lowercase).
3729        // This test locks the substring so an accidental rename of a labeled
3730        // spawn site breaks the assertion at compile time.
3731        assert_eq!(label.contains("cancel"), is_cancel);
3732    }
3733
3734    #[rstest]
3735    fn test_emit_partitioned_cancel_rejections_emits_each_cancel() {
3736        let clock = get_atomic_clock_realtime();
3737        let mut emitter = ExecutionEventEmitter::new(
3738            clock,
3739            TraderId::from("TRADER-001"),
3740            AccountId::from("DYDX-001"),
3741            AccountType::Margin,
3742            None,
3743        );
3744        let (sender, mut rx) = tokio::sync::mpsc::unbounded_channel::<ExecutionEvent>();
3745        emitter.set_sender(sender);
3746
3747        let instrument_id = InstrumentId::from("BTC-USD-PERP.DYDX");
3748        let orders = vec![
3749            DydxCancelOrderRequest {
3750                instrument_id,
3751                client_id: 1,
3752                order_flags: types::ORDER_FLAG_SHORT_TERM,
3753                strategy_id: StrategyId::from("S-001"),
3754                client_order_id: ClientOrderId::from("O-CANCEL-1"),
3755                venue_order_id: Some(VenueOrderId::from("V-1")),
3756            },
3757            DydxCancelOrderRequest {
3758                instrument_id,
3759                client_id: 2,
3760                order_flags: types::ORDER_FLAG_LONG_TERM,
3761                strategy_id: StrategyId::from("S-001"),
3762                client_order_id: ClientOrderId::from("O-CANCEL-2"),
3763                venue_order_id: Some(VenueOrderId::from("V-2")),
3764            },
3765        ];
3766
3767        emit_partitioned_cancel_rejections(&orders, &emitter, clock, "broadcast failed");
3768
3769        for order in &orders {
3770            match recv_order_event(&mut rx) {
3771                OrderEventAny::CancelRejected(event) => {
3772                    assert_eq!(event.client_order_id, order.client_order_id);
3773                    assert_eq!(event.instrument_id, order.instrument_id);
3774                    assert_eq!(event.venue_order_id, order.venue_order_id);
3775                    assert_eq!(event.reason, "broadcast failed");
3776                }
3777                event => panic!("Expected order cancel rejected event, was {event:?}"),
3778            }
3779        }
3780        assert!(rx.try_recv().is_err());
3781    }
3782
3783    #[rstest]
3784    fn test_find_matching_order_report_returns_later_match() {
3785        // Regression guard: earlier implementation fetched limit=1 and returned None
3786        // when the first order didn't match the filter. The fixed iteration logic must
3787        // scan the whole response and return the matching entry.
3788        let btc_inst = test_instrument("BTC-USD-PERP", "DYDX");
3789        let eth_inst = test_instrument("ETH-USD-PERP", "DYDX");
3790
3791        let orders = vec![
3792            test_order("order-eth", 1, "22222"),
3793            test_order("order-btc", 0, "11111"),
3794        ];
3795
3796        let encoder = ClientOrderIdEncoder::new();
3797        let report = find_matching_order_report(
3798            &orders,
3799            Some(btc_inst.id()),
3800            None,
3801            None,
3802            |clob_pair_id| match clob_pair_id {
3803                0 => Some(btc_inst.clone()),
3804                1 => Some(eth_inst.clone()),
3805                _ => None,
3806            },
3807            &encoder,
3808            AccountId::new("DYDX-001"),
3809            UnixNanos::default(),
3810        )
3811        .expect("lookup should succeed");
3812
3813        let report = report.expect("matching order should be found");
3814        assert_eq!(report.instrument_id, btc_inst.id());
3815        assert_eq!(report.venue_order_id.as_str(), "order-btc");
3816    }
3817
3818    #[rstest]
3819    fn test_find_matching_order_report_returns_none_when_no_match() {
3820        let btc_inst = test_instrument("BTC-USD-PERP", "DYDX");
3821        let eth_inst = test_instrument("ETH-USD-PERP", "DYDX");
3822
3823        let orders = vec![
3824            test_order("order-eth-1", 1, "22222"),
3825            test_order("order-eth-2", 1, "33333"),
3826        ];
3827
3828        let encoder = ClientOrderIdEncoder::new();
3829        let report = find_matching_order_report(
3830            &orders,
3831            Some(btc_inst.id()),
3832            None,
3833            None,
3834            |clob_pair_id| match clob_pair_id {
3835                0 => Some(btc_inst.clone()),
3836                1 => Some(eth_inst.clone()),
3837                _ => None,
3838            },
3839            &encoder,
3840            AccountId::new("DYDX-001"),
3841            UnixNanos::default(),
3842        )
3843        .expect("lookup should succeed");
3844
3845        assert!(report.is_none());
3846    }
3847
3848    #[rstest]
3849    fn test_find_matching_order_report_filters_by_venue_order_id() {
3850        let btc_inst = test_instrument("BTC-USD-PERP", "DYDX");
3851
3852        let orders = vec![
3853            test_order("order-a", 0, "11111"),
3854            test_order("order-b", 0, "22222"),
3855            test_order("order-c", 0, "33333"),
3856        ];
3857
3858        let encoder = ClientOrderIdEncoder::new();
3859        let target = VenueOrderId::new("order-b");
3860        let report = find_matching_order_report(
3861            &orders,
3862            None,
3863            None,
3864            Some(target),
3865            |_| Some(btc_inst.clone()),
3866            &encoder,
3867            AccountId::new("DYDX-001"),
3868            UnixNanos::default(),
3869        )
3870        .expect("lookup should succeed")
3871        .expect("matching order should be found");
3872
3873        assert_eq!(report.venue_order_id.as_str(), "order-b");
3874    }
3875
3876    #[rstest]
3877    fn test_find_matching_order_report_skips_orders_without_cached_instrument() {
3878        let btc_inst = test_instrument("BTC-USD-PERP", "DYDX");
3879
3880        let orders = vec![
3881            test_order("order-unknown", 99, "11111"),
3882            test_order("order-btc", 0, "22222"),
3883        ];
3884
3885        let encoder = ClientOrderIdEncoder::new();
3886        let report = find_matching_order_report(
3887            &orders,
3888            Some(btc_inst.id()),
3889            None,
3890            None,
3891            |clob_pair_id| (clob_pair_id == 0).then(|| btc_inst.clone()),
3892            &encoder,
3893            AccountId::new("DYDX-001"),
3894            UnixNanos::default(),
3895        )
3896        .expect("lookup should succeed")
3897        .expect("matching order should be found");
3898
3899        assert_eq!(report.venue_order_id.as_str(), "order-btc");
3900    }
3901}
nautilus_dydx/execution/mod.rs

nautilus_dydx/execution/
mod.rs
