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

nautilus_dydx/execution/
mod.rs
