Skip to main content

nautilus_kraken/execution/
spot.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 Spot execution client implementation.
17
18use std::{
19    collections::HashSet,
20    future::Future,
21    sync::Arc,
22    time::{Duration, Instant},
23};
24
25use anyhow::Context;
26use async_trait::async_trait;
27use futures_util::StreamExt;
28use jiff::Timestamp;
29use nautilus_common::{
30    cache::InstrumentLookupError,
31    clients::ExecutionClient,
32    live::runner::get_exec_event_sender,
33    messages::execution::{
34        BatchCancelOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
35        GenerateOrderStatusReport, GenerateOrderStatusReports, GeneratePositionStatusReports,
36        ModifyOrder, QueryAccount, QueryOrder, SubmitOrder, SubmitOrderList,
37    },
38};
39use nautilus_core::{
40    AtomicMap, DurationNanos, Params, UUID4, UnixNanos,
41    time::{AtomicTime, get_atomic_clock_realtime},
42};
43use nautilus_live::{
44    ExecutionClientCore, ExecutionEventEmitter, SocketControl, execution::failure::CommandFailure,
45    task::TaskGroup,
46};
47use nautilus_model::{
48    accounts::AccountAny,
49    enums::{
50        AccountType, OmsType, OrderSide, OrderType, TimeInForce, TrailingOffsetType, TriggerType,
51    },
52    events::OrderEventAny,
53    identifiers::{
54        AccountId, ClientId, ClientOrderId, InstrumentId, StrategyId, Venue, VenueOrderId,
55    },
56    instruments::{Instrument, InstrumentAny},
57    orders::{Order, OrderAny},
58    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
59    types::{AccountBalance, MarginBalance, Price, Quantity},
60};
61use rust_decimal::Decimal;
62use tokio_util::sync::CancellationToken;
63use ustr::Ustr;
64
65use super::{
66    absent_unless_unread, cached_order_for_status_command, command_failure_from_cancel_error,
67    command_failure_from_modify_error, command_failure_from_spot_batch_error,
68    command_failure_from_spot_batch_item, command_failure_from_spot_cancel_error,
69    command_failure_from_submit_error, earliest_venue_time, status_command_label,
70};
71use crate::{
72    common::{
73        consts::{KRAKEN_SPOT_POST_ONLY_ERROR, KRAKEN_VENUE},
74        enums::{KrakenOrderSide, KrakenOrderType, KrakenSpotTrigger, KrakenTimeInForce},
75        order_params::{
76            build_add_order_params, build_amend_order_params, build_cancel_order_params,
77            compute_ws_time_in_force, format_expire_time,
78        },
79        parse::truncate_cl_ord_id,
80    },
81    config::KrakenExecutionClientConfig,
82    http::{
83        KrakenSpotCancelOrderBatchParams, KrakenSpotCancelOrderParamsBuilder, KrakenSpotHttpClient,
84        spot::client::KRAKEN_SPOT_DEFAULT_RATE_LIMIT_PER_SECOND,
85    },
86    websocket::{
87        dispatch::{
88            self, OrderIdentity, WsDispatchState,
89            spot_orders::{OrderRequestState, PendingOperation, PendingRequest},
90        },
91        spot_v2::{
92            client::KrakenSpotWebSocketClient,
93            messages::{
94                KrakenSpotWsMessage, KrakenWsBatchAddOrder, KrakenWsBatchAddParams,
95                KrakenWsTriggerParams,
96            },
97        },
98    },
99};
100
101/// Kraken Spot execution client.
102///
103/// Provides order management and account operations for Kraken Spot markets.
104#[allow(dead_code)]
105#[derive(Debug)]
106pub struct KrakenSpotExecutionClient {
107    core: ExecutionClientCore,
108    clock: &'static AtomicTime,
109    config: KrakenExecutionClientConfig,
110    emitter: ExecutionEventEmitter,
111    http: KrakenSpotHttpClient,
112    ws: KrakenSpotWebSocketClient,
113    cancellation_token: CancellationToken,
114    session_tasks: TaskGroup,
115    pending_tasks: TaskGroup,
116    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
117    order_qty_cache: Arc<AtomicMap<String, Decimal>>,
118    truncated_id_map: Arc<AtomicMap<String, ClientOrderId>>,
119    ws_dispatch_state: Arc<WsDispatchState>,
120    order_request_state: Arc<OrderRequestState>,
121    order_event_rx: Option<tokio::sync::mpsc::UnboundedReceiver<OrderEventAny>>,
122}
123
124impl KrakenSpotExecutionClient {
125    /// Creates a new [`KrakenSpotExecutionClient`].
126    pub fn new(
127        core: ExecutionClientCore,
128        config: KrakenExecutionClientConfig,
129    ) -> anyhow::Result<Self> {
130        let clock = get_atomic_clock_realtime();
131        let emitter = ExecutionEventEmitter::new(
132            clock,
133            core.trader_id,
134            core.account_id,
135            config.spot_account_type,
136            None,
137        );
138
139        let session_tasks = TaskGroup::new();
140        let cancellation_token = session_tasks.cancellation_token();
141        let pending_tasks = TaskGroup::new();
142        let pending_spawner = pending_tasks
143            .spawner()
144            .context("Kraken Spot execution task admission is closed")?;
145        let api_key = config.api_key.expose_secret().to_owned();
146        let api_secret = config.api_secret.expose_secret().to_owned();
147        let proxy_url = config
148            .proxy_url
149            .as_ref()
150            .map(|value| value.expose_secret().to_owned());
151
152        let http = KrakenSpotHttpClient::with_credentials(
153            api_key,
154            api_secret,
155            config.environment,
156            config.base_url.clone(),
157            config.timeout_secs,
158            Some(config.max_retries),
159            None,
160            None,
161            proxy_url.clone(),
162            config
163                .max_requests_per_second
164                .unwrap_or(KRAKEN_SPOT_DEFAULT_RATE_LIMIT_PER_SECOND),
165        )?;
166
167        let data_config = crate::config::KrakenDataClientConfig {
168            api_key: Some(config.api_key.clone()),
169            api_secret: Some(config.api_secret.clone()),
170            product_type: config.product_type,
171            environment: config.environment,
172            base_url: config.base_url.clone(),
173            ws_public_url: None,
174            ws_private_url: Some(config.ws_url()),
175            ws_l3_url: None,
176            validate_l3_checksum: true,
177            validate_l2_checksum: true,
178            proxy_url: config.proxy_url.clone(),
179            timeout_secs: config.timeout_secs,
180            heartbeat_interval_secs: config.heartbeat_interval_secs,
181            // The authenticated order-routing WS keeps its prior behavior (no
182            // idle timeout); issue #4255 concerns the public spot data WS.
183            ws_idle_timeout_ms: 0,
184            max_requests_per_second: config.max_requests_per_second,
185            transport_backend: config.transport_backend,
186        };
187        let ws = KrakenSpotWebSocketClient::new(data_config, cancellation_token.clone(), proxy_url)
188            .with_socket_control(SocketControl::new(
189                core.client_id,
190                Some(*KRAKEN_VENUE),
191                "kraken-spot-user-streams",
192            ));
193
194        let ws_dispatch_state = Arc::new(WsDispatchState::new());
195        // Connect() swaps in a live cmd_tx; capture the shared handle so the
196        // dispatcher reads the current sender, not the dropped placeholder.
197        let cmd_tx_handle = ws.handler_command_handle();
198        let (order_event_tx, order_event_rx) = tokio::sync::mpsc::unbounded_channel();
199        let order_request_state = Arc::new(OrderRequestState::new(
200            cmd_tx_handle,
201            order_event_tx,
202            Arc::clone(&ws_dispatch_state),
203            ws.req_id_counter(),
204            Duration::from_secs(config.ws_request_timeout_secs),
205            core.trader_id,
206            core.account_id,
207            ws.auth_token_handle(),
208            pending_spawner,
209            clock,
210        ));
211
212        Ok(Self {
213            core,
214            clock,
215            config,
216            emitter,
217            http,
218            ws,
219            cancellation_token,
220            session_tasks,
221            pending_tasks,
222            instruments: Arc::new(AtomicMap::new()),
223            order_qty_cache: Arc::new(AtomicMap::new()),
224            truncated_id_map: Arc::new(AtomicMap::new()),
225            ws_dispatch_state,
226            order_request_state,
227            order_event_rx: Some(order_event_rx),
228        })
229    }
230
231    fn register_order_identity(&self, order: &OrderAny) {
232        // Quote-quantity orders submit a quote amount (e.g. 100 USD), but the
233        // venue reports fills in base units (e.g. 0.001 BTC). Registering the
234        // raw `order.quantity()` would make the cumulative-fill comparison in
235        // the fill-side dispatch mismatch base against quote, leaving the
236        // order "open" forever. These orders instead flow through the
237        // untracked path and the engine reconciles them from status reports.
238        if order.is_quote_quantity() {
239            return;
240        }
241        self.ws_dispatch_state.register_identity(
242            order.client_order_id(),
243            OrderIdentity {
244                strategy_id: order.strategy_id(),
245                instrument_id: order.instrument_id(),
246                order_side: order.order_side(),
247                order_type: order.order_type(),
248                quantity: order.quantity(),
249            },
250        );
251    }
252
253    /// Returns a reference to the clock.
254    #[must_use]
255    pub fn clock(&self) -> &'static AtomicTime {
256        self.clock
257    }
258
259    /// Returns a reference to the event emitter.
260    #[must_use]
261    pub fn emitter(&self) -> &ExecutionEventEmitter {
262        &self.emitter
263    }
264
265    fn spawn_task<F>(&self, description: &'static str, fut: F)
266    where
267        F: Future<Output = anyhow::Result<()>> + Send + 'static,
268    {
269        let future = async move {
270            if let Err(e) = fut.await {
271                log::warn!("{description} failed: {e:?}");
272            }
273        };
274
275        if let Err(e) = self.pending_tasks.spawn(future) {
276            log::warn!("Skipping Kraken Spot {description} after shutdown began: {e}");
277        }
278    }
279
280    async fn finish_tasks(&self) -> anyhow::Result<()> {
281        let (session_result, pending_result) = tokio::join!(
282            self.session_tasks
283                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
284            self.pending_tasks
285                .finish_shutdown(Duration::from_secs(2), Duration::from_secs(2)),
286        );
287        session_result.context("failed to finish Kraken Spot execution session tasks")?;
288        pending_result.context("failed to finish Kraken Spot execution command tasks")?;
289        Ok(())
290    }
291
292    async fn prepare_task_groups(&mut self) -> anyhow::Result<()> {
293        if !self.session_tasks.is_open() || !self.pending_tasks.is_open() {
294            self.session_tasks.begin_shutdown();
295            self.pending_tasks.begin_shutdown();
296            self.finish_tasks().await?;
297            self.session_tasks
298                .start_generation()
299                .context("failed to start Kraken Spot execution session task generation")?;
300            self.pending_tasks
301                .start_generation()
302                .context("failed to start Kraken Spot execution command task generation")?;
303            self.cancellation_token = self.session_tasks.cancellation_token();
304            let pending_spawner = self
305                .pending_tasks
306                .spawner()
307                .context("Kraken Spot execution task admission is closed")?;
308            self.order_request_state.reset_task_spawner(pending_spawner);
309        }
310        Ok(())
311    }
312
313    async fn teardown_partial_connect(&mut self) -> anyhow::Result<()> {
314        self.http.cancel_all_requests();
315        self.pending_tasks.begin_shutdown();
316        self.order_request_state.clear();
317        let ws_result = self.ws.close().await;
318        self.session_tasks.begin_shutdown();
319        let tasks_result = self.finish_tasks().await;
320        self.core.set_disconnected();
321        tasks_result?;
322        Ok(ws_result?)
323    }
324
325    fn submit_single_order(
326        &self,
327        command: &SubmitOrder,
328        order: &OrderAny,
329        task_name: &'static str,
330        leverage: Option<u16>,
331    ) {
332        if order.is_closed() {
333            log::warn!(
334                "Cannot submit closed order: client_order_id={}",
335                order.client_order_id()
336            );
337            return;
338        }
339
340        let order_type = order.order_type();
341        let time_in_force = order.time_in_force();
342
343        if time_in_force == TimeInForce::Fok && order_type != OrderType::Limit {
344            self.emitter.emit_order_denied(
345                order,
346                "FOK time in force only supported for LIMIT orders on Kraken Spot",
347            );
348            return;
349        }
350
351        if matches!(
352            order_type,
353            OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
354        ) && let Some(offset_type) = order.trailing_offset_type()
355            && offset_type != TrailingOffsetType::Price
356        {
357            self.emitter.emit_order_denied(
358                order,
359                &format!(
360                    "Kraken Spot only supports Price trailing offset type: received {offset_type:?}"
361                ),
362            );
363            return;
364        }
365
366        if order.is_reduce_only() && self.config.spot_account_type == AccountType::Cash {
367            self.emitter
368                .emit_order_denied(order, "reduce_only requires spot_account_type=Margin");
369            return;
370        }
371
372        let client_order_id = order.client_order_id();
373
374        log::debug!("OrderSubmitted: client_order_id={client_order_id}");
375        self.register_order_identity(order);
376        self.emitter.emit_order_submitted(order);
377
378        let kraken_cl_ord_id = truncate_cl_ord_id(&client_order_id);
379
380        if !order.is_quote_quantity() {
381            self.order_qty_cache
382                .insert(kraken_cl_ord_id.clone(), order.quantity().as_decimal());
383        }
384
385        if kraken_cl_ord_id != client_order_id.as_str() {
386            self.truncated_id_map
387                .insert(kraken_cl_ord_id, client_order_id);
388        }
389
390        // Quote-quantity orders submit a quote-currency amount but the venue echoes
391        // fills in base units. The WS dispatch identity is intentionally not
392        // registered for these (see register_order_identity), so a WS round-trip
393        // would never emit OrderAccepted to the strategy. Plus order.quantity()
394        // for quote-qty is a quote amount that would be wrongly sent as `order_qty`
395        // (base units) on the WS path. Force REST.
396        let use_ws_trade = resolve_use_ws_trade(command.params.as_ref(), self.config.use_ws_trade);
397        if use_ws_trade && self.ws.is_active() && !order.is_quote_quantity() {
398            match self.submit_via_ws(command, order, leverage) {
399                Ok(()) => return,
400                Err(e) => log::warn!("Kraken WS submit_order fallback to REST: {e}"),
401            }
402        }
403
404        self.submit_via_rest(order, task_name, leverage);
405    }
406
407    fn submit_via_rest(&self, order: &OrderAny, task_name: &'static str, leverage: Option<u16>) {
408        let account_id = self.core.account_id;
409        let client_order_id = order.client_order_id();
410        let strategy_id = order.strategy_id();
411        let instrument_id = order.instrument_id();
412        let order_side = order.order_side();
413        let order_type = order.order_type();
414        let quantity = order.quantity();
415        let time_in_force = order.time_in_force();
416        let expire_time = order.expire_time();
417        let price = order.price();
418        let trigger_price = order.trigger_price();
419        let trigger_type = order.trigger_type();
420        let trailing_offset = order.trailing_offset();
421        let limit_offset = order.limit_offset();
422        let is_reduce_only = order.is_reduce_only();
423        let is_post_only = order.is_post_only();
424        let is_quote_quantity = order.is_quote_quantity();
425        let display_qty = order.display_qty();
426
427        let http = self.http.clone();
428        let emitter = self.emitter.clone();
429        let clock = self.clock;
430        let dispatch_state = self.ws_dispatch_state.clone();
431        let spot_account_type = self.config.spot_account_type;
432
433        self.spawn_task(task_name, async move {
434            let result = http
435                .submit_order(
436                    account_id,
437                    instrument_id,
438                    client_order_id,
439                    order_side,
440                    order_type,
441                    quantity,
442                    time_in_force,
443                    expire_time,
444                    price,
445                    trigger_price,
446                    trigger_type,
447                    trailing_offset,
448                    limit_offset,
449                    is_reduce_only,
450                    is_post_only,
451                    is_quote_quantity,
452                    display_qty,
453                    leverage,
454                    spot_account_type,
455                )
456                .await;
457
458            match result {
459                Ok(_) => {}
460                Err(e) => match command_failure_from_submit_error(&e) {
461                    CommandFailure::Ambiguous(reason) => {
462                        log::warn!(
463                            "{task_name} outcome is ambiguous for client_order_id={client_order_id}: {reason}"
464                        );
465                    }
466                    CommandFailure::NotSent(reason) | CommandFailure::VenueRejected(reason) => {
467                        let ts_event = clock.get_time_ns();
468                        let error_msg = format!("{task_name} error: {reason}");
469                        let due_post_only = error_msg.contains("POST_ONLY_REJECTED")
470                            || error_msg.contains(KRAKEN_SPOT_POST_ONLY_ERROR);
471                        dispatch_state.cleanup_terminal(&client_order_id);
472                        emitter.emit_order_rejected_event(
473                            strategy_id,
474                            instrument_id,
475                            client_order_id,
476                            &error_msg,
477                            ts_event,
478                            due_post_only,
479                        );
480                    }
481                },
482            }
483
484            Ok(())
485        });
486    }
487    fn submit_via_ws(
488        &self,
489        command: &SubmitOrder,
490        order: &OrderAny,
491        leverage: Option<u16>,
492    ) -> anyhow::Result<()> {
493        let token = self
494            .ws
495            .auth_token_blocking()
496            .ok_or_else(|| anyhow::anyhow!("missing WS auth token"))?;
497
498        let params = build_add_order_params(command, order, token, leverage)?;
499        let identity = PendingRequest {
500            operation: PendingOperation::Submit,
501            client_order_ids: vec![command.client_order_id],
502            venue_order_ids: vec![None],
503            ts_sent_ns: 0,
504            new_quantity: None,
505            new_price: None,
506            new_trigger_price: None,
507        };
508        self.order_request_state
509            .submit(params, identity, self.clock.get_time_ns().as_u64())?;
510        Ok(())
511    }
512
513    fn cancel_single_order(&self, cmd: &CancelOrder) {
514        let use_ws_trade = resolve_use_ws_trade(cmd.params.as_ref(), self.config.use_ws_trade);
515        if use_ws_trade && self.ws.is_active() {
516            match self.cancel_via_ws(cmd) {
517                Ok(()) => return,
518                Err(e) => log::warn!("Kraken WS cancel_order fallback to REST: {e}"),
519            }
520        }
521
522        self.cancel_via_rest(cmd);
523    }
524
525    fn cancel_via_rest(&self, cmd: &CancelOrder) {
526        let account_id = self.core.account_id;
527        let client_order_id = cmd.client_order_id;
528        let venue_order_id = cmd.venue_order_id;
529        let strategy_id = cmd.strategy_id;
530        let instrument_id = cmd.instrument_id;
531
532        log::debug!(
533            "Canceling order: venue_order_id={venue_order_id:?}, client_order_id={client_order_id}"
534        );
535
536        let http = self.http.clone();
537        let emitter = self.emitter.clone();
538        let clock = self.clock;
539
540        self.spawn_task("cancel_order", async move {
541            if let Err(failure) = cancel_order_for_spot(
542                &http,
543                account_id,
544                instrument_id,
545                Some(client_order_id),
546                venue_order_id,
547            )
548            .await
549            {
550                handle_cancel_failure(
551                    &emitter,
552                    clock,
553                    strategy_id,
554                    instrument_id,
555                    client_order_id,
556                    venue_order_id,
557                    failure,
558                );
559            }
560            Ok(())
561        });
562    }
563
564    fn cancel_via_ws(&self, cmd: &CancelOrder) -> anyhow::Result<()> {
565        let token = self
566            .ws
567            .auth_token_blocking()
568            .ok_or_else(|| anyhow::anyhow!("missing WS auth token"))?;
569
570        let params = build_cancel_order_params(cmd, token);
571        let identity = PendingRequest {
572            operation: PendingOperation::Cancel,
573            client_order_ids: vec![cmd.client_order_id],
574            venue_order_ids: vec![cmd.venue_order_id],
575            ts_sent_ns: 0,
576            new_quantity: None,
577            new_price: None,
578            new_trigger_price: None,
579        };
580        self.order_request_state
581            .cancel(params, identity, self.clock.get_time_ns().as_u64())?;
582        Ok(())
583    }
584
585    fn spawn_message_handler(&mut self) -> anyhow::Result<()> {
586        let stream = self.ws.stream().map_err(|e| anyhow::anyhow!("{e}"))?;
587        let emitter = self.emitter.clone();
588        let instruments = self.instruments.clone();
589        let order_qty_cache = self.order_qty_cache.clone();
590        let truncated_id_map = self.truncated_id_map.clone();
591        let dispatch_state = self.ws_dispatch_state.clone();
592        let order_request_state = self.order_request_state.clone();
593        let account_id = self.core.account_id;
594        let clock = self.clock;
595        let cancellation_token = self.cancellation_token.clone();
596
597        let future = async move {
598            tokio::pin!(stream);
599
600            loop {
601                tokio::select! {
602                    () = cancellation_token.cancelled() => {
603                        log::debug!("Spot execution message handler cancelled");
604                        break;
605                    }
606                    msg = stream.next() => {
607                        match msg {
608                            Some(ws_msg) => {
609                                Self::handle_ws_message(
610                                    ws_msg,
611                                    &emitter,
612                                    &dispatch_state,
613                                    &order_request_state,
614                                    &instruments,
615                                    &order_qty_cache,
616                                    &truncated_id_map,
617                                    account_id,
618                                    clock,
619                                );
620                            }
621                            None => {
622                                log::debug!("Spot execution WebSocket stream ended");
623                                break;
624                            }
625                        }
626                    }
627                }
628            }
629        };
630
631        self.session_tasks
632            .spawn(future)
633            .context("failed to register Kraken Spot execution stream task")?;
634
635        let event_rx = self.order_event_rx.take();
636
637        if let Some(mut event_rx) = event_rx {
638            let emitter = self.emitter.clone();
639            let cancellation_token = self.cancellation_token.clone();
640
641            let future = async move {
642                loop {
643                    tokio::select! {
644                        () = cancellation_token.cancelled() => {
645                            log::debug!("Spot execution order-event forwarder cancelled");
646                            break;
647                        }
648                        event = event_rx.recv() => {
649                            match event {
650                                Some(event) => emitter.send_order_event(event),
651                                None => {
652                                    log::debug!("Spot execution order-event channel closed");
653                                    break;
654                                }
655                            }
656                        }
657                    }
658                }
659            };
660            self.session_tasks
661                .spawn(future)
662                .context("failed to register Kraken Spot order event task")?;
663        }
664
665        Ok(())
666    }
667
668    fn modify_single_order(&self, cmd: &ModifyOrder) {
669        let use_ws_trade = resolve_use_ws_trade(cmd.params.as_ref(), self.config.use_ws_trade);
670        if use_ws_trade && self.ws.is_active() {
671            match self.amend_via_ws(cmd) {
672                Ok(()) => return,
673                Err(e) => log::warn!("Kraken WS amend_order fallback to REST: {e}"),
674            }
675        }
676
677        self.amend_via_rest(cmd);
678    }
679
680    fn amend_via_rest(&self, cmd: &ModifyOrder) {
681        let client_order_id = cmd.client_order_id;
682        let venue_order_id = cmd.venue_order_id;
683        let strategy_id = cmd.strategy_id;
684        let instrument_id = cmd.instrument_id;
685        let quantity = cmd.quantity;
686        let price = cmd.price;
687
688        log::debug!(
689            "Modifying order: venue_order_id={venue_order_id:?}, client_order_id={client_order_id}"
690        );
691
692        let http = self.http.clone();
693        let emitter = self.emitter.clone();
694        let clock = self.clock;
695
696        self.spawn_task("modify_order", async move {
697            match http
698                .modify_order(
699                    instrument_id,
700                    Some(client_order_id),
701                    venue_order_id,
702                    quantity,
703                    price,
704                    None,
705                )
706                .await
707            {
708                Ok(_) => {}
709                Err(e) => match command_failure_from_modify_error(&e) {
710                    CommandFailure::Ambiguous(reason) => {
711                        log::warn!(
712                            "modify_order outcome is ambiguous for client_order_id={client_order_id}: {reason}"
713                        );
714                    }
715                    CommandFailure::NotSent(reason) | CommandFailure::VenueRejected(reason) => {
716                        let ts_event = clock.get_time_ns();
717                        emitter.emit_order_modify_rejected_event(
718                            strategy_id,
719                            instrument_id,
720                            client_order_id,
721                            venue_order_id,
722                            &format!("modify-order error: {reason}"),
723                            ts_event,
724                        );
725                    }
726                },
727            }
728            Ok(())
729        });
730    }
731
732    fn amend_via_ws(&self, cmd: &ModifyOrder) -> anyhow::Result<()> {
733        let token = self
734            .ws
735            .auth_token_blocking()
736            .ok_or_else(|| anyhow::anyhow!("missing WS auth token"))?;
737
738        let params = build_amend_order_params(cmd, token);
739        let identity = PendingRequest {
740            operation: PendingOperation::Amend,
741            client_order_ids: vec![cmd.client_order_id],
742            venue_order_ids: vec![cmd.venue_order_id],
743            ts_sent_ns: 0,
744            new_quantity: cmd.quantity,
745            new_price: cmd.price,
746            new_trigger_price: cmd.trigger_price,
747        };
748        self.order_request_state
749            .amend(params, identity, self.clock.get_time_ns().as_u64())?;
750        Ok(())
751    }
752
753    /// Polls the cache until the account is registered or timeout is reached.
754    async fn await_account_registered(&self, timeout_secs: f64) -> anyhow::Result<()> {
755        let account_id = self.core.account_id;
756
757        if self.core.cache().account(&account_id).is_some() {
758            log::info!("Account {account_id} registered");
759            return Ok(());
760        }
761
762        let start = Instant::now();
763        let timeout = Duration::from_secs_f64(timeout_secs);
764        let interval = Duration::from_millis(10);
765
766        loop {
767            tokio::time::sleep(interval).await;
768
769            if self.core.cache().account(&account_id).is_some() {
770                log::info!("Account {account_id} registered");
771                return Ok(());
772            }
773
774            if start.elapsed() >= timeout {
775                anyhow::bail!(
776                    "Timeout waiting for account {account_id} to be registered after {timeout_secs}s"
777                );
778            }
779        }
780    }
781
782    #[expect(clippy::too_many_arguments)]
783    fn handle_ws_message(
784        msg: KrakenSpotWsMessage,
785        emitter: &ExecutionEventEmitter,
786        dispatch_state: &Arc<WsDispatchState>,
787        order_request_state: &Arc<OrderRequestState>,
788        instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
789        order_qty_cache: &Arc<AtomicMap<String, Decimal>>,
790        truncated_id_map: &Arc<AtomicMap<String, ClientOrderId>>,
791        account_id: AccountId,
792        clock: &'static AtomicTime,
793    ) {
794        match msg {
795            KrakenSpotWsMessage::Execution(executions) => {
796                let ts_init = clock.get_time_ns();
797
798                for exec in &executions {
799                    dispatch::spot::execution(
800                        exec,
801                        dispatch_state,
802                        emitter,
803                        instruments,
804                        truncated_id_map,
805                        order_qty_cache,
806                        account_id,
807                        ts_init,
808                    );
809                }
810            }
811            KrakenSpotWsMessage::OrderResponse(response) => {
812                let ts_event = clock.get_time_ns().as_u64();
813                order_request_state.handle_response(&response, ts_event);
814            }
815            KrakenSpotWsMessage::Reconnected => {
816                log::info!("Spot execution WebSocket reconnected");
817            }
818            KrakenSpotWsMessage::Ticker(_)
819            | KrakenSpotWsMessage::Trade(_)
820            | KrakenSpotWsMessage::Book { .. }
821            | KrakenSpotWsMessage::Ohlc(_)
822            | KrakenSpotWsMessage::L3Snapshot(_)
823            | KrakenSpotWsMessage::L3Update(_)
824            | KrakenSpotWsMessage::SubscriptionAck { .. } => {}
825        }
826    }
827
828    fn batch_add_via_rest(
829        &self,
830        order_tuples: Vec<BatchOrderTuple>,
831        order_meta: Vec<(StrategyId, InstrumentId, ClientOrderId)>,
832    ) {
833        let http = self.http.clone();
834        let emitter = self.emitter.clone();
835        let clock = self.clock;
836        let dispatch_state = self.ws_dispatch_state.clone();
837        let spot_account_type = self.config.spot_account_type;
838
839        self.spawn_task("submit_order_list", async move {
840            let results = http
841                .send_order_batches(order_tuples, spot_account_type)
842                .await;
843
844            for (result, (strategy_id, instrument_id, client_order_id)) in
845                results.into_iter().zip(&order_meta)
846            {
847                let outcome = match result {
848                    Ok(item) => command_failure_from_spot_batch_item(item),
849                    Err(e) => Err(command_failure_from_spot_batch_error(&e)),
850                };
851
852                match outcome {
853                    Ok(()) => {}
854                    Err(CommandFailure::Ambiguous(reason)) => {
855                        log::warn!(
856                            "submit_order_list outcome is ambiguous for client_order_id={client_order_id}: {reason}"
857                        );
858                    }
859                    Err(
860                        CommandFailure::NotSent(reason)
861                        | CommandFailure::VenueRejected(reason),
862                    ) => {
863                        let ts_event = clock.get_time_ns();
864                        let due_post_only = reason.contains("POST_ONLY_REJECTED")
865                            || reason.contains(KRAKEN_SPOT_POST_ONLY_ERROR);
866                        dispatch_state.cleanup_terminal(client_order_id);
867                        emitter.emit_order_rejected_event(
868                            *strategy_id,
869                            *instrument_id,
870                            *client_order_id,
871                            &format!("submit_order_list batch item rejected: {reason}"),
872                            ts_event,
873                            due_post_only,
874                        );
875                    }
876                }
877            }
878            Ok(())
879        });
880    }
881
882    fn batch_add_via_ws(&self, orders: &[OrderAny], leverage: Option<u16>) -> anyhow::Result<()> {
883        let token = self
884            .ws
885            .auth_token_blocking()
886            .ok_or_else(|| anyhow::anyhow!("missing WS auth token"))?;
887
888        let first = orders
889            .first()
890            .ok_or_else(|| anyhow::anyhow!("batch_add requires at least one order"))?;
891        let symbol = first.instrument_id().symbol.inner().to_string();
892
893        let mut batch_orders = Vec::with_capacity(orders.len());
894        let mut client_order_ids = Vec::with_capacity(orders.len());
895        for order in orders {
896            batch_orders.push(build_batch_order(order, leverage)?);
897            client_order_ids.push(order.client_order_id());
898        }
899        let venue_order_ids = vec![None; orders.len()];
900
901        let params = KrakenWsBatchAddParams {
902            symbol,
903            orders: batch_orders,
904            token,
905        };
906        let identity = PendingRequest {
907            operation: PendingOperation::BatchAdd,
908            client_order_ids,
909            venue_order_ids,
910            ts_sent_ns: 0,
911            new_quantity: None,
912            new_price: None,
913            new_trigger_price: None,
914        };
915        self.order_request_state
916            .batch_add(params, identity, self.clock.get_time_ns().as_u64())?;
917        Ok(())
918    }
919}
920
921type BatchOrderTuple = (
922    InstrumentId,
923    ClientOrderId,
924    OrderSide,
925    OrderType,
926    Quantity,
927    TimeInForce,
928    Option<UnixNanos>,
929    Option<Price>,
930    Option<Price>,
931    Option<TriggerType>,
932    Option<Decimal>,
933    Option<Decimal>,
934    bool,
935    bool,
936    bool,
937    Option<Quantity>,
938    Option<u16>,
939);
940
941fn build_batch_order(
942    order: &OrderAny,
943    leverage: Option<u16>,
944) -> anyhow::Result<KrakenWsBatchAddOrder> {
945    let order_type = order.order_type();
946    let side = match order.order_side() {
947        OrderSide::Buy => KrakenOrderSide::Buy,
948        OrderSide::Sell => KrakenOrderSide::Sell,
949    };
950
951    if matches!(
952        order_type,
953        OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
954    ) {
955        anyhow::bail!(
956            "Trailing stop orders are not yet supported on the Kraken WS batch path; use REST",
957        );
958    }
959
960    if order.display_qty().is_some() {
961        anyhow::bail!(
962            "Iceberg (display_qty) orders are not supported on the Kraken WS batch path; use REST",
963        );
964    }
965
966    let kraken_order_type = match order_type {
967        OrderType::Market => KrakenOrderType::Market,
968        OrderType::Limit => KrakenOrderType::Limit,
969        OrderType::StopMarket => KrakenOrderType::StopLoss,
970        OrderType::StopLimit => KrakenOrderType::StopLossLimit,
971        OrderType::MarketIfTouched => KrakenOrderType::TakeProfit,
972        OrderType::LimitIfTouched => KrakenOrderType::TakeProfitLimit,
973        ty => anyhow::bail!("Unsupported order type for Kraken WS batch: {ty:?}"),
974    };
975
976    let is_limit_order = matches!(
977        order_type,
978        OrderType::Limit | OrderType::StopLimit | OrderType::LimitIfTouched
979    );
980
981    if is_limit_order && order.price().is_none() {
982        anyhow::bail!("limit_price is required for batch order type {order_type:?}");
983    }
984
985    let ws_tif =
986        compute_ws_time_in_force(is_limit_order, order.time_in_force(), order.expire_time())?;
987    let expire_time = match (ws_tif, order.expire_time()) {
988        (Some(KrakenTimeInForce::GoodTilDate), Some(ts)) => Some(format_expire_time(ts)),
989        _ => None,
990    };
991
992    let is_conditional = matches!(
993        order_type,
994        OrderType::StopMarket
995            | OrderType::StopLimit
996            | OrderType::MarketIfTouched
997            | OrderType::LimitIfTouched
998    );
999
1000    let trigger = if is_conditional {
1001        let trigger_ref = match order.trigger_type() {
1002            Some(TriggerType::IndexPrice) => KrakenSpotTrigger::Index,
1003            Some(TriggerType::LastPrice | TriggerType::Default) | None => KrakenSpotTrigger::Last,
1004            Some(other) => anyhow::bail!(
1005                "Unsupported trigger type for Kraken Spot WS batch: {other:?} (only LastPrice and IndexPrice supported)",
1006            ),
1007        };
1008        order.trigger_price().map(|tp| KrakenWsTriggerParams {
1009            reference: trigger_ref,
1010            price: tp.as_decimal(),
1011            price_type: None,
1012        })
1013    } else {
1014        None
1015    };
1016
1017    if is_conditional && trigger.is_none() {
1018        anyhow::bail!(
1019            "Conditional order type {order_type:?} requires trigger_price for Kraken WS batch",
1020        );
1021    }
1022
1023    Ok(KrakenWsBatchAddOrder {
1024        order_type: kraken_order_type,
1025        side,
1026        order_qty: order.quantity().as_decimal(),
1027        limit_price: order.price().map(|p| p.as_decimal()),
1028        cl_ord_id: Some(truncate_cl_ord_id(&order.client_order_id())),
1029        time_in_force: ws_tif,
1030        expire_time,
1031        post_only: order.is_post_only().then_some(true),
1032        reduce_only: order.is_reduce_only().then_some(true),
1033        leverage,
1034        trigger,
1035    })
1036}
1037
1038#[async_trait(?Send)]
1039impl ExecutionClient for KrakenSpotExecutionClient {
1040    fn is_connected(&self) -> bool {
1041        self.core.is_connected()
1042    }
1043
1044    fn client_id(&self) -> ClientId {
1045        self.core.client_id
1046    }
1047
1048    fn account_id(&self) -> AccountId {
1049        self.core.account_id
1050    }
1051
1052    fn venue(&self) -> Venue {
1053        *KRAKEN_VENUE
1054    }
1055
1056    fn oms_type(&self) -> OmsType {
1057        self.core.oms_type
1058    }
1059
1060    fn provides_bulk_position_coverage(&self, instrument_id: InstrumentId) -> bool {
1061        // Deferred to the HTTP client so the answer is derived from the read that produces the
1062        // reports, rather than restated here where it could drift.
1063        self.http.covers_bulk_position_reports(
1064            instrument_id,
1065            self.config.spot_account_type,
1066            self.config.use_spot_position_reports,
1067            Ustr::from(self.config.spot_positions_quote_currency.as_str()),
1068        )
1069    }
1070
1071    fn get_account(&self) -> Option<AccountAny> {
1072        self.core.cache().account_owned(&self.core.account_id)
1073    }
1074
1075    fn generate_account_state(
1076        &self,
1077        balances: Vec<AccountBalance>,
1078        margins: Vec<MarginBalance>,
1079        reported: bool,
1080        ts_event: UnixNanos,
1081        info: Option<Params>,
1082    ) -> anyhow::Result<()> {
1083        self.emitter
1084            .emit_account_state(balances, margins, reported, ts_event, info);
1085        Ok(())
1086    }
1087
1088    fn start(&mut self) -> anyhow::Result<()> {
1089        if self.core.is_started() {
1090            return Ok(());
1091        }
1092
1093        self.emitter.set_sender(get_exec_event_sender());
1094        self.core.set_started();
1095
1096        log::info!(
1097            "Started: client_id={}, account_id={}, product_type=Spot, environment={:?}",
1098            self.core.client_id,
1099            self.core.account_id,
1100            self.config.environment
1101        );
1102        Ok(())
1103    }
1104
1105    fn stop(&mut self) -> anyhow::Result<()> {
1106        if self.core.is_stopped() {
1107            return Ok(());
1108        }
1109
1110        self.http.cancel_all_requests();
1111        self.session_tasks.begin_shutdown();
1112        self.pending_tasks.begin_shutdown();
1113        self.ws.begin_shutdown();
1114        self.core.set_stopped();
1115        self.core.set_disconnected();
1116        log::info!("Stopped: client_id={}", self.core.client_id);
1117        Ok(())
1118    }
1119
1120    async fn connect(&mut self) -> anyhow::Result<()> {
1121        if self.core.is_connected() && self.session_tasks.is_open() && self.pending_tasks.is_open()
1122        {
1123            return Ok(());
1124        }
1125
1126        self.http.reset_cancellation_token();
1127        self.prepare_task_groups().await?;
1128
1129        if !self.core.instruments_initialized() {
1130            let instruments = self
1131                .http
1132                .request_instruments(None)
1133                .await
1134                .context("Failed to load Kraken spot instruments")?;
1135            log::debug!("Loaded {} Spot instruments", instruments.len());
1136            self.http.cache_instruments(&instruments);
1137            self.core.set_instruments_initialized();
1138        }
1139
1140        let session_result = async {
1141            self.ws
1142                .connect()
1143                .await
1144                .context("Failed to connect spot WebSocket")?;
1145            self.ws
1146                .wait_until_active(10.0)
1147                .await
1148                .context("Spot WebSocket failed to become active")?;
1149
1150            self.ws
1151                .authenticate()
1152                .await
1153                .context("Failed to authenticate spot WebSocket")?;
1154
1155            // Request initial account state and await registration before spawning
1156            // the message handler. Report events from execution snapshots conflict
1157            // with ExecEngine borrows during startup, so account registration must
1158            // complete first.
1159            let account_state = self
1160                .http
1161                .request_account_state(
1162                    self.core.account_id,
1163                    self.config.spot_account_type,
1164                    self.config.margin_balance_asset.as_deref(),
1165                )
1166                .await
1167                .context("Failed to request Kraken account state")?;
1168
1169            if !account_state.balances.is_empty() {
1170                log::debug!(
1171                    "Received account state with {} balance(s)",
1172                    account_state.balances.len()
1173                );
1174            }
1175
1176            self.emitter.send_account_state(account_state);
1177            self.await_account_registered(30.0).await?;
1178
1179            self.spawn_message_handler()?;
1180
1181            self.instruments.rcu(|m| {
1182                for instrument in self.http.instruments_cache.load().values() {
1183                    m.insert(instrument.id(), instrument.clone());
1184                }
1185            });
1186
1187            self.ws
1188                .subscribe_executions(false, false)
1189                .await
1190                .context("Failed to subscribe to executions")?;
1191
1192            log::debug!("Spot WebSocket authenticated and subscribed to executions");
1193
1194            Ok::<(), anyhow::Error>(())
1195        }
1196        .await;
1197
1198        if let Err(e) = session_result {
1199            if let Err(teardown_error) = self.teardown_partial_connect().await {
1200                return Err(e.context(format!(
1201                    "Kraken Spot execution startup teardown failed: {teardown_error}"
1202                )));
1203            }
1204            return Err(e);
1205        }
1206
1207        self.core.set_connected();
1208        log::info!("Connected: client_id={}", self.core.client_id);
1209        Ok(())
1210    }
1211
1212    async fn disconnect(&mut self) -> anyhow::Result<()> {
1213        self.teardown_partial_connect().await?;
1214        log::info!("Disconnected: client_id={}", self.core.client_id);
1215        Ok(())
1216    }
1217
1218    async fn generate_order_status_report(
1219        &self,
1220        cmd: &GenerateOrderStatusReport,
1221    ) -> anyhow::Result<Option<OrderStatusReport>> {
1222        log::debug!(
1223            "Generating order status report: venue_order_id={:?}, client_order_id={:?}",
1224            cmd.venue_order_id,
1225            cmd.client_order_id
1226        );
1227
1228        let account_id = self.core.account_id;
1229        // Scoped to the command's instrument, so an unresolvable open order on another pair is out
1230        // of scope for this lookup rather than failing it.
1231        let (reports, coverage) = self
1232            .http
1233            .request_order_status_reports_checked(account_id, cmd.instrument_id, None, None, false)
1234            .await?;
1235
1236        // Match by venue_order_id or client_order_id (comparing truncated form
1237        // since Kraken stores the truncated cl_ord_id for long IDs)
1238        let matched = reports.into_iter().find(|r| {
1239            cmd.venue_order_id
1240                .is_some_and(|id| r.venue_order_id.as_str() == id.as_str())
1241                || cmd.client_order_id.is_some_and(|id| {
1242                    r.client_order_id
1243                        .as_ref()
1244                        .is_some_and(|r_id| r_id.as_str() == truncate_cl_ord_id(&id))
1245                })
1246        });
1247
1248        if matched.is_some() {
1249            return Ok(matched);
1250        }
1251
1252        // Absence answers that the venue has no such order, so it needs a read that rules the
1253        // order out; otherwise reconciliation defers.
1254        let cached = cached_order_for_status_command(&self.core.cache(), cmd);
1255        let venue_order_id = cmd
1256            .venue_order_id
1257            .or_else(|| cached.as_ref().and_then(|order| order.venue_order_id()));
1258
1259        absent_unless_unread(
1260            &coverage,
1261            venue_order_id,
1262            cached.as_ref().map(earliest_venue_time),
1263            &status_command_label(cmd, venue_order_id),
1264        )
1265    }
1266
1267    async fn generate_order_status_reports(
1268        &self,
1269        cmd: &GenerateOrderStatusReports,
1270    ) -> anyhow::Result<Vec<OrderStatusReport>> {
1271        log::debug!(
1272            "Generating order status reports: instrument_id={:?}, open_only={}",
1273            cmd.instrument_id,
1274            cmd.open_only
1275        );
1276
1277        let account_id = self.core.account_id;
1278        let start = cmd.start.map(Timestamp::from);
1279        let end = cmd.end.map(Timestamp::from);
1280        self.http
1281            .request_order_status_reports(account_id, cmd.instrument_id, start, end, cmd.open_only)
1282            .await
1283    }
1284
1285    async fn generate_fill_reports(
1286        &self,
1287        cmd: GenerateFillReports,
1288    ) -> anyhow::Result<Vec<FillReport>> {
1289        log::debug!(
1290            "Generating fill reports: instrument_id={:?}",
1291            cmd.instrument_id
1292        );
1293
1294        let account_id = self.core.account_id;
1295        let start = cmd.start.map(Timestamp::from);
1296        let end = cmd.end.map(Timestamp::from);
1297        self.http
1298            .request_fill_reports(account_id, cmd.instrument_id, start, end)
1299            .await
1300    }
1301
1302    async fn generate_position_status_reports(
1303        &self,
1304        cmd: &GeneratePositionStatusReports,
1305    ) -> anyhow::Result<Vec<PositionStatusReport>> {
1306        log::debug!(
1307            "Generating position status reports: instrument_id={:?}",
1308            cmd.instrument_id
1309        );
1310
1311        let account_id = self.core.account_id;
1312        self.http
1313            .request_position_status_reports(
1314                account_id,
1315                cmd.instrument_id,
1316                self.config.spot_account_type,
1317                self.config.use_spot_position_reports,
1318                Ustr::from(self.config.spot_positions_quote_currency.as_str()),
1319            )
1320            .await
1321    }
1322
1323    async fn generate_mass_status(
1324        &self,
1325        lookback_mins: Option<u64>,
1326    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
1327        log::debug!("Generating mass status: lookback_mins={lookback_mins:?}");
1328
1329        let ts_init = self.clock.get_time_ns();
1330        // Saturating arithmetic: an unclamped lookback must not overflow, and a cutoff before the
1331        // epoch must not panic converting into `UnixNanos`.
1332        let lookback_start = lookback_mins.map(|mins| {
1333            let nanos = mins.saturating_mul(60).saturating_mul(1_000_000_000);
1334            ts_init.saturating_sub(DurationNanos::new(nanos))
1335        });
1336        let start = lookback_start.map(Timestamp::from);
1337
1338        let account_id = self.core.account_id;
1339        // Read closed orders as well as open ones, matching the shared default. An order that
1340        // reached a terminal state while the node was down is only visible through ClosedOrders.
1341        // The lookback cutoff and the closed-order page cap bound the read.
1342        let (order_reports, orders_coverage) = self
1343            .http
1344            .request_order_status_reports_checked(account_id, None, start, None, false)
1345            .await?;
1346        let orders_complete = orders_coverage.is_complete();
1347        let (fill_reports, fills_complete) = self
1348            .http
1349            .request_fill_reports_checked(account_id, None, start, None)
1350            .await?;
1351        let position_reports = self
1352            .http
1353            .request_position_status_reports(
1354                account_id,
1355                None,
1356                self.config.spot_account_type,
1357                self.config.use_spot_position_reports,
1358                Ustr::from(self.config.spot_positions_quote_currency.as_str()),
1359            )
1360            .await?;
1361
1362        let mut mass_status = ExecutionMassStatus::new(
1363            self.core.client_id,
1364            self.core.account_id,
1365            *KRAKEN_VENUE,
1366            ts_init,
1367            None,
1368        );
1369        mass_status.add_order_reports(order_reports);
1370        mass_status.add_fill_reports(fill_reports);
1371        mass_status.add_position_reports(position_reports);
1372        // One cutoff covers every historical query above, so record it with the completeness of
1373        // the sources the engine needs to interpret the bounded set.
1374        mass_status.set_report_window(lookback_start, orders_complete && fills_complete);
1375
1376        Ok(Some(mass_status))
1377    }
1378
1379    fn query_account(&self, cmd: QueryAccount) -> anyhow::Result<()> {
1380        log::debug!("Querying account: {cmd}");
1381
1382        let account_id = self.core.account_id;
1383        let http = self.http.clone();
1384        let emitter = self.emitter.clone();
1385
1386        let spot_account_type = self.config.spot_account_type;
1387        let margin_balance_asset = self.config.margin_balance_asset.clone();
1388        self.spawn_task("query_account", async move {
1389            let account_state = http
1390                .request_account_state(
1391                    account_id,
1392                    spot_account_type,
1393                    margin_balance_asset.as_deref(),
1394                )
1395                .await?;
1396            emitter.emit_account_state(
1397                account_state.balances.clone(),
1398                account_state.margins.clone(),
1399                account_state.is_reported,
1400                account_state.ts_event,
1401                account_state.info,
1402            );
1403            Ok(())
1404        });
1405
1406        Ok(())
1407    }
1408
1409    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
1410        log::debug!("Querying order: {cmd}");
1411
1412        let venue_order_id = cmd
1413            .venue_order_id
1414            .context("venue_order_id required for query_order")?;
1415        let account_id = self.core.account_id;
1416        let instrument_id = cmd.instrument_id;
1417        let http = self.http.clone();
1418        let emitter = self.emitter.clone();
1419
1420        self.spawn_task("query_order", async move {
1421            // Scoped to the queried instrument, so an unresolvable open order on another pair is
1422            // out of scope for this lookup rather than failing it.
1423            let reports = http
1424                .request_order_status_reports(account_id, Some(instrument_id), None, None, true)
1425                .await
1426                .context("Failed to query order")?;
1427
1428            match reports
1429                .into_iter()
1430                .find(|r| r.venue_order_id == venue_order_id)
1431            {
1432                Some(report) => emitter.send_order_status_report(report),
1433                None => log::warn!(
1434                    "No open order {venue_order_id} found for {instrument_id}; no report emitted"
1435                ),
1436            }
1437            Ok(())
1438        });
1439
1440        Ok(())
1441    }
1442
1443    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
1444        let order = self.core.cache().try_order_owned(&cmd.client_order_id)?;
1445        let leverage = match resolve_leverage(cmd.params.as_ref(), self.config.default_leverage) {
1446            Ok(lev) => lev,
1447            Err(reason) => {
1448                self.emitter.emit_order_denied(&order, &reason);
1449                return Ok(());
1450            }
1451        };
1452        self.submit_single_order(&cmd, &order, "submit_order", leverage);
1453        Ok(())
1454    }
1455
1456    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
1457        let orders = self.core.get_orders_for_list(&cmd.order_list)?;
1458
1459        log::debug!(
1460            "Submitting order list: order_list_id={}, count={}",
1461            cmd.order_list.id,
1462            orders.len()
1463        );
1464
1465        let leverage = match resolve_leverage(cmd.params.as_ref(), self.config.default_leverage) {
1466            Ok(lev) => lev,
1467            Err(reason) => {
1468                for order in &orders {
1469                    self.emitter.emit_order_denied(order, &reason);
1470                }
1471                return Ok(());
1472            }
1473        };
1474
1475        let mut order_tuples = Vec::with_capacity(orders.len());
1476        let mut order_meta = Vec::with_capacity(orders.len());
1477        let mut prepared_orders = Vec::with_capacity(orders.len());
1478
1479        for order in &orders {
1480            if order.is_closed() {
1481                log::warn!(
1482                    "Cannot submit closed order: client_order_id={}",
1483                    order.client_order_id()
1484                );
1485                continue;
1486            }
1487
1488            if order.time_in_force() == TimeInForce::Fok && order.order_type() != OrderType::Limit {
1489                self.emitter.emit_order_denied(
1490                    order,
1491                    "FOK time in force only supported for LIMIT orders on Kraken Spot",
1492                );
1493                continue;
1494            }
1495
1496            if matches!(
1497                order.order_type(),
1498                OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
1499            ) && let Some(offset_type) = order.trailing_offset_type()
1500                && offset_type != TrailingOffsetType::Price
1501            {
1502                self.emitter.emit_order_denied(
1503                    order,
1504                    &format!(
1505                        "Kraken Spot only supports Price trailing offset type: received {offset_type:?}"
1506                    ),
1507                );
1508                continue;
1509            }
1510
1511            if order.is_reduce_only() && self.config.spot_account_type == AccountType::Cash {
1512                self.emitter
1513                    .emit_order_denied(order, "reduce_only requires spot_account_type=Margin");
1514                continue;
1515            }
1516
1517            let client_order_id = order.client_order_id();
1518            let kraken_cl_ord_id = truncate_cl_ord_id(&client_order_id);
1519
1520            self.register_order_identity(order);
1521            self.emitter.emit_order_submitted(order);
1522
1523            if !order.is_quote_quantity() {
1524                self.order_qty_cache
1525                    .insert(kraken_cl_ord_id.clone(), order.quantity().as_decimal());
1526            }
1527
1528            if kraken_cl_ord_id != client_order_id.as_str() {
1529                self.truncated_id_map
1530                    .insert(kraken_cl_ord_id, client_order_id);
1531            }
1532            order_tuples.push((
1533                order.instrument_id(),
1534                client_order_id,
1535                order.order_side(),
1536                order.order_type(),
1537                order.quantity(),
1538                order.time_in_force(),
1539                order.expire_time(),
1540                order.price(),
1541                order.trigger_price(),
1542                order.trigger_type(),
1543                order.trailing_offset(),
1544                order.limit_offset(),
1545                order.is_reduce_only(),
1546                order.is_post_only(),
1547                order.is_quote_quantity(),
1548                order.display_qty(),
1549                leverage,
1550            ));
1551
1552            order_meta.push((order.strategy_id(), order.instrument_id(), client_order_id));
1553            prepared_orders.push(order.clone());
1554        }
1555
1556        if order_tuples.is_empty() {
1557            return Ok(());
1558        }
1559
1560        let use_ws_trade = resolve_use_ws_trade(cmd.params.as_ref(), self.config.use_ws_trade);
1561        if use_ws_trade && self.ws.is_active() {
1562            let any_quote_qty = prepared_orders.iter().any(|o| o.is_quote_quantity());
1563            let symbols_match = prepared_orders
1564                .windows(2)
1565                .all(|w| w[0].instrument_id() == w[1].instrument_id());
1566
1567            if any_quote_qty {
1568                log::debug!(
1569                    "Kraken WS batch_add does not support quote-quantity orders, falling back to REST for order_list_id={}",
1570                    cmd.order_list.id,
1571                );
1572            } else if symbols_match {
1573                match self.batch_add_via_ws(&prepared_orders, leverage) {
1574                    Ok(()) => return Ok(()),
1575                    Err(e) => log::warn!("Kraken WS batch_add fallback to REST: {e}"),
1576                }
1577            } else {
1578                log::debug!(
1579                    "Kraken WS batch_add requires single shared symbol, falling back to REST for order_list_id={}",
1580                    cmd.order_list.id,
1581                );
1582            }
1583        }
1584
1585        self.batch_add_via_rest(order_tuples, order_meta);
1586
1587        Ok(())
1588    }
1589
1590    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
1591        self.modify_single_order(&cmd);
1592        Ok(())
1593    }
1594
1595    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
1596        self.cancel_single_order(&cmd);
1597        Ok(())
1598    }
1599
1600    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
1601        let instrument_id = cmd.instrument_id;
1602
1603        log::debug!(
1604            "Canceling all orders: instrument_id={instrument_id}, side={:?}",
1605            cmd.order_side
1606        );
1607
1608        // Kraken's account-wide `CancelAll` ignores the requested instrument, so select the
1609        // matching open orders and cancel them by explicit venue id instead.
1610        let cancels: Vec<CancelOrder> = {
1611            let cache = self.core.cache();
1612            let ts_init = self.clock.get_time_ns();
1613            let correlation_id = cmd.correlation_id.or(Some(cmd.command_id));
1614
1615            // In-flight orders are included because the venue can have accepted an order the
1616            // cache still records as `Submitted`. The venue-wide cancellation this replaced
1617            // reached those, so selecting only open orders would silently stop covering them.
1618            let mut seen = HashSet::new();
1619
1620            cache
1621                .orders_open(None, Some(&instrument_id), None, None, None)
1622                .into_iter()
1623                .chain(cache.orders_inflight(None, Some(&instrument_id), None, None, None))
1624                .filter(|order| cmd.order_side.is_none_or(|side| order.order_side() == side))
1625                .filter(|order| seen.insert(order.client_order_id()))
1626                .map(|order| {
1627                    CancelOrder::new(
1628                        cmd.trader_id,
1629                        cmd.client_id,
1630                        // Each cancel keeps the owning strategy of the order it targets.
1631                        order.strategy_id(),
1632                        order.instrument_id(),
1633                        order.client_order_id(),
1634                        order.venue_order_id(),
1635                        UUID4::new(),
1636                        ts_init,
1637                        cmd.params.clone(),
1638                        correlation_id,
1639                    )
1640                })
1641                .collect()
1642        };
1643
1644        if cancels.is_empty() {
1645            log::debug!("No open orders to cancel for {instrument_id}");
1646            return Ok(());
1647        }
1648
1649        let http = self.http.clone();
1650
1651        self.spawn_task("cancel_all_orders", async move {
1652            batch_cancel_orders_for_spot(&http, &cancels).await;
1653            Ok(())
1654        });
1655
1656        Ok(())
1657    }
1658
1659    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
1660        log::debug!(
1661            "Batch canceling orders: instrument_id={}, count={}",
1662            cmd.instrument_id,
1663            cmd.cancels.len()
1664        );
1665
1666        let use_ws_trade = resolve_use_ws_trade(cmd.params.as_ref(), self.config.use_ws_trade);
1667        if use_ws_trade && self.ws.is_active() {
1668            for cancel in &cmd.cancels {
1669                self.cancel_single_order(cancel);
1670            }
1671
1672            return Ok(());
1673        }
1674
1675        let http = self.http.clone();
1676        let cancels = cmd.cancels;
1677
1678        self.spawn_task("batch_cancel_orders", async move {
1679            batch_cancel_orders_for_spot(&http, &cancels).await;
1680            Ok(())
1681        });
1682
1683        Ok(())
1684    }
1685}
1686
1687async fn cancel_order_for_spot(
1688    http: &KrakenSpotHttpClient,
1689    _account_id: AccountId,
1690    instrument_id: InstrumentId,
1691    client_order_id: Option<ClientOrderId>,
1692    venue_order_id: Option<VenueOrderId>,
1693) -> Result<(), CommandFailure> {
1694    http.get_cached_instrument(&instrument_id.symbol.inner())
1695        .ok_or_else(|| {
1696            CommandFailure::not_sent(InstrumentLookupError::not_found(instrument_id).to_string())
1697        })?;
1698
1699    let txid = venue_order_id.as_ref().map(ToString::to_string);
1700    let cl_ord_id = client_order_id.as_ref().map(truncate_cl_ord_id);
1701
1702    if txid.is_none() && cl_ord_id.is_none() {
1703        return Err(CommandFailure::not_sent(
1704            "Either client_order_id or venue_order_id must be provided",
1705        ));
1706    }
1707
1708    let mut builder = KrakenSpotCancelOrderParamsBuilder::default();
1709
1710    if let Some(ref id) = txid {
1711        builder.txid(id.clone());
1712    } else if let Some(ref id) = cl_ord_id {
1713        builder.cl_ord_id(id.clone());
1714    }
1715
1716    let params = builder
1717        .build()
1718        .map_err(|e| CommandFailure::not_sent(format!("Failed to build cancel params: {e}")))?;
1719
1720    http.inner
1721        .cancel_order(&params)
1722        .await
1723        .map_err(command_failure_from_spot_cancel_error)?;
1724
1725    Ok(())
1726}
1727
1728/// How Kraken identifies one order in a batch cancellation.
1729///
1730/// Kraken keys transaction IDs and client order IDs in separate request fields, so the two cannot
1731/// be mixed into one array.
1732enum SpotBatchCancelId {
1733    VenueOrderId(String),
1734    ClientOrderId(String),
1735}
1736
1737const SPOT_BATCH_CANCEL_LIMIT: usize = 50;
1738
1739async fn batch_cancel_orders_for_spot(http: &KrakenSpotHttpClient, cancels: &[CancelOrder]) {
1740    let mut ids = Vec::new();
1741
1742    for cancel in cancels {
1743        match batch_cancel_item_for_spot(http, cancel) {
1744            Ok(id) => ids.push(id),
1745            Err(CommandFailure::NotSent(reason)) => {
1746                log::warn!(
1747                    "Batch cancel command failed local validation for {}: {reason}",
1748                    cancel.client_order_id
1749                );
1750            }
1751            Err(CommandFailure::Ambiguous(reason) | CommandFailure::VenueRejected(reason)) => {
1752                log::warn!(
1753                    "Batch cancel command ambiguous failure for {}, awaiting reconciliation: {reason}",
1754                    cancel.client_order_id
1755                );
1756            }
1757        }
1758    }
1759
1760    // The venue limit counts both fields together, so chunk before splitting them.
1761    for chunk in ids.chunks(SPOT_BATCH_CANCEL_LIMIT) {
1762        let mut orders = Vec::new();
1763        let mut cl_ord_ids = Vec::new();
1764
1765        for id in chunk {
1766            match id {
1767                SpotBatchCancelId::VenueOrderId(venue_order_id) => {
1768                    orders.push(venue_order_id.clone());
1769                }
1770                SpotBatchCancelId::ClientOrderId(client_order_id) => {
1771                    cl_ord_ids.push(client_order_id.clone());
1772                }
1773            }
1774        }
1775
1776        let params = KrakenSpotCancelOrderBatchParams { orders, cl_ord_ids };
1777
1778        match http.inner.cancel_order_batch(&params).await {
1779            Ok(response) => {
1780                if response.count < chunk.len() as i32 {
1781                    log::warn!(
1782                        "Batch cancel accepted {} of {} request(s) without per-order results; awaiting reconciliation",
1783                        response.count,
1784                        chunk.len()
1785                    );
1786                }
1787            }
1788            Err(e) => match command_failure_from_cancel_error(e) {
1789                CommandFailure::NotSent(reason) => {
1790                    log::warn!("Batch cancel failed local validation: {reason}");
1791                }
1792                CommandFailure::Ambiguous(reason) | CommandFailure::VenueRejected(reason) => {
1793                    log::warn!(
1794                        "Batch cancel failed without per-order results, awaiting reconciliation: {reason}"
1795                    );
1796                }
1797            },
1798        }
1799    }
1800}
1801
1802fn batch_cancel_item_for_spot(
1803    http: &KrakenSpotHttpClient,
1804    cancel: &CancelOrder,
1805) -> Result<SpotBatchCancelId, CommandFailure> {
1806    http.get_cached_instrument(&cancel.instrument_id.symbol.inner())
1807        .ok_or_else(|| {
1808            CommandFailure::not_sent(
1809                InstrumentLookupError::not_found(cancel.instrument_id).to_string(),
1810            )
1811        })?;
1812
1813    if let Some(venue_order_id) = cancel.venue_order_id {
1814        Ok(SpotBatchCancelId::VenueOrderId(venue_order_id.to_string()))
1815    } else {
1816        Ok(SpotBatchCancelId::ClientOrderId(truncate_cl_ord_id(
1817            &cancel.client_order_id,
1818        )))
1819    }
1820}
1821
1822fn handle_cancel_failure(
1823    emitter: &ExecutionEventEmitter,
1824    clock: &'static AtomicTime,
1825    strategy_id: StrategyId,
1826    instrument_id: InstrumentId,
1827    client_order_id: ClientOrderId,
1828    venue_order_id: Option<VenueOrderId>,
1829    failure: CommandFailure,
1830) {
1831    match failure {
1832        CommandFailure::VenueRejected(reason) => {
1833            emitter.emit_order_cancel_rejected_event(
1834                strategy_id,
1835                instrument_id,
1836                client_order_id,
1837                venue_order_id,
1838                &reason,
1839                clock.get_time_ns(),
1840            );
1841        }
1842        CommandFailure::NotSent(reason) => {
1843            log::warn!("Cancel command failed local validation for {client_order_id}: {reason}");
1844        }
1845        CommandFailure::Ambiguous(reason) => {
1846            log::warn!(
1847                "Ambiguous cancel failure for {client_order_id}, awaiting reconciliation: {reason}"
1848            );
1849        }
1850    }
1851}
1852
1853fn resolve_leverage(params: Option<&Params>, default: Option<u16>) -> Result<Option<u16>, String> {
1854    let Some(p) = params else {
1855        return Ok(default);
1856    };
1857    let Some(raw) = p.get("leverage") else {
1858        return Ok(default);
1859    };
1860    let n = raw.as_u64().ok_or_else(|| {
1861        format!("Invalid leverage param: expected unsigned integer, received {raw}")
1862    })?;
1863    let lev =
1864        u16::try_from(n).map_err(|_| format!("leverage {n} exceeds maximum ({})", u16::MAX))?;
1865    Ok(Some(lev))
1866}
1867
1868/// Resolves the per-call `params["use_ws_trade"]` override against the
1869/// configured default. Non-boolean values warn and fall back to the default.
1870fn resolve_use_ws_trade(params: Option<&Params>, default: bool) -> bool {
1871    let Some(p) = params else {
1872        return default;
1873    };
1874    let Some(raw) = p.get("use_ws_trade") else {
1875        return default;
1876    };
1877
1878    match raw.as_bool() {
1879        Some(b) => b,
1880        None => {
1881            log::warn!(
1882                "Invalid use_ws_trade param: expected boolean, received {raw}; using default {default}",
1883            );
1884            default
1885        }
1886    }
1887}
1888
1889#[cfg(test)]
1890mod tests {
1891    use std::{
1892        cell::RefCell,
1893        rc::Rc,
1894        sync::{
1895            Arc,
1896            atomic::{AtomicUsize, Ordering},
1897        },
1898    };
1899
1900    use axum::{Router, http::StatusCode, routing::post};
1901    use nautilus_common::{
1902        cache::{Cache, InstrumentLookupError},
1903        clock::VirtualClock,
1904        factories::ExecutionClientFactory,
1905        messages::execution::CancelOrder,
1906    };
1907    use nautilus_core::{Params, UUID4, UnixNanos};
1908    use nautilus_model::identifiers::{
1909        AccountId, ClientOrderId, InstrumentId, StrategyId, TraderId, VenueOrderId,
1910    };
1911    use rstest::rstest;
1912    use serde_json::json;
1913
1914    use super::{
1915        AccountType, ClientId, CommandFailure, ExecutionClientCore,
1916        KrakenSpotCancelOrderParamsBuilder, KrakenSpotExecutionClient, OmsType,
1917        batch_cancel_item_for_spot, cancel_order_for_spot, resolve_leverage, resolve_use_ws_trade,
1918    };
1919    use crate::{
1920        common::{consts::KRAKEN_VENUE, enums::KrakenProductType},
1921        config::KrakenExecutionClientConfig,
1922        factories::KrakenExecutionClientFactory,
1923        http::KrakenSpotHttpClient,
1924    };
1925
1926    const TEST_INSTRUMENT_ID: &str = "BTC/USDT.KRAKEN";
1927
1928    #[tokio::test]
1929    async fn test_execution_config_max_retries_zero_disables_spot_cancel_retries() {
1930        let request_count = Arc::new(AtomicUsize::new(0));
1931        let handler_count = request_count.clone();
1932        let router = Router::new().route(
1933            "/0/private/CancelOrder",
1934            post(move || {
1935                let handler_count = handler_count.clone();
1936                async move {
1937                    handler_count.fetch_add(1, Ordering::Relaxed);
1938                    StatusCode::TOO_MANY_REQUESTS
1939                }
1940            }),
1941        );
1942        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
1943        let addr = listener.local_addr().unwrap();
1944        tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
1945
1946        let cache = Rc::new(RefCell::new(Cache::default()));
1947        let core = ExecutionClientCore::new(
1948            TraderId::from("TRADER-001"),
1949            ClientId::from("KRAKEN"),
1950            *KRAKEN_VENUE,
1951            OmsType::Netting,
1952            AccountId::from("KRAKEN-001"),
1953            AccountType::Cash,
1954            None,
1955            cache,
1956        );
1957        let config = KrakenExecutionClientConfig {
1958            api_key: "test-key".into(),
1959            api_secret: "c2VjcmV0".into(),
1960            base_url: Some(format!("http://{addr}")),
1961            max_retries: 0,
1962            ..Default::default()
1963        };
1964        let client = KrakenSpotExecutionClient::new(core, config).unwrap();
1965        let params = KrakenSpotCancelOrderParamsBuilder::default()
1966            .txid("V-001".to_string())
1967            .build()
1968            .unwrap();
1969
1970        let result = client.http.inner.cancel_order(&params).await;
1971
1972        assert!(result.is_err());
1973        assert_eq!(request_count.load(Ordering::Relaxed), 1);
1974    }
1975
1976    fn params_with(key: &str, val: serde_json::Value) -> Params {
1977        let mut map = indexmap::IndexMap::new();
1978        map.insert(key.to_owned(), val);
1979        Params::from_index_map(map)
1980    }
1981
1982    #[tokio::test]
1983    async fn test_cancel_order_for_spot_missing_cached_instrument_returns_canonical_error() {
1984        let http = KrakenSpotHttpClient::default();
1985        let instrument_id = InstrumentId::from(TEST_INSTRUMENT_ID);
1986        let client_order_id = ClientOrderId::from("C-001");
1987        let venue_order_id = VenueOrderId::from("V-001");
1988
1989        let result = cancel_order_for_spot(
1990            &http,
1991            AccountId::from("KRAKEN-001"),
1992            instrument_id,
1993            Some(client_order_id),
1994            Some(venue_order_id),
1995        )
1996        .await;
1997
1998        match result {
1999            Err(CommandFailure::NotSent(reason)) => {
2000                assert_eq!(
2001                    reason,
2002                    InstrumentLookupError::not_found(instrument_id).to_string()
2003                );
2004            }
2005            _ => panic!("Expected local validation failure"),
2006        }
2007    }
2008
2009    #[rstest]
2010    fn test_batch_cancel_item_for_spot_missing_cached_instrument_returns_canonical_error() {
2011        let http = KrakenSpotHttpClient::default();
2012        let instrument_id = InstrumentId::from(TEST_INSTRUMENT_ID);
2013        let cancel = CancelOrder::new(
2014            TraderId::from("TESTER-001"),
2015            None,
2016            StrategyId::from("S-001"),
2017            instrument_id,
2018            ClientOrderId::from("C-001"),
2019            Some(VenueOrderId::from("V-001")),
2020            UUID4::new(),
2021            UnixNanos::default(),
2022            None,
2023            None,
2024        );
2025
2026        let result = batch_cancel_item_for_spot(&http, &cancel);
2027
2028        match result {
2029            Err(CommandFailure::NotSent(reason)) => {
2030                assert_eq!(
2031                    reason,
2032                    InstrumentLookupError::not_found(instrument_id).to_string()
2033                );
2034            }
2035            _ => panic!("Expected local validation failure"),
2036        }
2037    }
2038
2039    #[rstest]
2040    fn test_resolve_leverage_absent_uses_default() {
2041        let p = params_with("other", json!(1));
2042        assert_eq!(resolve_leverage(Some(&p), Some(3)).unwrap(), Some(3));
2043        assert_eq!(resolve_leverage(None, Some(5)).unwrap(), Some(5));
2044        assert_eq!(resolve_leverage(None, None).unwrap(), None);
2045    }
2046
2047    #[rstest]
2048    fn test_resolve_leverage_valid_integer() {
2049        let p = params_with("leverage", json!(5u64));
2050        assert_eq!(resolve_leverage(Some(&p), Some(3)).unwrap(), Some(5));
2051    }
2052
2053    #[rstest]
2054    fn test_resolve_leverage_string_value_errors() {
2055        let p = params_with("leverage", json!("5"));
2056        let err = resolve_leverage(Some(&p), Some(3)).unwrap_err();
2057        assert!(err.contains("Invalid leverage param"), "unexpected: {err}");
2058    }
2059
2060    #[rstest]
2061    fn test_resolve_leverage_overflow_errors() {
2062        let p = params_with("leverage", json!(65539u64));
2063        let err = resolve_leverage(Some(&p), None).unwrap_err();
2064        assert!(err.contains("exceeds maximum"), "unexpected: {err}");
2065    }
2066
2067    #[rstest]
2068    fn test_resolve_use_ws_trade_absent_uses_default() {
2069        let p = params_with("other", json!(1));
2070        assert!(resolve_use_ws_trade(Some(&p), true));
2071        assert!(!resolve_use_ws_trade(Some(&p), false));
2072        assert!(resolve_use_ws_trade(None, true));
2073        assert!(!resolve_use_ws_trade(None, false));
2074    }
2075
2076    #[rstest]
2077    fn test_resolve_use_ws_trade_overrides_default() {
2078        let p_false = params_with("use_ws_trade", json!(false));
2079        let p_true = params_with("use_ws_trade", json!(true));
2080        assert!(!resolve_use_ws_trade(Some(&p_false), true));
2081        assert!(resolve_use_ws_trade(Some(&p_true), false));
2082    }
2083
2084    #[rstest]
2085    fn test_resolve_use_ws_trade_non_boolean_falls_back_to_default() {
2086        let p = params_with("use_ws_trade", json!("true"));
2087        assert!(resolve_use_ws_trade(Some(&p), true));
2088        assert!(!resolve_use_ws_trade(Some(&p), false));
2089    }
2090
2091    #[rstest]
2092    fn test_execution_client_constructs_with_ws_trade_enabled() {
2093        let factory = KrakenExecutionClientFactory::new();
2094        let config = KrakenExecutionClientConfig {
2095            product_type: KrakenProductType::Spot,
2096            use_ws_trade: true,
2097            ws_request_timeout_secs: 7,
2098            ..Default::default()
2099        };
2100        let cache = Rc::new(RefCell::new(Cache::default()));
2101        let clock = Rc::new(RefCell::new(VirtualClock::new()));
2102
2103        let result = factory.create(
2104            TraderId::from("TRADER-001"),
2105            "KRAKEN-WS",
2106            &config,
2107            cache.into(),
2108            clock,
2109        );
2110        assert!(result.is_ok(), "construction failed: {:?}", result.err());
2111    }
2112}