nautilus_kraken/execution/
futures.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//! Kraken Futures execution client implementation.
17
18use std::{
19    future::Future,
20    sync::{Arc, Mutex},
21    time::{Duration, Instant},
22};
23
24use anyhow::Context;
25use async_trait::async_trait;
26use chrono::{DateTime, Utc};
27use nautilus_common::{
28    cache::InstrumentLookupError,
29    clients::ExecutionClient,
30    live::{get_runtime, runner::get_exec_event_sender},
31    messages::execution::{
32        BatchCancelOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
33        GenerateOrderStatusReport, GenerateOrderStatusReports, GeneratePositionStatusReports,
34        ModifyOrder, QueryAccount, QueryOrder, SubmitOrder, SubmitOrderList,
35    },
36};
37use nautilus_core::{
38    AtomicMap, MUTEX_POISONED, UnixNanos,
39    time::{AtomicTime, get_atomic_clock_realtime},
40};
41use nautilus_live::{ExecutionClientCore, ExecutionEventEmitter};
42use nautilus_model::{
43    accounts::AccountAny,
44    enums::{AccountType, OmsType, OrderSide, OrderStatus, OrderType},
45    identifiers::{
46        AccountId, ClientId, ClientOrderId, InstrumentId, StrategyId, Venue, VenueOrderId,
47    },
48    instruments::{Instrument, InstrumentAny},
49    orders::{Order, OrderAny},
50    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
51    types::{AccountBalance, MarginBalance, Quantity},
52};
53use rust_decimal::Decimal;
54use tokio::task::JoinHandle;
55use tokio_util::sync::CancellationToken;
56
57use super::{CancelCommandFailure, classify_cancel_http_failure};
58use crate::{
59    common::{
60        consts::KRAKEN_VENUE,
61        credential::KrakenCredential,
62        enums::{KrakenApiResult, KrakenSendStatus},
63        parse::truncate_cl_ord_id,
64    },
65    config::KrakenExecClientConfig,
66    http::{
67        KrakenFuturesHttpClient,
68        futures::{
69            client::KRAKEN_FUTURES_DEFAULT_RATE_LIMIT_PER_SECOND, models::FuturesBatchCancelStatus,
70            query::KrakenFuturesBatchCancelItem,
71        },
72    },
73    websocket::{
74        dispatch::{self, OrderIdentity, WsDispatchState},
75        futures::{client::KrakenFuturesWebSocketClient, messages::KrakenFuturesWsMessage},
76    },
77};
78
79const FUTURES_BATCH_CANCEL_LIMIT: usize = 50;
80
81/// Kraken Futures execution client.
82///
83/// Provides order management, account operations, and position management
84/// for Kraken Futures markets.
85#[allow(dead_code)]
86#[derive(Debug)]
87pub struct KrakenFuturesExecutionClient {
88    core: ExecutionClientCore,
89    clock: &'static AtomicTime,
90    config: KrakenExecClientConfig,
91    emitter: ExecutionEventEmitter,
92    http: KrakenFuturesHttpClient,
93    ws: KrakenFuturesWebSocketClient,
94    cancellation_token: CancellationToken,
95    ws_stream_handle: Option<JoinHandle<()>>,
96    pending_tasks: Mutex<Vec<JoinHandle<()>>>,
97    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
98    truncated_id_map: Arc<AtomicMap<String, ClientOrderId>>,
99    order_instrument_map: Arc<AtomicMap<String, InstrumentId>>,
100    venue_client_map: Arc<AtomicMap<String, ClientOrderId>>,
101    venue_order_qty: Arc<AtomicMap<String, Quantity>>,
102    ws_dispatch_state: Arc<WsDispatchState>,
103}
104
105impl KrakenFuturesExecutionClient {
106    /// Creates a new [`KrakenFuturesExecutionClient`].
107    pub fn new(core: ExecutionClientCore, config: KrakenExecClientConfig) -> anyhow::Result<Self> {
108        let clock = get_atomic_clock_realtime();
109        let emitter = ExecutionEventEmitter::new(
110            clock,
111            core.trader_id,
112            core.account_id,
113            AccountType::Margin,
114            None,
115        );
116
117        let cancellation_token = CancellationToken::new();
118
119        let http = KrakenFuturesHttpClient::with_credentials(
120            config.api_key.clone(),
121            config.api_secret.clone(),
122            config.environment,
123            config.base_url.clone(),
124            config.timeout_secs,
125            None,
126            None,
127            None,
128            config.proxy_url.clone(),
129            config
130                .max_requests_per_second
131                .unwrap_or(KRAKEN_FUTURES_DEFAULT_RATE_LIMIT_PER_SECOND),
132        )?;
133
134        let credential = KrakenCredential::new(config.api_key.clone(), config.api_secret.clone());
135        let ws = KrakenFuturesWebSocketClient::with_credentials(
136            config.ws_url(),
137            config.heartbeat_interval_secs,
138            Some(credential),
139            config.transport_backend,
140            config.proxy_url.clone(),
141        );
142
143        Ok(Self {
144            core,
145            clock,
146            config,
147            emitter,
148            http,
149            ws,
150            cancellation_token,
151            ws_stream_handle: None,
152            pending_tasks: Mutex::new(Vec::new()),
153            instruments: Arc::new(AtomicMap::new()),
154            truncated_id_map: Arc::new(AtomicMap::new()),
155            order_instrument_map: Arc::new(AtomicMap::new()),
156            venue_client_map: Arc::new(AtomicMap::new()),
157            venue_order_qty: Arc::new(AtomicMap::new()),
158            ws_dispatch_state: Arc::new(WsDispatchState::new()),
159        })
160    }
161
162    fn register_order_identity(&self, order: &OrderAny) {
163        self.ws_dispatch_state.register_identity(
164            order.client_order_id(),
165            OrderIdentity {
166                strategy_id: order.strategy_id(),
167                instrument_id: order.instrument_id(),
168                order_side: order.order_side(),
169                order_type: order.order_type(),
170                quantity: order.quantity(),
171            },
172        );
173    }
174
175    /// Returns a reference to the clock.
176    #[must_use]
177    pub fn clock(&self) -> &'static AtomicTime {
178        self.clock
179    }
180
181    /// Returns a reference to the event emitter.
182    #[must_use]
183    pub fn emitter(&self) -> &ExecutionEventEmitter {
184        &self.emitter
185    }
186
187    fn spawn_task<F>(&self, description: &'static str, fut: F)
188    where
189        F: Future<Output = anyhow::Result<()>> + Send + 'static,
190    {
191        let runtime = get_runtime();
192        let handle = runtime.spawn(async move {
193            if let Err(e) = fut.await {
194                log::warn!("{description} failed: {e:?}");
195            }
196        });
197
198        let mut tasks = self.pending_tasks.lock().expect(MUTEX_POISONED);
199        tasks.retain(|handle| !handle.is_finished());
200        tasks.push(handle);
201    }
202
203    fn submit_single_order(&self, order: &OrderAny, task_name: &'static str) {
204        if order.is_closed() {
205            log::warn!(
206                "Cannot submit closed order: client_order_id={}",
207                order.client_order_id()
208            );
209            return;
210        }
211
212        let account_id = self.core.account_id;
213        let client_order_id = order.client_order_id();
214        let strategy_id = order.strategy_id();
215        let instrument_id = order.instrument_id();
216        let order_side = order.order_side();
217        let order_type = order.order_type();
218        let quantity = order.quantity();
219        let time_in_force = order.time_in_force();
220        let price = order.price();
221        let trigger_price = order.trigger_price();
222        let trigger_type = order.trigger_type();
223        let is_reduce_only = order.is_reduce_only();
224        let is_post_only = order.is_post_only();
225
226        log::debug!("OrderSubmitted: client_order_id={client_order_id}");
227        self.register_order_identity(order);
228        self.emitter.emit_order_submitted(order);
229
230        let kraken_cl_ord_id = truncate_cl_ord_id(&client_order_id);
231
232        if kraken_cl_ord_id != client_order_id.as_str() {
233            self.truncated_id_map
234                .insert(kraken_cl_ord_id, client_order_id);
235        }
236
237        let http = self.http.clone();
238        let emitter = self.emitter.clone();
239        let clock = self.clock;
240        let dispatch_state = self.ws_dispatch_state.clone();
241
242        self.spawn_task(task_name, async move {
243            let result = http
244                .submit_order(
245                    account_id,
246                    instrument_id,
247                    client_order_id,
248                    order_side,
249                    order_type,
250                    quantity,
251                    time_in_force,
252                    price,
253                    trigger_price,
254                    trigger_type,
255                    is_reduce_only,
256                    is_post_only,
257                )
258                .await;
259
260            match result {
261                Ok(_report) => Ok(()),
262                Err(e) => {
263                    let ts_event = clock.get_time_ns();
264                    let error_msg = format!("{task_name} error: {e}");
265                    let due_post_only = error_msg.contains("POST_ONLY_REJECTED");
266                    // The order will never appear on the wire, so its
267                    // dispatch identity has to be cleaned up here.
268                    dispatch_state.cleanup_terminal(&client_order_id);
269                    emitter.emit_order_rejected_event(
270                        strategy_id,
271                        instrument_id,
272                        client_order_id,
273                        &error_msg,
274                        ts_event,
275                        due_post_only,
276                    );
277                    Ok(())
278                }
279            }
280        });
281    }
282
283    fn cancel_single_order(&self, cmd: &CancelOrder) {
284        let account_id = self.core.account_id;
285        let client_order_id = cmd.client_order_id;
286        let venue_order_id = cmd.venue_order_id;
287        let strategy_id = cmd.strategy_id;
288        let instrument_id = cmd.instrument_id;
289
290        log::debug!(
291            "Canceling order: venue_order_id={venue_order_id:?}, client_order_id={client_order_id}"
292        );
293
294        let http = self.http.clone();
295        let emitter = self.emitter.clone();
296        let clock = self.clock;
297
298        self.spawn_task("cancel_order", async move {
299            if let Err(failure) = cancel_order_for_futures(
300                &http,
301                account_id,
302                instrument_id,
303                Some(client_order_id),
304                venue_order_id,
305            )
306            .await
307            {
308                handle_cancel_failure(
309                    &emitter,
310                    clock,
311                    strategy_id,
312                    instrument_id,
313                    client_order_id,
314                    venue_order_id,
315                    failure,
316                );
317            }
318            Ok(())
319        });
320    }
321
322    fn spawn_message_handler(&mut self) -> anyhow::Result<()> {
323        let mut rx = self
324            .ws
325            .take_output_rx()
326            .context("Failed to take futures WebSocket output receiver")?;
327        let emitter = self.emitter.clone();
328        let instruments = self.instruments.clone();
329        let truncated_id_map = self.truncated_id_map.clone();
330        let order_instrument_map = self.order_instrument_map.clone();
331        let venue_client_map = self.venue_client_map.clone();
332        let venue_order_qty = self.venue_order_qty.clone();
333        let dispatch_state = self.ws_dispatch_state.clone();
334        let account_id = self.core.account_id;
335        let clock = self.clock;
336        let cancellation_token = self.cancellation_token.clone();
337
338        let handle = get_runtime().spawn(async move {
339            loop {
340                tokio::select! {
341                    () = cancellation_token.cancelled() => {
342                        log::debug!("Futures execution message handler cancelled");
343                        break;
344                    }
345                    msg = rx.recv() => {
346                        match msg {
347                            Some(ws_msg) => {
348                                Self::handle_ws_message(
349                                    ws_msg,
350                                    &emitter,
351                                    &dispatch_state,
352                                    &instruments,
353                                    &truncated_id_map,
354                                    &order_instrument_map,
355                                    &venue_client_map,
356                                    &venue_order_qty,
357                                    account_id,
358                                    clock,
359                                );
360                            }
361                            None => {
362                                log::debug!("Futures execution WebSocket stream ended");
363                                break;
364                            }
365                        }
366                    }
367                }
368            }
369        });
370
371        self.ws_stream_handle = Some(handle);
372        Ok(())
373    }
374
375    #[expect(clippy::too_many_arguments)]
376    fn handle_ws_message(
377        msg: KrakenFuturesWsMessage,
378        emitter: &ExecutionEventEmitter,
379        dispatch_state: &Arc<WsDispatchState>,
380        instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
381        truncated_id_map: &Arc<AtomicMap<String, ClientOrderId>>,
382        order_instrument_map: &Arc<AtomicMap<String, InstrumentId>>,
383        venue_client_map: &Arc<AtomicMap<String, ClientOrderId>>,
384        venue_order_qty: &Arc<AtomicMap<String, Quantity>>,
385        account_id: AccountId,
386        clock: &'static AtomicTime,
387    ) {
388        let ts_init = clock.get_time_ns();
389
390        match msg {
391            KrakenFuturesWsMessage::OpenOrdersDelta(delta) => {
392                dispatch::futures::open_orders_delta(
393                    &delta,
394                    dispatch_state,
395                    emitter,
396                    instruments,
397                    truncated_id_map,
398                    order_instrument_map,
399                    venue_client_map,
400                    venue_order_qty,
401                    account_id,
402                    ts_init,
403                );
404            }
405            KrakenFuturesWsMessage::OpenOrdersCancel(cancel) => {
406                dispatch::futures::open_orders_cancel(
407                    &cancel,
408                    dispatch_state,
409                    emitter,
410                    truncated_id_map,
411                    order_instrument_map,
412                    venue_client_map,
413                    venue_order_qty,
414                    account_id,
415                    ts_init,
416                );
417            }
418            KrakenFuturesWsMessage::FillsDelta(fills_delta) => {
419                dispatch::futures::fills_delta(
420                    &fills_delta,
421                    dispatch_state,
422                    emitter,
423                    instruments,
424                    truncated_id_map,
425                    venue_client_map,
426                    account_id,
427                    ts_init,
428                );
429            }
430            KrakenFuturesWsMessage::Challenge(challenge) => {
431                log::debug!("Received challenge: length={}", challenge.len());
432            }
433            KrakenFuturesWsMessage::Reconnected => {
434                log::info!("Futures execution WebSocket reconnected");
435            }
436            KrakenFuturesWsMessage::Ticker(_)
437            | KrakenFuturesWsMessage::Trade(_)
438            | KrakenFuturesWsMessage::BookSnapshot(_)
439            | KrakenFuturesWsMessage::BookDelta(_) => {}
440        }
441    }
442
443    async fn await_account_registered(&self, timeout_secs: f64) -> anyhow::Result<()> {
444        let account_id = self.core.account_id;
445
446        if self.core.cache().account(&account_id).is_some() {
447            log::info!("Account {account_id} registered");
448            return Ok(());
449        }
450
451        let start = Instant::now();
452        let timeout = Duration::from_secs_f64(timeout_secs);
453        let interval = Duration::from_millis(10);
454
455        loop {
456            tokio::time::sleep(interval).await;
457
458            if self.core.cache().account(&account_id).is_some() {
459                log::info!("Account {account_id} registered");
460                return Ok(());
461            }
462
463            if start.elapsed() >= timeout {
464                anyhow::bail!(
465                    "Timeout waiting for account {account_id} to be registered after {timeout_secs}s"
466                );
467            }
468        }
469    }
470
471    fn modify_single_order(&self, cmd: &ModifyOrder) {
472        let client_order_id = cmd.client_order_id;
473        let venue_order_id = cmd.venue_order_id;
474        let strategy_id = cmd.strategy_id;
475        let instrument_id = cmd.instrument_id;
476        let quantity = cmd.quantity;
477        let price = cmd.price;
478
479        log::debug!(
480            "Modifying order: venue_order_id={venue_order_id:?}, client_order_id={client_order_id}"
481        );
482
483        let http = self.http.clone();
484        let emitter = self.emitter.clone();
485        let clock = self.clock;
486
487        self.spawn_task("modify_order", async move {
488            if let Err(e) = http
489                .modify_order(
490                    instrument_id,
491                    Some(client_order_id),
492                    venue_order_id,
493                    quantity,
494                    price,
495                    None,
496                )
497                .await
498            {
499                let ts_event = clock.get_time_ns();
500                emitter.emit_order_modify_rejected_event(
501                    strategy_id,
502                    instrument_id,
503                    client_order_id,
504                    venue_order_id,
505                    &format!("modify-order error: {e}"),
506                    ts_event,
507                );
508                anyhow::bail!("Modify order failed: {e}");
509            }
510            Ok(())
511        });
512    }
513}
514
515#[async_trait(?Send)]
516impl ExecutionClient for KrakenFuturesExecutionClient {
517    fn is_connected(&self) -> bool {
518        self.core.is_connected()
519    }
520
521    fn client_id(&self) -> ClientId {
522        self.core.client_id
523    }
524
525    fn account_id(&self) -> AccountId {
526        self.core.account_id
527    }
528
529    fn venue(&self) -> Venue {
530        *KRAKEN_VENUE
531    }
532
533    fn oms_type(&self) -> OmsType {
534        self.core.oms_type
535    }
536
537    fn get_account(&self) -> Option<AccountAny> {
538        self.core.cache().account_owned(&self.core.account_id)
539    }
540
541    fn generate_account_state(
542        &self,
543        balances: Vec<AccountBalance>,
544        margins: Vec<MarginBalance>,
545        reported: bool,
546        ts_event: UnixNanos,
547    ) -> anyhow::Result<()> {
548        self.emitter
549            .emit_account_state(balances, margins, reported, ts_event);
550        Ok(())
551    }
552
553    fn start(&mut self) -> anyhow::Result<()> {
554        if self.core.is_started() {
555            return Ok(());
556        }
557
558        self.emitter.set_sender(get_exec_event_sender());
559        self.core.set_started();
560
561        log::info!(
562            "Started: client_id={}, account_id={}, product_type=Futures, environment={:?}",
563            self.core.client_id,
564            self.core.account_id,
565            self.config.environment
566        );
567        Ok(())
568    }
569
570    fn stop(&mut self) -> anyhow::Result<()> {
571        if self.core.is_stopped() {
572            return Ok(());
573        }
574
575        self.cancellation_token.cancel();
576        self.core.set_stopped();
577        self.core.set_disconnected();
578        log::info!("Stopped: client_id={}", self.core.client_id);
579        Ok(())
580    }
581
582    async fn connect(&mut self) -> anyhow::Result<()> {
583        if self.core.is_connected() {
584            return Ok(());
585        }
586
587        if !self.core.instruments_initialized() {
588            let instruments = self
589                .http
590                .request_instruments()
591                .await
592                .context("Failed to load Kraken futures instruments")?;
593            log::debug!("Loaded {} Futures instruments", instruments.len());
594            self.http.cache_instruments(&instruments);
595            self.core.set_instruments_initialized();
596        }
597
598        self.instruments.rcu(|m| {
599            for instrument in self.http.instruments_cache.load().values() {
600                m.insert(instrument.id(), instrument.clone());
601            }
602        });
603
604        self.ws
605            .connect()
606            .await
607            .context("Failed to connect futures WebSocket")?;
608        self.ws
609            .wait_until_active(10.0)
610            .await
611            .context("Futures WebSocket failed to become active")?;
612
613        self.ws
614            .authenticate()
615            .await
616            .context("Failed to authenticate futures WebSocket")?;
617
618        // Request and register account state before message handler
619        let account_state = self
620            .http
621            .request_account_state(self.core.account_id)
622            .await
623            .context("Failed to request Kraken futures account state")?;
624
625        if !account_state.balances.is_empty() {
626            log::debug!(
627                "Received account state with {} balance(s)",
628                account_state.balances.len()
629            );
630        }
631        self.emitter.send_account_state(account_state);
632        self.await_account_registered(30.0).await?;
633
634        self.spawn_message_handler()?;
635
636        self.ws
637            .subscribe_executions()
638            .await
639            .context("Failed to subscribe to executions")?;
640
641        log::debug!("Futures WebSocket authenticated and subscribed to executions");
642
643        self.core.set_connected();
644        log::info!("Connected: client_id={}", self.core.client_id);
645        Ok(())
646    }
647
648    async fn disconnect(&mut self) -> anyhow::Result<()> {
649        if self.core.is_disconnected() {
650            return Ok(());
651        }
652
653        self.cancellation_token.cancel();
654
655        if let Some(handle) = self.ws_stream_handle.take() {
656            handle.abort();
657        }
658
659        let _ = self.ws.close().await;
660
661        self.cancellation_token = CancellationToken::new();
662        self.core.set_disconnected();
663        log::info!("Disconnected: client_id={}", self.core.client_id);
664        Ok(())
665    }
666
667    async fn generate_order_status_report(
668        &self,
669        cmd: &GenerateOrderStatusReport,
670    ) -> anyhow::Result<Option<OrderStatusReport>> {
671        log::debug!(
672            "Generating order status report: venue_order_id={:?}, client_order_id={:?}",
673            cmd.venue_order_id,
674            cmd.client_order_id
675        );
676
677        let account_id = self.core.account_id;
678        let reports = self
679            .http
680            .request_order_status_reports(account_id, None, None, None, false)
681            .await?;
682
683        // Match by venue_order_id or client_order_id (comparing truncated form
684        // since Kraken stores the truncated cl_ord_id for long IDs)
685        let matched = reports.into_iter().find(|r| {
686            cmd.venue_order_id
687                .is_some_and(|id| r.venue_order_id.as_str() == id.as_str())
688                || cmd.client_order_id.is_some_and(|id| {
689                    r.client_order_id
690                        .as_ref()
691                        .is_some_and(|r_id| r_id.as_str() == truncate_cl_ord_id(&id))
692                })
693        });
694
695        if matched.is_some() {
696            return Ok(matched);
697        }
698
699        let Some(order) = self.get_cached_order_for_status_command(cmd) else {
700            return Ok(None);
701        };
702
703        let now = Utc::now();
704        let start = now - Duration::from_secs(5 * 60);
705        let fills = self
706            .http
707            .request_fill_reports(
708                account_id,
709                Some(order.instrument_id()),
710                Some(start),
711                Some(now),
712            )
713            .await?;
714
715        Ok(synthesize_filled_order_status_report(cmd, &order, &fills))
716    }
717
718    async fn generate_order_status_reports(
719        &self,
720        cmd: &GenerateOrderStatusReports,
721    ) -> anyhow::Result<Vec<OrderStatusReport>> {
722        log::debug!(
723            "Generating order status reports: instrument_id={:?}, open_only={}",
724            cmd.instrument_id,
725            cmd.open_only
726        );
727
728        let account_id = self.core.account_id;
729        let start = cmd.start.map(DateTime::<Utc>::from);
730        let end = cmd.end.map(DateTime::<Utc>::from);
731        self.http
732            .request_order_status_reports(account_id, cmd.instrument_id, start, end, cmd.open_only)
733            .await
734    }
735
736    async fn generate_fill_reports(
737        &self,
738        cmd: GenerateFillReports,
739    ) -> anyhow::Result<Vec<FillReport>> {
740        log::debug!(
741            "Generating fill reports: instrument_id={:?}",
742            cmd.instrument_id
743        );
744
745        let account_id = self.core.account_id;
746        let start = cmd.start.map(DateTime::<Utc>::from);
747        let end = cmd.end.map(DateTime::<Utc>::from);
748        let mut reports = self
749            .http
750            .request_fill_reports(account_id, cmd.instrument_id, start, end)
751            .await?;
752
753        if let Some(venue_order_id) = cmd.venue_order_id {
754            reports.retain(|report| report.venue_order_id == venue_order_id);
755        }
756
757        Ok(reports)
758    }
759
760    async fn generate_position_status_reports(
761        &self,
762        cmd: &GeneratePositionStatusReports,
763    ) -> anyhow::Result<Vec<PositionStatusReport>> {
764        log::debug!(
765            "Generating position status reports: instrument_id={:?}",
766            cmd.instrument_id
767        );
768
769        let account_id = self.core.account_id;
770        self.http
771            .request_position_status_reports(account_id, cmd.instrument_id)
772            .await
773    }
774
775    async fn generate_mass_status(
776        &self,
777        lookback_mins: Option<u64>,
778    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
779        log::debug!("Generating mass status: lookback_mins={lookback_mins:?}");
780
781        let start = lookback_mins.map(|mins| Utc::now() - Duration::from_secs(mins * 60));
782
783        let account_id = self.core.account_id;
784        let order_reports = self
785            .http
786            .request_order_status_reports(account_id, None, start, None, true)
787            .await?;
788        let fill_reports = self
789            .http
790            .request_fill_reports(account_id, None, start, None)
791            .await?;
792        let position_reports = self
793            .http
794            .request_position_status_reports(account_id, None)
795            .await?;
796
797        let mut mass_status = ExecutionMassStatus::new(
798            self.core.client_id,
799            self.core.account_id,
800            *KRAKEN_VENUE,
801            self.clock.get_time_ns(),
802            None,
803        );
804        mass_status.add_order_reports(order_reports);
805        mass_status.add_fill_reports(fill_reports);
806        mass_status.add_position_reports(position_reports);
807
808        Ok(Some(mass_status))
809    }
810
811    fn query_account(&self, cmd: QueryAccount) -> anyhow::Result<()> {
812        log::debug!("Querying account: {cmd:?}");
813
814        let account_id = self.core.account_id;
815        let http = self.http.clone();
816        let emitter = self.emitter.clone();
817
818        self.spawn_task("query_account", async move {
819            let account_state = http.request_account_state(account_id).await?;
820            emitter.emit_account_state(
821                account_state.balances.clone(),
822                account_state.margins.clone(),
823                account_state.is_reported,
824                account_state.ts_event,
825            );
826            Ok(())
827        });
828
829        Ok(())
830    }
831
832    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
833        log::debug!("Querying order: {cmd:?}");
834
835        let venue_order_id = cmd
836            .venue_order_id
837            .context("venue_order_id required for query_order")?;
838        let account_id = self.core.account_id;
839        let http = self.http.clone();
840        let emitter = self.emitter.clone();
841
842        self.spawn_task("query_order", async move {
843            let reports = http
844                .request_order_status_reports(account_id, None, None, None, true)
845                .await
846                .context("Failed to query order")?;
847
848            if let Some(report) = reports
849                .into_iter()
850                .find(|r| r.venue_order_id == venue_order_id)
851            {
852                emitter.send_order_status_report(report);
853            }
854            Ok(())
855        });
856
857        Ok(())
858    }
859
860    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
861        let order = self.core.cache().try_order_owned(&cmd.client_order_id)?;
862        self.submit_single_order(&order, "submit_order");
863        Ok(())
864    }
865
866    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
867        let orders = self.core.get_orders_for_list(&cmd.order_list)?;
868
869        log::debug!(
870            "Submitting order list: order_list_id={}, count={}",
871            cmd.order_list.id,
872            orders.len()
873        );
874
875        let mut order_tuples = Vec::with_capacity(orders.len());
876        let mut order_meta = Vec::with_capacity(orders.len());
877
878        for order in &orders {
879            if order.is_closed() {
880                log::warn!(
881                    "Cannot submit closed order: client_order_id={}",
882                    order.client_order_id()
883                );
884                continue;
885            }
886
887            // Kraken batch endpoint only supports limit and stop orders,
888            // submit market orders individually
889            if order.order_type() == OrderType::Market {
890                self.submit_single_order(order, "submit_order_list");
891                continue;
892            }
893
894            let client_order_id = order.client_order_id();
895            let kraken_cl_ord_id = truncate_cl_ord_id(&client_order_id);
896
897            if kraken_cl_ord_id != client_order_id.as_str() {
898                self.truncated_id_map
899                    .insert(kraken_cl_ord_id, client_order_id);
900            }
901
902            self.register_order_identity(order);
903            self.emitter.emit_order_submitted(order);
904
905            order_tuples.push((
906                order.instrument_id(),
907                client_order_id,
908                order.order_side(),
909                order.order_type(),
910                order.quantity(),
911                order.time_in_force(),
912                order.price(),
913                order.trigger_price(),
914                order.trigger_type(),
915                order.is_reduce_only(),
916                order.is_post_only(),
917            ));
918
919            order_meta.push((order.strategy_id(), order.instrument_id(), client_order_id));
920        }
921
922        if order_tuples.is_empty() {
923            return Ok(());
924        }
925
926        let http = self.http.clone();
927        let emitter = self.emitter.clone();
928        let clock = self.clock;
929        let dispatch_state = self.ws_dispatch_state.clone();
930
931        self.spawn_task("submit_order_list", async move {
932            match http.submit_orders_batch(order_tuples).await {
933                Ok(statuses) => {
934                    for (i, status) in statuses.iter().enumerate() {
935                        if status.status != "placed"
936                            && status.status != "filled"
937                            && let Some((strategy_id, instrument_id, client_order_id)) =
938                                order_meta.get(i)
939                        {
940                            let ts_event = clock.get_time_ns();
941                            let error_msg = format!(
942                                "submit_order_list batch item rejected: {}",
943                                status.status,
944                            );
945                            dispatch_state.cleanup_terminal(client_order_id);
946                            emitter.emit_order_rejected_event(
947                                *strategy_id,
948                                *instrument_id,
949                                *client_order_id,
950                                &error_msg,
951                                ts_event,
952                                status.status == "postWouldExecute",
953                            );
954                        }
955                    }
956                    Ok(())
957                }
958                Err(e) => {
959                    let ts_event = clock.get_time_ns();
960
961                    for (strategy_id, instrument_id, client_order_id) in &order_meta {
962                        let error_msg = format!("submit_order_list batch error: {e}");
963                        dispatch_state.cleanup_terminal(client_order_id);
964                        emitter.emit_order_rejected_event(
965                            *strategy_id,
966                            *instrument_id,
967                            *client_order_id,
968                            &error_msg,
969                            ts_event,
970                            false,
971                        );
972                    }
973                    Ok(())
974                }
975            }
976        });
977
978        Ok(())
979    }
980
981    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
982        self.modify_single_order(&cmd);
983        Ok(())
984    }
985
986    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
987        self.cancel_single_order(&cmd);
988        Ok(())
989    }
990
991    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
992        let instrument_id = cmd.instrument_id;
993
994        if cmd.order_side == OrderSide::NoOrderSide {
995            log::debug!("Canceling all orders: instrument_id={instrument_id} (bulk)");
996
997            let http = self.http.clone();
998            let symbol = instrument_id.symbol.to_string();
999
1000            self.spawn_task("cancel_all_orders", async move {
1001                match http.inner.cancel_all_orders(Some(symbol)).await {
1002                    Ok(response) => {
1003                        if response.result != KrakenApiResult::Success
1004                            && response.cancel_status.cancelled_orders.is_empty()
1005                        {
1006                            log::warn!(
1007                                "Cancel-all failed without per-order results, awaiting reconciliation: status={}",
1008                                response.cancel_status.status
1009                            );
1010                        }
1011                    }
1012                    Err(e) => match classify_cancel_http_failure(e) {
1013                        CancelCommandFailure::LocalValidation(reason) => {
1014                            log::warn!("Cancel-all failed local validation: {reason}");
1015                        }
1016                        CancelCommandFailure::Ambiguous(reason)
1017                        | CancelCommandFailure::VenueReject(reason) => {
1018                            log::warn!(
1019                                "Cancel-all ambiguous failure, awaiting reconciliation: {reason}"
1020                            );
1021                        }
1022                    },
1023                }
1024                Ok(())
1025            });
1026
1027            return Ok(());
1028        }
1029
1030        log::debug!(
1031            "Canceling all orders: instrument_id={instrument_id}, side={:?}",
1032            cmd.order_side
1033        );
1034
1035        let orders_to_cancel: Vec<_> = {
1036            let cache = self.core.cache();
1037            let open_orders = cache.orders_open(None, Some(&instrument_id), None, None, None);
1038
1039            open_orders
1040                .into_iter()
1041                .filter(|order| order.order_side() == cmd.order_side)
1042                .filter_map(|order| {
1043                    Some((
1044                        order.venue_order_id()?,
1045                        order.client_order_id(),
1046                        order.instrument_id(),
1047                        order.strategy_id(),
1048                    ))
1049                })
1050                .collect()
1051        };
1052
1053        let account_id = self.core.account_id;
1054
1055        for (venue_order_id, client_order_id, order_instrument_id, strategy_id) in orders_to_cancel
1056        {
1057            let http = self.http.clone();
1058            let emitter = self.emitter.clone();
1059            let clock = self.clock;
1060
1061            self.spawn_task("cancel_order_by_side", async move {
1062                if let Err(failure) = cancel_order_for_futures(
1063                    &http,
1064                    account_id,
1065                    order_instrument_id,
1066                    Some(client_order_id),
1067                    Some(venue_order_id),
1068                )
1069                .await
1070                {
1071                    handle_cancel_failure(
1072                        &emitter,
1073                        clock,
1074                        strategy_id,
1075                        order_instrument_id,
1076                        client_order_id,
1077                        Some(venue_order_id),
1078                        failure,
1079                    );
1080                }
1081                Ok(())
1082            });
1083        }
1084
1085        Ok(())
1086    }
1087
1088    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
1089        log::debug!(
1090            "Batch canceling orders: instrument_id={}, count={}",
1091            cmd.instrument_id,
1092            cmd.cancels.len()
1093        );
1094
1095        let http = self.http.clone();
1096        let emitter = self.emitter.clone();
1097        let clock = self.clock;
1098        let cancels = cmd.cancels;
1099
1100        self.spawn_task("batch_cancel_orders", async move {
1101            batch_cancel_orders_for_futures(&http, &emitter, clock, &cancels).await;
1102            Ok(())
1103        });
1104
1105        Ok(())
1106    }
1107}
1108
1109#[derive(Debug, Clone)]
1110struct CancelRequestContext {
1111    strategy_id: StrategyId,
1112    instrument_id: InstrumentId,
1113    client_order_id: ClientOrderId,
1114    truncated_client_order_id: String,
1115    venue_order_id: Option<VenueOrderId>,
1116}
1117
1118async fn cancel_order_for_futures(
1119    http: &KrakenFuturesHttpClient,
1120    _account_id: AccountId,
1121    instrument_id: InstrumentId,
1122    client_order_id: Option<ClientOrderId>,
1123    venue_order_id: Option<VenueOrderId>,
1124) -> Result<(), CancelCommandFailure> {
1125    http.get_cached_instrument(&instrument_id.symbol.inner())
1126        .ok_or_else(|| {
1127            CancelCommandFailure::local(InstrumentLookupError::not_found(instrument_id).to_string())
1128        })?;
1129
1130    let order_id = venue_order_id.as_ref().map(ToString::to_string);
1131    let cli_ord_id = client_order_id.as_ref().map(truncate_cl_ord_id);
1132
1133    if order_id.is_none() && cli_ord_id.is_none() {
1134        return Err(CancelCommandFailure::local(
1135            "Either client_order_id or venue_order_id must be provided",
1136        ));
1137    }
1138
1139    let response = http
1140        .inner
1141        .cancel_order(order_id, cli_ord_id)
1142        .await
1143        .map_err(classify_cancel_http_failure)?;
1144
1145    if response.result != KrakenApiResult::Success
1146        || response.cancel_status.status != KrakenSendStatus::Cancelled
1147    {
1148        return Err(CancelCommandFailure::venue_reject(format!(
1149            "cancel-order rejected: status={}",
1150            response.cancel_status.status
1151        )));
1152    }
1153
1154    Ok(())
1155}
1156
1157async fn batch_cancel_orders_for_futures(
1158    http: &KrakenFuturesHttpClient,
1159    emitter: &ExecutionEventEmitter,
1160    clock: &'static AtomicTime,
1161    cancels: &[CancelOrder],
1162) {
1163    let mut contexts = Vec::new();
1164    let mut items = Vec::new();
1165
1166    for cancel in cancels {
1167        match batch_cancel_item_for_futures(http, cancel) {
1168            Ok((context, item)) => {
1169                contexts.push(context);
1170                items.push(item);
1171            }
1172            Err(CancelCommandFailure::LocalValidation(reason)) => {
1173                log::warn!(
1174                    "Batch cancel command failed local validation for {}: {reason}",
1175                    cancel.client_order_id
1176                );
1177            }
1178            Err(
1179                CancelCommandFailure::Ambiguous(reason) | CancelCommandFailure::VenueReject(reason),
1180            ) => {
1181                log::warn!(
1182                    "Batch cancel command ambiguous failure for {}, awaiting reconciliation: {reason}",
1183                    cancel.client_order_id
1184                );
1185            }
1186        }
1187    }
1188
1189    for (item_chunk, context_chunk) in items
1190        .chunks(FUTURES_BATCH_CANCEL_LIMIT)
1191        .zip(contexts.chunks(FUTURES_BATCH_CANCEL_LIMIT))
1192    {
1193        let response = match http
1194            .inner
1195            .cancel_order_items_batch(item_chunk.to_vec())
1196            .await
1197        {
1198            Ok(response) => response,
1199            Err(e) => {
1200                match classify_cancel_http_failure(e) {
1201                    CancelCommandFailure::LocalValidation(reason) => {
1202                        log::warn!("Batch cancel failed local validation: {reason}");
1203                    }
1204                    CancelCommandFailure::Ambiguous(reason)
1205                    | CancelCommandFailure::VenueReject(reason) => {
1206                        log::warn!(
1207                            "Batch cancel failed without per-order results, awaiting reconciliation: {reason}"
1208                        );
1209                    }
1210                }
1211                continue;
1212            }
1213        };
1214
1215        if response.batch_status.is_empty() {
1216            if response.result != KrakenApiResult::Success {
1217                let reason = response.error.as_deref().unwrap_or("Unknown error");
1218                log::warn!(
1219                    "Batch cancel failed without per-order results, awaiting reconciliation: {reason}"
1220                );
1221            }
1222            continue;
1223        }
1224
1225        if response.batch_status.len() != context_chunk.len() {
1226            log::warn!(
1227                "Batch cancel returned {} per-order result(s) for {} request(s); unmatched results await reconciliation",
1228                response.batch_status.len(),
1229                context_chunk.len()
1230            );
1231        }
1232
1233        for (index, status) in response.batch_status.iter().enumerate() {
1234            let Some(cancel_status) = batch_cancel_status(status) else {
1235                log::warn!("Batch cancel result without status at index {index}");
1236                continue;
1237            };
1238
1239            if cancel_status == KrakenSendStatus::Cancelled {
1240                continue;
1241            }
1242
1243            let Some(context) = batch_cancel_context(status, context_chunk, index) else {
1244                log::warn!(
1245                    "Batch cancel rejected item without matching request context at index {index}: status={cancel_status}"
1246                );
1247                continue;
1248            };
1249
1250            emitter.emit_order_cancel_rejected_event(
1251                context.strategy_id,
1252                context.instrument_id,
1253                context.client_order_id,
1254                context.venue_order_id,
1255                &format!("batch-cancel rejected: status={cancel_status}"),
1256                clock.get_time_ns(),
1257            );
1258        }
1259    }
1260}
1261
1262fn batch_cancel_item_for_futures(
1263    http: &KrakenFuturesHttpClient,
1264    cancel: &CancelOrder,
1265) -> Result<(CancelRequestContext, KrakenFuturesBatchCancelItem), CancelCommandFailure> {
1266    http.get_cached_instrument(&cancel.instrument_id.symbol.inner())
1267        .ok_or_else(|| {
1268            CancelCommandFailure::local(
1269                InstrumentLookupError::not_found(cancel.instrument_id).to_string(),
1270            )
1271        })?;
1272
1273    let truncated_client_order_id = truncate_cl_ord_id(&cancel.client_order_id);
1274    let item = if let Some(venue_order_id) = cancel.venue_order_id {
1275        KrakenFuturesBatchCancelItem::from_order_id(venue_order_id.to_string())
1276    } else {
1277        KrakenFuturesBatchCancelItem::from_client_order_id(truncated_client_order_id.clone())
1278    };
1279
1280    Ok((
1281        CancelRequestContext {
1282            strategy_id: cancel.strategy_id,
1283            instrument_id: cancel.instrument_id,
1284            client_order_id: cancel.client_order_id,
1285            truncated_client_order_id,
1286            venue_order_id: cancel.venue_order_id,
1287        },
1288        item,
1289    ))
1290}
1291
1292fn batch_cancel_status(status: &FuturesBatchCancelStatus) -> Option<KrakenSendStatus> {
1293    status
1294        .cancel_status
1295        .as_ref()
1296        .map(|cancel_status| cancel_status.status)
1297        .or(status.status)
1298}
1299
1300fn batch_cancel_context<'a>(
1301    status: &FuturesBatchCancelStatus,
1302    contexts: &'a [CancelRequestContext],
1303    index: usize,
1304) -> Option<&'a CancelRequestContext> {
1305    if let Some(order_id) = status.order_id.as_deref()
1306        && let Some(context) = contexts.iter().find(|context| {
1307            context
1308                .venue_order_id
1309                .is_some_and(|venue_order_id| venue_order_id.as_str() == order_id)
1310        })
1311    {
1312        return Some(context);
1313    }
1314
1315    if let Some(cli_ord_id) = status.cli_ord_id.as_deref()
1316        && let Some(context) = contexts
1317            .iter()
1318            .find(|context| context.truncated_client_order_id == cli_ord_id)
1319    {
1320        return Some(context);
1321    }
1322
1323    if index < contexts.len() && status.order_id.is_none() && status.cli_ord_id.is_none() {
1324        return contexts.get(index);
1325    }
1326
1327    None
1328}
1329
1330fn handle_cancel_failure(
1331    emitter: &ExecutionEventEmitter,
1332    clock: &'static AtomicTime,
1333    strategy_id: StrategyId,
1334    instrument_id: InstrumentId,
1335    client_order_id: ClientOrderId,
1336    venue_order_id: Option<VenueOrderId>,
1337    failure: CancelCommandFailure,
1338) {
1339    match failure {
1340        CancelCommandFailure::VenueReject(reason) => {
1341            emitter.emit_order_cancel_rejected_event(
1342                strategy_id,
1343                instrument_id,
1344                client_order_id,
1345                venue_order_id,
1346                &reason,
1347                clock.get_time_ns(),
1348            );
1349        }
1350        CancelCommandFailure::LocalValidation(reason) => {
1351            log::warn!("Cancel command failed local validation for {client_order_id}: {reason}");
1352        }
1353        CancelCommandFailure::Ambiguous(reason) => {
1354            log::warn!(
1355                "Ambiguous cancel failure for {client_order_id}, awaiting reconciliation: {reason}"
1356            );
1357        }
1358    }
1359}
1360
1361impl KrakenFuturesExecutionClient {
1362    fn get_cached_order_for_status_command(
1363        &self,
1364        cmd: &GenerateOrderStatusReport,
1365    ) -> Option<OrderAny> {
1366        let cache = self.core.cache();
1367
1368        if let Some(client_order_id) = cmd.client_order_id {
1369            return cache.order(&client_order_id).map(|o| o.clone());
1370        }
1371
1372        let venue_order_id = cmd.venue_order_id?;
1373        let client_order_id = *cache.client_order_id(&venue_order_id)?;
1374        cache.order(&client_order_id).map(|o| o.clone())
1375    }
1376}
1377
1378fn synthesize_filled_order_status_report(
1379    cmd: &GenerateOrderStatusReport,
1380    order: &OrderAny,
1381    fills: &[FillReport],
1382) -> Option<OrderStatusReport> {
1383    let venue_order_id = cmd.venue_order_id.or(order.venue_order_id());
1384    let truncated_client_order_id = truncate_cl_ord_id(&order.client_order_id());
1385
1386    let mut matched: Vec<&FillReport> = if let Some(venue_order_id) = venue_order_id {
1387        fills
1388            .iter()
1389            .filter(|fill| fill.venue_order_id == venue_order_id)
1390            .collect()
1391    } else {
1392        Vec::new()
1393    };
1394
1395    if matched.is_empty() {
1396        matched = fills
1397            .iter()
1398            .filter(|fill| {
1399                fill.client_order_id == Some(order.client_order_id())
1400                    || fill
1401                        .client_order_id
1402                        .as_ref()
1403                        .is_some_and(|fill_client_order_id| {
1404                            fill_client_order_id.as_str() == truncated_client_order_id
1405                        })
1406            })
1407            .collect();
1408    }
1409
1410    if matched.is_empty() {
1411        return None;
1412    }
1413
1414    matched.sort_by_key(|fill| fill.ts_event);
1415    let first_fill = *matched.first()?;
1416    let last_fill = *matched.last()?;
1417
1418    let total_filled = matched
1419        .iter()
1420        .fold(Decimal::ZERO, |acc, fill| acc + fill.last_qty.as_decimal());
1421    if total_filled < order.quantity().as_decimal() {
1422        return None;
1423    }
1424
1425    let total_notional = matched.iter().fold(Decimal::ZERO, |acc, fill| {
1426        acc + fill.last_qty.as_decimal() * fill.last_px.as_decimal()
1427    });
1428    let avg_px = if total_filled.is_zero() {
1429        None
1430    } else {
1431        Some(total_notional / total_filled)
1432    };
1433    let venue_order_id = venue_order_id.unwrap_or(first_fill.venue_order_id);
1434
1435    let mut report = OrderStatusReport::new(
1436        first_fill.account_id,
1437        order.instrument_id(),
1438        Some(order.client_order_id()),
1439        venue_order_id,
1440        order.order_side(),
1441        order.order_type(),
1442        order.time_in_force(),
1443        OrderStatus::Filled,
1444        order.quantity(),
1445        order.quantity(),
1446        first_fill.ts_event,
1447        last_fill.ts_event,
1448        last_fill.ts_init,
1449        None,
1450    );
1451    report.order_list_id = order.order_list_id();
1452    report.venue_position_id = matched.iter().rev().find_map(|fill| fill.venue_position_id);
1453    report.linked_order_ids = order
1454        .linked_order_ids()
1455        .map(|linked_order_ids| linked_order_ids.to_vec());
1456    report.parent_order_id = order.parent_order_id();
1457    report.expire_time = order.expire_time();
1458    report.price = order.price();
1459    report.trigger_price = order.trigger_price();
1460    report.trigger_type = order.trigger_type();
1461    report.avg_px = avg_px;
1462    report.display_qty = order.display_qty();
1463    report.post_only = order.is_post_only();
1464    report.reduce_only = order.is_reduce_only();
1465    Some(report)
1466}
1467
1468#[cfg(test)]
1469mod tests {
1470    use nautilus_core::{UUID4, UnixNanos};
1471    use nautilus_model::{
1472        enums::{LiquiditySide, OrderSide, OrderType, TimeInForce},
1473        identifiers::{
1474            AccountId, ClientOrderId, InstrumentId, StrategyId, TradeId, TraderId, VenueOrderId,
1475        },
1476        orders::OrderTestBuilder,
1477        reports::FillReport,
1478        types::{Currency, Money, Price, Quantity},
1479    };
1480    use rstest::rstest;
1481
1482    use super::*;
1483
1484    const TEST_INSTRUMENT_ID: &str = "PF_XBTUSD.KRAKEN";
1485
1486    #[tokio::test]
1487    async fn test_cancel_order_for_futures_missing_cached_instrument_returns_canonical_error() {
1488        let http = KrakenFuturesHttpClient::default();
1489        let instrument_id = InstrumentId::from(TEST_INSTRUMENT_ID);
1490        let client_order_id = ClientOrderId::from("C-001");
1491        let venue_order_id = VenueOrderId::from("V-001");
1492
1493        let result = cancel_order_for_futures(
1494            &http,
1495            AccountId::from("KRAKEN-001"),
1496            instrument_id,
1497            Some(client_order_id),
1498            Some(venue_order_id),
1499        )
1500        .await;
1501
1502        match result {
1503            Err(CancelCommandFailure::LocalValidation(reason)) => {
1504                assert_eq!(
1505                    reason,
1506                    InstrumentLookupError::not_found(instrument_id).to_string()
1507                );
1508            }
1509            _ => panic!("Expected local validation failure"),
1510        }
1511    }
1512
1513    #[rstest]
1514    fn test_batch_cancel_item_for_futures_missing_cached_instrument_returns_canonical_error() {
1515        let http = KrakenFuturesHttpClient::default();
1516        let instrument_id = InstrumentId::from(TEST_INSTRUMENT_ID);
1517        let cancel = CancelOrder::new(
1518            TraderId::from("TESTER-001"),
1519            None,
1520            StrategyId::from("S-001"),
1521            instrument_id,
1522            ClientOrderId::from("C-001"),
1523            Some(VenueOrderId::from("V-001")),
1524            UUID4::new(),
1525            UnixNanos::default(),
1526            None,
1527            None,
1528        );
1529
1530        let result = batch_cancel_item_for_futures(&http, &cancel);
1531
1532        match result {
1533            Err(CancelCommandFailure::LocalValidation(reason)) => {
1534                assert_eq!(
1535                    reason,
1536                    InstrumentLookupError::not_found(instrument_id).to_string()
1537                );
1538            }
1539            _ => panic!("Expected local validation failure"),
1540        }
1541    }
1542
1543    fn make_fill(
1544        venue_order_id: &str,
1545        client_order_id: Option<&str>,
1546        quantity: &str,
1547        price: &str,
1548        ts_event: u64,
1549    ) -> FillReport {
1550        FillReport::new(
1551            AccountId::from("KRAKEN-001"),
1552            InstrumentId::from(TEST_INSTRUMENT_ID),
1553            VenueOrderId::from(venue_order_id),
1554            TradeId::from(format!("T-{ts_event}").as_str()),
1555            OrderSide::Buy,
1556            Quantity::from(quantity),
1557            Price::from(price),
1558            Money::new(0.0, Currency::USD()),
1559            LiquiditySide::Taker,
1560            client_order_id.map(ClientOrderId::from),
1561            None,
1562            UnixNanos::from(ts_event),
1563            UnixNanos::from(ts_event),
1564            None,
1565        )
1566    }
1567
1568    fn make_cmd(
1569        client_order_id: Option<&str>,
1570        venue_order_id: Option<&str>,
1571    ) -> GenerateOrderStatusReport {
1572        GenerateOrderStatusReport::new(
1573            UUID4::new(),
1574            UnixNanos::default(),
1575            Some(InstrumentId::from(TEST_INSTRUMENT_ID)),
1576            client_order_id.map(ClientOrderId::from),
1577            venue_order_id.map(VenueOrderId::from),
1578            None,
1579            None,
1580        )
1581    }
1582
1583    fn make_order(client_order_id: &str) -> OrderAny {
1584        OrderTestBuilder::new(OrderType::Market)
1585            .instrument_id(InstrumentId::from(TEST_INSTRUMENT_ID))
1586            .client_order_id(ClientOrderId::from(client_order_id))
1587            .side(OrderSide::Buy)
1588            .quantity(Quantity::from("100"))
1589            .time_in_force(TimeInForce::Ioc)
1590            .build()
1591    }
1592
1593    #[rstest]
1594    fn test_synthesize_filled_order_status_report_matches_full_fill_by_venue_order_id() {
1595        let order = make_order("O-123456");
1596        let cmd = make_cmd(Some("O-123456"), Some("KRAKEN-789"));
1597        let fills = vec![
1598            make_fill("KRAKEN-789", Some("O-123456"), "40", "50000.0", 1),
1599            make_fill("KRAKEN-789", Some("O-123456"), "60", "50010.0", 2),
1600            make_fill("KRAKEN-OTHER", Some("O-123456"), "999", "1.0", 3),
1601        ];
1602
1603        let report = synthesize_filled_order_status_report(&cmd, &order, &fills)
1604            .expect("expected a filled report");
1605
1606        assert_eq!(report.venue_order_id, VenueOrderId::from("KRAKEN-789"));
1607        assert_eq!(
1608            report.client_order_id,
1609            Some(ClientOrderId::from("O-123456"))
1610        );
1611        assert_eq!(report.order_status, OrderStatus::Filled);
1612        assert_eq!(report.order_type, OrderType::Market);
1613        assert_eq!(report.time_in_force, TimeInForce::Ioc);
1614        assert_eq!(report.quantity, Quantity::from("100"));
1615        assert_eq!(report.filled_qty, Quantity::from("100"));
1616        assert_eq!(
1617            report.avg_px,
1618            Some(Decimal::from_str_exact("50006.0").unwrap())
1619        );
1620    }
1621
1622    #[rstest]
1623    fn test_synthesize_filled_order_status_report_requires_full_fill_size() {
1624        let order = make_order("O-123457");
1625        let cmd = make_cmd(Some("O-123457"), Some("KRAKEN-790"));
1626        let fills = vec![make_fill(
1627            "KRAKEN-790",
1628            Some("O-123457"),
1629            "40",
1630            "50000.0",
1631            1,
1632        )];
1633
1634        assert!(synthesize_filled_order_status_report(&cmd, &order, &fills).is_none());
1635    }
1636
1637    #[rstest]
1638    fn test_synthesize_filled_order_status_report_matches_truncated_client_order_id() {
1639        let long_client_order_id = "O202602270023210040011";
1640        let order = make_order(long_client_order_id);
1641        let cmd = make_cmd(Some(long_client_order_id), None);
1642        let fills = vec![make_fill(
1643            "KRAKEN-791",
1644            Some(truncate_cl_ord_id(&ClientOrderId::from(long_client_order_id)).as_str()),
1645            "100",
1646            "50000.0",
1647            1,
1648        )];
1649
1650        let report = synthesize_filled_order_status_report(&cmd, &order, &fills)
1651            .expect("expected a filled report");
1652
1653        assert_eq!(
1654            report.client_order_id,
1655            Some(ClientOrderId::from(long_client_order_id))
1656        );
1657        assert_eq!(report.venue_order_id, VenueOrderId::from("KRAKEN-791"));
1658        assert_eq!(report.order_status, OrderStatus::Filled);
1659    }
1660}
nautilus_kraken/execution/futures.rs

nautilus_kraken/execution/
futures.rs
