nautilus_kraken/execution/

futures.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Kraken Futures execution client implementation.

use std::{
    future::Future,
    sync::Arc,
    time::{Duration, Instant},
};

use anyhow::Context;
use async_trait::async_trait;
use jiff::Timestamp;
use nautilus_common::{
    cache::InstrumentLookupError,
    clients::ExecutionClient,
    live::runner::get_exec_event_sender,
    messages::execution::{
        BatchCancelOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
        GenerateOrderStatusReport, GenerateOrderStatusReports, GeneratePositionStatusReports,
        ModifyOrder, QueryAccount, QueryOrder, SubmitOrder, SubmitOrderList,
    },
};
use nautilus_core::{
    AtomicMap, Params, UnixNanos,
    time::{AtomicTime, get_atomic_clock_realtime},
};
use nautilus_live::{
    ExecutionClientCore, ExecutionEventEmitter, SocketControl, execution::failure::CommandFailure,
    task::TaskGroup,
};
use nautilus_model::{
    accounts::AccountAny,
    enums::{AccountType, OmsType, OrderStatus, OrderType},
    identifiers::{
        AccountId, ClientId, ClientOrderId, InstrumentId, StrategyId, Venue, VenueOrderId,
    },
    instruments::{Instrument, InstrumentAny},
    orders::{Order, OrderAny},
    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
    types::{AccountBalance, MarginBalance, Quantity},
};
use rust_decimal::Decimal;
use tokio_util::sync::CancellationToken;

use super::{
    command_failure_from_cancel_error, command_failure_from_futures_batch_error,
    command_failure_from_futures_batch_item, command_failure_from_modify_error,
    command_failure_from_submit_error,
};
use crate::{
    common::{
        consts::KRAKEN_VENUE,
        credential::KrakenCredential,
        enums::{KrakenApiResult, KrakenProductType, KrakenSendStatus, product_type_from_symbol},
        parse::truncate_cl_ord_id,
    },
    config::KrakenExecutionClientConfig,
    http::{
        KrakenFuturesHttpClient,
        futures::{
            client::KRAKEN_FUTURES_DEFAULT_RATE_LIMIT_PER_SECOND, models::FuturesBatchCancelStatus,
            query::KrakenFuturesBatchCancelItem,
        },
    },
    websocket::{
        dispatch::{self, OrderIdentity, WsDispatchState},
        futures::{client::KrakenFuturesWebSocketClient, messages::KrakenFuturesWsMessage},
    },
};

const FUTURES_BATCH_CANCEL_LIMIT: usize = 50;

/// Maximum order IDs per `/orders/status` request for Kraken Futures API.
const FUTURES_ORDERS_STATUS_LIMIT: usize = 50;

/// Kraken Futures execution client.
///
/// Provides order management, account operations, and position management
/// for Kraken Futures markets.
#[allow(dead_code)]
#[derive(Debug)]
pub struct KrakenFuturesExecutionClient {
    core: ExecutionClientCore,
    clock: &'static AtomicTime,
    config: KrakenExecutionClientConfig,
    emitter: ExecutionEventEmitter,
    http: KrakenFuturesHttpClient,
    ws: KrakenFuturesWebSocketClient,
    cancellation_token: CancellationToken,
    session_tasks: TaskGroup,
    pending_tasks: TaskGroup,
    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
    truncated_id_map: Arc<AtomicMap<String, ClientOrderId>>,
    order_instrument_map: Arc<AtomicMap<String, InstrumentId>>,
    venue_client_map: Arc<AtomicMap<String, ClientOrderId>>,
    venue_order_qty: Arc<AtomicMap<String, Quantity>>,
    ws_dispatch_state: Arc<WsDispatchState>,
}

impl KrakenFuturesExecutionClient {
    /// Creates a new [`KrakenFuturesExecutionClient`].
    pub fn new(
        core: ExecutionClientCore,
        config: KrakenExecutionClientConfig,
    ) -> anyhow::Result<Self> {
        let clock = get_atomic_clock_realtime();
        let emitter = ExecutionEventEmitter::new(
            clock,
            core.trader_id,
            core.account_id,
            AccountType::Margin,
            None,
        );

        let session_tasks = TaskGroup::new();
        let cancellation_token = session_tasks.cancellation_token();
        let pending_tasks = TaskGroup::new();
        let api_key = config.api_key.expose_secret().to_owned();
        let api_secret = config.api_secret.expose_secret().to_owned();
        let proxy_url = config
            .proxy_url
            .as_ref()
            .map(|value| value.expose_secret().to_owned());

        let http = KrakenFuturesHttpClient::with_credentials(
            api_key.clone(),
            api_secret.clone(),
            config.environment,
            config.base_url.clone(),
            config.timeout_secs,
            Some(config.max_retries),
            None,
            None,
            proxy_url.clone(),
            config
                .max_requests_per_second
                .unwrap_or(KRAKEN_FUTURES_DEFAULT_RATE_LIMIT_PER_SECOND),
        )?;

        let credential = KrakenCredential::new(api_key, api_secret);
        let ws = KrakenFuturesWebSocketClient::with_credentials(
            config.ws_url(),
            config.heartbeat_interval_secs,
            Some(credential),
            config.auth_timeout_secs,
            config.transport_backend,
            proxy_url,
        )
        .with_socket_control(SocketControl::new(
            core.client_id,
            Some(*KRAKEN_VENUE),
            "kraken-futures-user-streams",
        ));

        Ok(Self {
            core,
            clock,
            config,
            emitter,
            http,
            ws,
            cancellation_token,
            session_tasks,
            pending_tasks,
            instruments: Arc::new(AtomicMap::new()),
            truncated_id_map: Arc::new(AtomicMap::new()),
            order_instrument_map: Arc::new(AtomicMap::new()),
            venue_client_map: Arc::new(AtomicMap::new()),
            venue_order_qty: Arc::new(AtomicMap::new()),
            ws_dispatch_state: Arc::new(WsDispatchState::new()),
        })
    }

    fn register_order_identity(&self, order: &OrderAny) {
        self.ws_dispatch_state.register_identity(
            order.client_order_id(),
            OrderIdentity {
                strategy_id: order.strategy_id(),
                instrument_id: order.instrument_id(),
                order_side: order.order_side(),
                order_type: order.order_type(),
                quantity: order.quantity(),
            },
        );
    }

    /// Returns a reference to the clock.
    #[must_use]
    pub fn clock(&self) -> &'static AtomicTime {
        self.clock
    }

    /// Returns a reference to the event emitter.
    #[must_use]
    pub fn emitter(&self) -> &ExecutionEventEmitter {
        &self.emitter
    }

    fn spawn_task<F>(&self, description: &'static str, fut: F)
    where
        F: Future<Output = anyhow::Result<()>> + Send + 'static,
    {
        let future = async move {
            if let Err(e) = fut.await {
                log::warn!("{description} failed: {e:?}");
            }
        };

        if let Err(e) = self.pending_tasks.spawn(future) {
            log::warn!("Skipping Kraken Futures {description} after shutdown began: {e}");
        }
    }

    async fn finish_tasks(&self) -> anyhow::Result<()> {
        let (session_result, pending_result) = tokio::join!(
            self.session_tasks
                .finish_shutdown(Duration::from_secs(1), Duration::from_secs(2)),
            self.pending_tasks
                .finish_shutdown(Duration::from_secs(2), Duration::from_secs(2)),
        );
        session_result.context("failed to finish Kraken Futures execution session tasks")?;
        pending_result.context("failed to finish Kraken Futures execution command tasks")?;
        Ok(())
    }

    async fn prepare_task_groups(&mut self) -> anyhow::Result<()> {
        if !self.session_tasks.is_open() || !self.pending_tasks.is_open() {
            self.session_tasks.begin_shutdown();
            self.pending_tasks.begin_shutdown();
            self.finish_tasks().await?;
            self.session_tasks
                .start_generation()
                .context("failed to start Kraken Futures execution session task generation")?;
            self.pending_tasks
                .start_generation()
                .context("failed to start Kraken Futures execution command task generation")?;
            self.cancellation_token = self.session_tasks.cancellation_token();
        }
        Ok(())
    }

    async fn teardown_partial_connect(&mut self) -> anyhow::Result<()> {
        self.http.cancel_all_requests();
        self.pending_tasks.begin_shutdown();
        let ws_result = self.ws.close().await;
        self.session_tasks.begin_shutdown();
        let tasks_result = self.finish_tasks().await;
        self.core.set_disconnected();
        tasks_result?;
        Ok(ws_result?)
    }

    fn submit_single_order(&self, order: &OrderAny, task_name: &'static str) {
        if order.is_closed() {
            log::warn!(
                "Cannot submit closed order: client_order_id={}",
                order.client_order_id()
            );
            return;
        }

        let account_id = self.core.account_id;
        let client_order_id = order.client_order_id();
        let strategy_id = order.strategy_id();
        let instrument_id = order.instrument_id();
        let order_side = order.order_side();
        let order_type = order.order_type();
        let quantity = order.quantity();
        let time_in_force = order.time_in_force();
        let price = order.price();
        let trigger_price = order.trigger_price();
        let trigger_type = order.trigger_type();
        let is_reduce_only = order.is_reduce_only();
        let is_post_only = order.is_post_only();

        log::debug!("OrderSubmitted: client_order_id={client_order_id}");
        self.register_order_identity(order);
        self.emitter.emit_order_submitted(order);

        let kraken_cl_ord_id = truncate_cl_ord_id(&client_order_id);

        if kraken_cl_ord_id != client_order_id.as_str() {
            self.truncated_id_map
                .insert(kraken_cl_ord_id, client_order_id);
        }

        let http = self.http.clone();
        let emitter = self.emitter.clone();
        let clock = self.clock;
        let dispatch_state = self.ws_dispatch_state.clone();

        self.spawn_task(task_name, async move {
            let result = http
                .submit_order(
                    account_id,
                    instrument_id,
                    client_order_id,
                    order_side,
                    order_type,
                    quantity,
                    time_in_force,
                    price,
                    trigger_price,
                    trigger_type,
                    is_reduce_only,
                    is_post_only,
                )
                .await;

            match result {
                Ok(_) => {}
                Err(e) => match command_failure_from_submit_error(&e) {
                    CommandFailure::Ambiguous(reason) => {
                        log::warn!(
                            "{task_name} outcome is ambiguous for client_order_id={client_order_id}: {reason}"
                        );
                    }
                    CommandFailure::NotSent(reason) | CommandFailure::VenueRejected(reason) => {
                        let ts_event = clock.get_time_ns();
                        let error_msg = format!("{task_name} error: {reason}");
                        let due_post_only = error_msg.contains("POST_ONLY_REJECTED");
                        dispatch_state.cleanup_terminal(&client_order_id);
                        emitter.emit_order_rejected_event(
                            strategy_id,
                            instrument_id,
                            client_order_id,
                            &error_msg,
                            ts_event,
                            due_post_only,
                        );
                    }
                },
            }
            Ok(())
        });
    }

    fn cancel_single_order(&self, cmd: &CancelOrder) {
        let account_id = self.core.account_id;
        let client_order_id = cmd.client_order_id;
        let venue_order_id = cmd.venue_order_id;
        let strategy_id = cmd.strategy_id;
        let instrument_id = cmd.instrument_id;

        log::debug!(
            "Canceling order: venue_order_id={venue_order_id:?}, client_order_id={client_order_id}"
        );

        let http = self.http.clone();
        let emitter = self.emitter.clone();
        let clock = self.clock;

        self.spawn_task("cancel_order", async move {
            if let Err(failure) = cancel_order_for_futures(
                &http,
                account_id,
                instrument_id,
                Some(client_order_id),
                venue_order_id,
            )
            .await
            {
                handle_cancel_failure(
                    &emitter,
                    clock,
                    strategy_id,
                    instrument_id,
                    client_order_id,
                    venue_order_id,
                    failure,
                );
            }
            Ok(())
        });
    }

    fn spawn_message_handler(&mut self) -> anyhow::Result<()> {
        let mut rx = self
            .ws
            .take_output_rx()
            .context("Failed to take futures WebSocket output receiver")?;
        let emitter = self.emitter.clone();
        let instruments = self.instruments.clone();
        let truncated_id_map = self.truncated_id_map.clone();
        let order_instrument_map = self.order_instrument_map.clone();
        let venue_client_map = self.venue_client_map.clone();
        let venue_order_qty = self.venue_order_qty.clone();
        let dispatch_state = self.ws_dispatch_state.clone();
        let account_id = self.core.account_id;
        let clock = self.clock;
        let cancellation_token = self.cancellation_token.clone();

        let future = async move {
            loop {
                tokio::select! {
                    () = cancellation_token.cancelled() => {
                        log::debug!("Futures execution message handler cancelled");
                        break;
                    }
                    msg = rx.recv() => {
                        match msg {
                            Some(ws_msg) => {
                                Self::handle_ws_message(
                                    ws_msg,
                                    &emitter,
                                    &dispatch_state,
                                    &instruments,
                                    &truncated_id_map,
                                    &order_instrument_map,
                                    &venue_client_map,
                                    &venue_order_qty,
                                    account_id,
                                    clock,
                                );
                            }
                            None => {
                                log::debug!("Futures execution WebSocket stream ended");
                                break;
                            }
                        }
                    }
                }
            }
        };

        self.session_tasks
            .spawn(future)
            .context("failed to register Kraken Futures execution stream task")
    }

    #[expect(clippy::too_many_arguments)]
    fn handle_ws_message(
        msg: KrakenFuturesWsMessage,
        emitter: &ExecutionEventEmitter,
        dispatch_state: &Arc<WsDispatchState>,
        instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
        truncated_id_map: &Arc<AtomicMap<String, ClientOrderId>>,
        order_instrument_map: &Arc<AtomicMap<String, InstrumentId>>,
        venue_client_map: &Arc<AtomicMap<String, ClientOrderId>>,
        venue_order_qty: &Arc<AtomicMap<String, Quantity>>,
        account_id: AccountId,
        clock: &'static AtomicTime,
    ) {
        let ts_init = clock.get_time_ns();

        match msg {
            KrakenFuturesWsMessage::OpenOrdersDelta(delta) => {
                dispatch::futures::open_orders_delta(
                    &delta,
                    dispatch_state,
                    emitter,
                    instruments,
                    truncated_id_map,
                    order_instrument_map,
                    venue_client_map,
                    venue_order_qty,
                    account_id,
                    ts_init,
                );
            }
            KrakenFuturesWsMessage::OpenOrdersCancel(cancel) => {
                dispatch::futures::open_orders_cancel(
                    &cancel,
                    dispatch_state,
                    emitter,
                    truncated_id_map,
                    order_instrument_map,
                    venue_client_map,
                    venue_order_qty,
                    account_id,
                    ts_init,
                );
            }
            KrakenFuturesWsMessage::FillsDelta(fills_delta) => {
                dispatch::futures::fills_delta(
                    &fills_delta,
                    dispatch_state,
                    emitter,
                    instruments,
                    truncated_id_map,
                    venue_client_map,
                    account_id,
                    ts_init,
                );
            }
            KrakenFuturesWsMessage::Challenge(challenge) => {
                log::debug!("Received challenge: length={}", challenge.len());
            }
            KrakenFuturesWsMessage::Reconnected => {
                log::info!("Futures execution WebSocket reconnected");
            }
            KrakenFuturesWsMessage::Ticker(_)
            | KrakenFuturesWsMessage::Trade(_)
            | KrakenFuturesWsMessage::BookSnapshot(_)
            | KrakenFuturesWsMessage::BookDelta(_) => {}
        }
    }

    async fn await_account_registered(&self, timeout_secs: f64) -> anyhow::Result<()> {
        let account_id = self.core.account_id;

        if self.core.cache().account(&account_id).is_some() {
            log::info!("Account {account_id} registered");
            return Ok(());
        }

        let start = Instant::now();
        let timeout = Duration::from_secs_f64(timeout_secs);
        let interval = Duration::from_millis(10);

        loop {
            tokio::time::sleep(interval).await;

            if self.core.cache().account(&account_id).is_some() {
                log::info!("Account {account_id} registered");
                return Ok(());
            }

            if start.elapsed() >= timeout {
                anyhow::bail!(
                    "Timeout waiting for account {account_id} to be registered after {timeout_secs}s"
                );
            }
        }
    }

    fn modify_single_order(&self, cmd: &ModifyOrder) {
        let client_order_id = cmd.client_order_id;
        let venue_order_id = cmd.venue_order_id;
        let strategy_id = cmd.strategy_id;
        let instrument_id = cmd.instrument_id;
        let quantity = cmd.quantity;
        let price = cmd.price;

        log::debug!(
            "Modifying order: venue_order_id={venue_order_id:?}, client_order_id={client_order_id}"
        );

        let http = self.http.clone();
        let emitter = self.emitter.clone();
        let clock = self.clock;

        self.spawn_task("modify_order", async move {
            match http
                .modify_order(
                    instrument_id,
                    Some(client_order_id),
                    venue_order_id,
                    quantity,
                    price,
                    None,
                )
                .await
            {
                Ok(_) => {}
                Err(e) => match command_failure_from_modify_error(&e) {
                    CommandFailure::Ambiguous(reason) => {
                        log::warn!(
                            "modify_order outcome is ambiguous for client_order_id={client_order_id}: {reason}"
                        );
                    }
                    CommandFailure::NotSent(reason) | CommandFailure::VenueRejected(reason) => {
                        let ts_event = clock.get_time_ns();
                        emitter.emit_order_modify_rejected_event(
                            strategy_id,
                            instrument_id,
                            client_order_id,
                            venue_order_id,
                            &format!("modify-order error: {reason}"),
                            ts_event,
                        );
                    }
                },
            }
            Ok(())
        });
    }
}

#[async_trait(?Send)]
impl ExecutionClient for KrakenFuturesExecutionClient {
    fn is_connected(&self) -> bool {
        self.core.is_connected()
    }

    fn client_id(&self) -> ClientId {
        self.core.client_id
    }

    fn account_id(&self) -> AccountId {
        self.core.account_id
    }

    fn venue(&self) -> Venue {
        *KRAKEN_VENUE
    }

    fn oms_type(&self) -> OmsType {
        self.core.oms_type
    }

    fn get_account(&self) -> Option<AccountAny> {
        self.core.cache().account_owned(&self.core.account_id)
    }

    fn generate_account_state(
        &self,
        balances: Vec<AccountBalance>,
        margins: Vec<MarginBalance>,
        reported: bool,
        ts_event: UnixNanos,
        info: Option<Params>,
    ) -> anyhow::Result<()> {
        self.emitter
            .emit_account_state(balances, margins, reported, ts_event, info);
        Ok(())
    }

    fn start(&mut self) -> anyhow::Result<()> {
        if self.core.is_started() {
            return Ok(());
        }

        self.emitter.set_sender(get_exec_event_sender());
        self.core.set_started();

        log::info!(
            "Started: client_id={}, account_id={}, product_type=Futures, environment={:?}",
            self.core.client_id,
            self.core.account_id,
            self.config.environment
        );
        Ok(())
    }

    fn stop(&mut self) -> anyhow::Result<()> {
        if self.core.is_stopped() {
            return Ok(());
        }

        self.http.cancel_all_requests();
        self.session_tasks.begin_shutdown();
        self.pending_tasks.begin_shutdown();
        self.ws.begin_shutdown();
        self.core.set_stopped();
        self.core.set_disconnected();
        log::info!("Stopped: client_id={}", self.core.client_id);
        Ok(())
    }

    async fn connect(&mut self) -> anyhow::Result<()> {
        if self.core.is_connected() && self.session_tasks.is_open() && self.pending_tasks.is_open()
        {
            return Ok(());
        }

        self.http.reset_cancellation_token();
        self.prepare_task_groups().await?;

        if !self.core.instruments_initialized() {
            let instruments = self
                .http
                .request_instruments()
                .await
                .context("Failed to load Kraken futures instruments")?;
            log::debug!("Loaded {} Futures instruments", instruments.len());
            self.http.cache_instruments(&instruments);
            self.core.set_instruments_initialized();
        }

        self.instruments.rcu(|m| {
            for instrument in self.http.instruments_cache.load().values() {
                m.insert(instrument.id(), instrument.clone());
            }
        });

        let session_result = async {
            self.ws
                .connect()
                .await
                .context("Failed to connect futures WebSocket")?;
            self.ws
                .wait_until_active(10.0)
                .await
                .context("Futures WebSocket failed to become active")?;

            self.ws
                .authenticate()
                .await
                .context("Failed to authenticate futures WebSocket")?;

            let account_state = self
                .http
                .request_account_state(self.core.account_id)
                .await
                .context("Failed to request Kraken futures account state")?;

            if !account_state.balances.is_empty() {
                log::debug!(
                    "Received account state with {} balance(s)",
                    account_state.balances.len()
                );
            }
            self.emitter.send_account_state(account_state);
            self.await_account_registered(30.0).await?;

            self.spawn_message_handler()?;

            self.ws
                .subscribe_executions()
                .await
                .context("Failed to subscribe to executions")?;

            log::debug!("Futures WebSocket authenticated and subscribed to executions");

            Ok::<(), anyhow::Error>(())
        }
        .await;

        if let Err(e) = session_result {
            if let Err(teardown_error) = self.teardown_partial_connect().await {
                return Err(e.context(format!(
                    "Kraken Futures execution startup teardown failed: {teardown_error}"
                )));
            }
            return Err(e);
        }

        self.core.set_connected();
        log::info!("Connected: client_id={}", self.core.client_id);
        Ok(())
    }

    async fn disconnect(&mut self) -> anyhow::Result<()> {
        self.teardown_partial_connect().await?;
        log::info!("Disconnected: client_id={}", self.core.client_id);
        Ok(())
    }

    async fn generate_order_status_report(
        &self,
        cmd: &GenerateOrderStatusReport,
    ) -> anyhow::Result<Option<OrderStatusReport>> {
        log::debug!(
            "Generating order status report: venue_order_id={:?}, client_order_id={:?}",
            cmd.venue_order_id,
            cmd.client_order_id
        );

        let account_id = self.core.account_id;
        let reports = self
            .http
            .request_order_status_reports(account_id, None, None, None, false)
            .await?;

        // Match by venue_order_id or client_order_id (comparing truncated form
        // since Kraken stores the truncated cl_ord_id for long IDs)
        let matched = reports.into_iter().find(|r| {
            cmd.venue_order_id
                .is_some_and(|id| r.venue_order_id.as_str() == id.as_str())
                || cmd.client_order_id.is_some_and(|id| {
                    r.client_order_id
                        .as_ref()
                        .is_some_and(|r_id| r_id.as_str() == truncate_cl_ord_id(&id))
                })
        });

        if matched.is_some() {
            return Ok(matched);
        }

        let Some(order) = self.get_cached_order_for_status_command(cmd) else {
            return Ok(None);
        };

        // Held orders never appear on /openorders; query the 5-second window
        let order_ids: Vec<String> = cmd
            .venue_order_id
            .or(order.venue_order_id())
            .map(|id| id.to_string())
            .into_iter()
            .collect();
        let cli_ord_ids: Vec<String> = cmd
            .client_order_id
            .map(|id| truncate_cl_ord_id(&id))
            .into_iter()
            .collect();

        let recent_reports = self
            .http
            .request_orders_status_reports(account_id, &order_ids, &cli_ord_ids)
            .await?;

        let matched_recent = recent_reports
            .iter()
            .find(|report| {
                cmd.venue_order_id
                    .is_some_and(|id| report.venue_order_id == id)
                    || cmd.client_order_id.is_some_and(|id| {
                        report
                            .client_order_id
                            .as_ref()
                            .is_some_and(|report_id| report_id.as_str() == truncate_cl_ord_id(&id))
                    })
            })
            .cloned();

        // Window filled reports have no avg_px; price them from fills below
        if matched_recent
            .as_ref()
            .is_some_and(|report| report.order_status != OrderStatus::Filled)
        {
            return Ok(matched_recent);
        }

        let now = Timestamp::now();
        let start = now - Duration::from_secs(5 * 60);
        let fills = self
            .http
            .request_fill_reports(
                account_id,
                Some(order.instrument_id()),
                Some(start),
                Some(now),
            )
            .await?;

        match (
            synthesize_filled_order_status_report(cmd, &order, &fills),
            matched_recent,
        ) {
            (Some(report), _) => Ok(Some(report)),
            // Unpriced filled reports would close at the order price
            (None, Some(_)) => anyhow::bail!(
                "Order {} fully executed in the orders-status window without visible \
                 fills; deferring until the fills feed prices it",
                order.client_order_id(),
            ),
            (None, None) => Ok(None),
        }
    }

    async fn generate_order_status_reports(
        &self,
        cmd: &GenerateOrderStatusReports,
    ) -> anyhow::Result<Vec<OrderStatusReport>> {
        log::debug!(
            "Generating order status reports: instrument_id={:?}, open_only={}",
            cmd.instrument_id,
            cmd.open_only
        );

        let account_id = self.core.account_id;
        let start = cmd.start.map(Timestamp::from);
        let end = cmd.end.map(Timestamp::from);
        let mut reports = self
            .http
            .request_order_status_reports(account_id, cmd.instrument_id, start, end, cmd.open_only)
            .await?;

        if cmd.open_only {
            let extension = self
                .reports_for_open_orders_absent_from_venue(account_id, cmd.instrument_id, &reports)
                .await?;

            for report in extension {
                if report.order_status == OrderStatus::Filled {
                    log::debug!(
                        "Deferring fully executed order {} from the bulk response: fills-paired \
                         pricing applies",
                        report.venue_order_id,
                    );
                    continue;
                }

                reports.push(report);
            }
        }

        Ok(reports)
    }

    async fn generate_fill_reports(
        &self,
        cmd: GenerateFillReports,
    ) -> anyhow::Result<Vec<FillReport>> {
        log::debug!(
            "Generating fill reports: instrument_id={:?}",
            cmd.instrument_id
        );

        let account_id = self.core.account_id;
        let start = cmd.start.map(Timestamp::from);
        let end = cmd.end.map(Timestamp::from);
        let mut reports = self
            .http
            .request_fill_reports(account_id, cmd.instrument_id, start, end)
            .await?;

        if let Some(venue_order_id) = cmd.venue_order_id {
            reports.retain(|report| report.venue_order_id == venue_order_id);
        }

        Ok(reports)
    }

    async fn generate_position_status_reports(
        &self,
        cmd: &GeneratePositionStatusReports,
    ) -> anyhow::Result<Vec<PositionStatusReport>> {
        log::debug!(
            "Generating position status reports: instrument_id={:?}",
            cmd.instrument_id
        );

        let account_id = self.core.account_id;
        self.http
            .request_position_status_reports(account_id, cmd.instrument_id)
            .await
    }

    async fn generate_mass_status(
        &self,
        lookback_mins: Option<u64>,
    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
        log::debug!("Generating mass status: lookback_mins={lookback_mins:?}");

        let ts_init = self.clock.get_time_ns();
        let start = lookback_mins.map(|mins| Timestamp::now() - Duration::from_secs(mins * 60));
        let account_id = self.core.account_id;

        let mut order_reports = self
            .http
            .request_order_status_reports(account_id, None, start, None, true)
            .await?;
        let extension = self
            .reports_for_open_orders_absent_from_venue(account_id, None, &order_reports)
            .await?;

        // Snapshot recon would infer uncovered fills at the order price
        for report in extension {
            if report.order_status == OrderStatus::Filled {
                log::debug!(
                    "Deferring fully executed order {} from mass status: fills-paired \
                     pricing applies",
                    report.venue_order_id,
                );
                continue;
            }

            order_reports.push(report);
        }

        let fill_reports = self
            .http
            .request_fill_reports(account_id, None, start, None)
            .await?;
        let position_reports = self
            .http
            .request_position_status_reports(account_id, None)
            .await?;

        let mut mass_status = ExecutionMassStatus::new(
            self.core.client_id,
            self.core.account_id,
            *KRAKEN_VENUE,
            ts_init,
            None,
        );
        mass_status.add_order_reports(order_reports);
        mass_status.add_fill_reports(fill_reports);
        mass_status.add_position_reports(position_reports);

        Ok(Some(mass_status))
    }

    fn query_account(&self, cmd: QueryAccount) -> anyhow::Result<()> {
        log::debug!("Querying account: {cmd}");

        let account_id = self.core.account_id;
        let http = self.http.clone();
        let emitter = self.emitter.clone();

        self.spawn_task("query_account", async move {
            let account_state = http.request_account_state(account_id).await?;
            emitter.emit_account_state(
                account_state.balances.clone(),
                account_state.margins.clone(),
                account_state.is_reported,
                account_state.ts_event,
                account_state.info,
            );
            Ok(())
        });

        Ok(())
    }

    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
        log::debug!("Querying order: {cmd}");

        let venue_order_id = cmd
            .venue_order_id
            .context("venue_order_id required for query_order")?;
        let account_id = self.core.account_id;
        let http = self.http.clone();
        let emitter = self.emitter.clone();

        self.spawn_task("query_order", async move {
            let reports = http
                .request_order_status_reports(account_id, None, None, None, true)
                .await
                .context("Failed to query order")?;

            if let Some(report) = reports
                .into_iter()
                .find(|r| r.venue_order_id == venue_order_id)
            {
                emitter.send_order_status_report(report);
            }
            Ok(())
        });

        Ok(())
    }

    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
        let order = self.core.cache().try_order_owned(&cmd.client_order_id)?;
        self.submit_single_order(&order, "submit_order");
        Ok(())
    }

    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
        let orders = self.core.get_orders_for_list(&cmd.order_list)?;

        log::debug!(
            "Submitting order list: order_list_id={}, count={}",
            cmd.order_list.id,
            orders.len()
        );

        let mut order_tuples = Vec::with_capacity(orders.len());
        let mut order_meta = Vec::with_capacity(orders.len());

        for order in &orders {
            if order.is_closed() {
                log::warn!(
                    "Cannot submit closed order: client_order_id={}",
                    order.client_order_id()
                );
                continue;
            }

            // Kraken batch endpoint only supports limit and stop orders,
            // submit market orders individually
            if order.order_type() == OrderType::Market {
                self.submit_single_order(order, "submit_order_list");
                continue;
            }

            let client_order_id = order.client_order_id();
            let kraken_cl_ord_id = truncate_cl_ord_id(&client_order_id);

            if kraken_cl_ord_id != client_order_id.as_str() {
                self.truncated_id_map
                    .insert(kraken_cl_ord_id, client_order_id);
            }

            self.register_order_identity(order);
            self.emitter.emit_order_submitted(order);

            order_tuples.push((
                order.instrument_id(),
                client_order_id,
                order.order_side(),
                order.order_type(),
                order.quantity(),
                order.time_in_force(),
                order.price(),
                order.trigger_price(),
                order.trigger_type(),
                order.is_reduce_only(),
                order.is_post_only(),
            ));

            order_meta.push((order.strategy_id(), order.instrument_id(), client_order_id));
        }

        if order_tuples.is_empty() {
            return Ok(());
        }

        let http = self.http.clone();
        let emitter = self.emitter.clone();
        let clock = self.clock;
        let dispatch_state = self.ws_dispatch_state.clone();

        self.spawn_task("submit_order_list", async move {
            let results = http.send_order_batches(order_tuples).await;
            for (result, (strategy_id, instrument_id, client_order_id)) in
                results.into_iter().zip(&order_meta)
            {
                let outcome = match result {
                    Ok(item) => command_failure_from_futures_batch_item(item),
                    Err(e) => Err(command_failure_from_futures_batch_error(&e)),
                };

                match outcome {
                    Ok(()) => {}
                    Err(CommandFailure::Ambiguous(reason)) => {
                        log::warn!(
                            "submit_order_list outcome is ambiguous for client_order_id={client_order_id}: {reason}"
                        );
                    }
                    Err(
                        CommandFailure::NotSent(reason)
                        | CommandFailure::VenueRejected(reason),
                    ) => {
                        let ts_event = clock.get_time_ns();
                        let error_msg =
                            format!("submit_order_list batch item rejected: {reason}");
                        dispatch_state.cleanup_terminal(client_order_id);
                        emitter.emit_order_rejected_event(
                            *strategy_id,
                            *instrument_id,
                            *client_order_id,
                            &error_msg,
                            ts_event,
                            reason == "postWouldExecute",
                        );
                    }
                }
            }
            Ok(())
        });

        Ok(())
    }

    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
        self.modify_single_order(&cmd);
        Ok(())
    }

    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
        self.cancel_single_order(&cmd);
        Ok(())
    }

    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;

        if cmd.order_side.is_none() {
            log::debug!("Canceling all orders: instrument_id={instrument_id} (bulk)");

            let http = self.http.clone();
            let symbol = instrument_id.symbol.to_string();

            self.spawn_task("cancel_all_orders", async move {
                match http.inner.cancel_all_orders(Some(symbol)).await {
                    Ok(response) => {
                        if response.result != KrakenApiResult::Success
                            && response.cancel_status.cancelled_orders.is_empty()
                        {
                            log::warn!(
                                "Cancel-all failed without per-order results, awaiting reconciliation: status={}",
                                response.cancel_status.status
                            );
                        }
                    }
                    Err(e) => match command_failure_from_cancel_error(e) {
                        CommandFailure::NotSent(reason) => {
                            log::warn!("Cancel-all failed local validation: {reason}");
                        }
                        CommandFailure::Ambiguous(reason)
                        | CommandFailure::VenueRejected(reason) => {
                            log::warn!(
                                "Cancel-all ambiguous failure, awaiting reconciliation: {reason}"
                            );
                        }
                    },
                }
                Ok(())
            });

            return Ok(());
        }

        log::debug!(
            "Canceling all orders: instrument_id={instrument_id}, side={:?}",
            cmd.order_side
        );

        let orders_to_cancel: Vec<_> = {
            let cache = self.core.cache();
            let open_orders = cache.orders_open(None, Some(&instrument_id), None, None, None);

            open_orders
                .into_iter()
                .filter(|order| Some(order.order_side()) == cmd.order_side)
                .filter_map(|order| {
                    Some((
                        order.venue_order_id()?,
                        order.client_order_id(),
                        order.instrument_id(),
                        order.strategy_id(),
                    ))
                })
                .collect()
        };

        let account_id = self.core.account_id;

        for (venue_order_id, client_order_id, order_instrument_id, strategy_id) in orders_to_cancel
        {
            let http = self.http.clone();
            let emitter = self.emitter.clone();
            let clock = self.clock;

            self.spawn_task("cancel_order_by_side", async move {
                if let Err(failure) = cancel_order_for_futures(
                    &http,
                    account_id,
                    order_instrument_id,
                    Some(client_order_id),
                    Some(venue_order_id),
                )
                .await
                {
                    handle_cancel_failure(
                        &emitter,
                        clock,
                        strategy_id,
                        order_instrument_id,
                        client_order_id,
                        Some(venue_order_id),
                        failure,
                    );
                }
                Ok(())
            });
        }

        Ok(())
    }

    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
        log::debug!(
            "Batch canceling orders: instrument_id={}, count={}",
            cmd.instrument_id,
            cmd.cancels.len()
        );

        let http = self.http.clone();
        let emitter = self.emitter.clone();
        let clock = self.clock;
        let cancels = cmd.cancels;

        self.spawn_task("batch_cancel_orders", async move {
            batch_cancel_orders_for_futures(&http, &emitter, clock, &cancels).await;
            Ok(())
        });

        Ok(())
    }
}

#[derive(Debug, Clone)]
struct CancelRequestContext {
    strategy_id: StrategyId,
    instrument_id: InstrumentId,
    client_order_id: ClientOrderId,
    truncated_client_order_id: String,
    venue_order_id: Option<VenueOrderId>,
}

async fn cancel_order_for_futures(
    http: &KrakenFuturesHttpClient,
    _account_id: AccountId,
    instrument_id: InstrumentId,
    client_order_id: Option<ClientOrderId>,
    venue_order_id: Option<VenueOrderId>,
) -> Result<(), CommandFailure> {
    http.get_cached_instrument(&instrument_id.symbol.inner())
        .ok_or_else(|| {
            CommandFailure::not_sent(InstrumentLookupError::not_found(instrument_id).to_string())
        })?;

    let order_id = venue_order_id.as_ref().map(ToString::to_string);
    let cli_ord_id = client_order_id.as_ref().map(truncate_cl_ord_id);

    if order_id.is_none() && cli_ord_id.is_none() {
        return Err(CommandFailure::not_sent(
            "Either client_order_id or venue_order_id must be provided",
        ));
    }

    let response = http
        .inner
        .cancel_order(order_id, cli_ord_id)
        .await
        .map_err(command_failure_from_cancel_error)?;

    if response.result != KrakenApiResult::Success
        || response.cancel_status.status != KrakenSendStatus::Cancelled
    {
        return Err(CommandFailure::venue_rejected(format!(
            "cancel-order rejected: status={}",
            response.cancel_status.status
        )));
    }

    Ok(())
}

async fn batch_cancel_orders_for_futures(
    http: &KrakenFuturesHttpClient,
    emitter: &ExecutionEventEmitter,
    clock: &'static AtomicTime,
    cancels: &[CancelOrder],
) {
    let mut contexts = Vec::new();
    let mut items = Vec::new();

    for cancel in cancels {
        match batch_cancel_item_for_futures(http, cancel) {
            Ok((context, item)) => {
                contexts.push(context);
                items.push(item);
            }
            Err(CommandFailure::NotSent(reason)) => {
                log::warn!(
                    "Batch cancel command failed local validation for {}: {reason}",
                    cancel.client_order_id
                );
            }
            Err(CommandFailure::Ambiguous(reason) | CommandFailure::VenueRejected(reason)) => {
                log::warn!(
                    "Batch cancel command ambiguous failure for {}, awaiting reconciliation: {reason}",
                    cancel.client_order_id
                );
            }
        }
    }

    for (item_chunk, context_chunk) in items
        .chunks(FUTURES_BATCH_CANCEL_LIMIT)
        .zip(contexts.chunks(FUTURES_BATCH_CANCEL_LIMIT))
    {
        let response = match http
            .inner
            .cancel_order_items_batch(item_chunk.to_vec())
            .await
        {
            Ok(response) => response,
            Err(e) => {
                match command_failure_from_cancel_error(e) {
                    CommandFailure::NotSent(reason) => {
                        log::warn!("Batch cancel failed local validation: {reason}");
                    }
                    CommandFailure::Ambiguous(reason) | CommandFailure::VenueRejected(reason) => {
                        log::warn!(
                            "Batch cancel failed without per-order results, awaiting reconciliation: {reason}"
                        );
                    }
                }
                continue;
            }
        };

        if response.batch_status.is_empty() {
            if response.result != KrakenApiResult::Success {
                let reason = response.error.as_deref().unwrap_or("Unknown error");
                log::warn!(
                    "Batch cancel failed without per-order results, awaiting reconciliation: {reason}"
                );
            }
            continue;
        }

        if response.batch_status.len() != context_chunk.len() {
            log::warn!(
                "Batch cancel returned {} per-order result(s) for {} request(s); unmatched results await reconciliation",
                response.batch_status.len(),
                context_chunk.len()
            );
        }

        for (index, status) in response.batch_status.iter().enumerate() {
            let Some(cancel_status) = batch_cancel_status(status) else {
                log::warn!("Batch cancel result without status at index {index}");
                continue;
            };

            if cancel_status == KrakenSendStatus::Cancelled {
                continue;
            }

            let Some(context) = batch_cancel_context(status, context_chunk, index) else {
                log::warn!(
                    "Batch cancel rejected item without matching request context at index {index}: status={cancel_status}"
                );
                continue;
            };

            emitter.emit_order_cancel_rejected_event(
                context.strategy_id,
                context.instrument_id,
                context.client_order_id,
                context.venue_order_id,
                &format!("batch-cancel rejected: status={cancel_status}"),
                clock.get_time_ns(),
            );
        }
    }
}

fn batch_cancel_item_for_futures(
    http: &KrakenFuturesHttpClient,
    cancel: &CancelOrder,
) -> Result<(CancelRequestContext, KrakenFuturesBatchCancelItem), CommandFailure> {
    http.get_cached_instrument(&cancel.instrument_id.symbol.inner())
        .ok_or_else(|| {
            CommandFailure::not_sent(
                InstrumentLookupError::not_found(cancel.instrument_id).to_string(),
            )
        })?;

    let truncated_client_order_id = truncate_cl_ord_id(&cancel.client_order_id);
    let item = if let Some(venue_order_id) = cancel.venue_order_id {
        KrakenFuturesBatchCancelItem::from_order_id(venue_order_id.to_string())
    } else {
        KrakenFuturesBatchCancelItem::from_client_order_id(truncated_client_order_id.clone())
    };

    Ok((
        CancelRequestContext {
            strategy_id: cancel.strategy_id,
            instrument_id: cancel.instrument_id,
            client_order_id: cancel.client_order_id,
            truncated_client_order_id,
            venue_order_id: cancel.venue_order_id,
        },
        item,
    ))
}

fn batch_cancel_status(status: &FuturesBatchCancelStatus) -> Option<KrakenSendStatus> {
    status
        .cancel_status
        .as_ref()
        .map(|cancel_status| cancel_status.status)
        .or(status.status)
}

fn batch_cancel_context<'a>(
    status: &FuturesBatchCancelStatus,
    contexts: &'a [CancelRequestContext],
    index: usize,
) -> Option<&'a CancelRequestContext> {
    if let Some(order_id) = status.order_id.as_deref()
        && let Some(context) = contexts.iter().find(|context| {
            context
                .venue_order_id
                .is_some_and(|venue_order_id| venue_order_id.as_str() == order_id)
        })
    {
        return Some(context);
    }

    if let Some(cli_ord_id) = status.cli_ord_id.as_deref()
        && let Some(context) = contexts
            .iter()
            .find(|context| context.truncated_client_order_id == cli_ord_id)
    {
        return Some(context);
    }

    if index < contexts.len() && status.order_id.is_none() && status.cli_ord_id.is_none() {
        return contexts.get(index);
    }

    None
}

fn handle_cancel_failure(
    emitter: &ExecutionEventEmitter,
    clock: &'static AtomicTime,
    strategy_id: StrategyId,
    instrument_id: InstrumentId,
    client_order_id: ClientOrderId,
    venue_order_id: Option<VenueOrderId>,
    failure: CommandFailure,
) {
    match failure {
        CommandFailure::VenueRejected(reason) => {
            emitter.emit_order_cancel_rejected_event(
                strategy_id,
                instrument_id,
                client_order_id,
                venue_order_id,
                &reason,
                clock.get_time_ns(),
            );
        }
        CommandFailure::NotSent(reason) => {
            log::warn!("Cancel command failed local validation for {client_order_id}: {reason}");
        }
        CommandFailure::Ambiguous(reason) => {
            log::warn!(
                "Ambiguous cancel failure for {client_order_id}, awaiting reconciliation: {reason}"
            );
        }
    }
}

impl KrakenFuturesExecutionClient {
    /// Returns `/orders/status` reports for cached-open orders the given
    /// venue reports do not cover.
    ///
    /// A Maker Protection hold never reaches the book, so it is invisible to
    /// an open-orders snapshot while the venue still knows the order within
    /// its 5-second `/orders/status` window. Orders without a venue order ID
    /// yet (their submit acknowledgement is still inside the hold window) are
    /// queried by their truncated client order ID, the only venue handle that
    /// exists during the window.
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying request fails or the venue rejects
    /// it, so callers defer rather than treat the orders as missing.
    async fn reports_for_open_orders_absent_from_venue(
        &self,
        account_id: AccountId,
        instrument_id: Option<InstrumentId>,
        reported: &[OrderStatusReport],
    ) -> anyhow::Result<Vec<OrderStatusReport>> {
        let mut order_ids = Vec::new();
        let mut cli_ord_ids = Vec::new();

        {
            let cache = self.core.cache();
            for order in cache.orders_open(
                Some(&*KRAKEN_VENUE),
                instrument_id.as_ref(),
                None,
                None,
                None,
            ) {
                // Spot and Futures share the KRAKEN venue and one cache
                if product_type_from_symbol(order.instrument_id().symbol.inner().as_str())
                    != KrakenProductType::Futures
                {
                    continue;
                }

                match order.venue_order_id() {
                    Some(venue_order_id) => {
                        let already_reported = reported
                            .iter()
                            .any(|report| report.venue_order_id == venue_order_id);
                        if !already_reported {
                            order_ids.push(venue_order_id.to_string());
                        }
                    }
                    None => {
                        cli_ord_ids.push(truncate_cl_ord_id(&order.client_order_id()));
                    }
                }
            }
        }

        if order_ids.is_empty() && cli_ord_ids.is_empty() {
            return Ok(Vec::new());
        }

        log::debug!(
            "Resolving {} venue order ID(s) and {} client order ID(s) from the orders-status window",
            order_ids.len(),
            cli_ord_ids.len(),
        );

        let mut reports = Vec::new();
        for chunk in order_ids.chunks(FUTURES_ORDERS_STATUS_LIMIT) {
            reports.extend(
                self.http
                    .request_orders_status_reports(account_id, chunk, &[])
                    .await?,
            );
        }

        for chunk in cli_ord_ids.chunks(FUTURES_ORDERS_STATUS_LIMIT) {
            reports.extend(
                self.http
                    .request_orders_status_reports(account_id, &[], chunk)
                    .await?,
            );
        }

        Ok(reports)
    }

    fn get_cached_order_for_status_command(
        &self,
        cmd: &GenerateOrderStatusReport,
    ) -> Option<OrderAny> {
        let cache = self.core.cache();

        if let Some(client_order_id) = cmd.client_order_id {
            return cache.order(&client_order_id).map(|o| o.clone());
        }

        let venue_order_id = cmd.venue_order_id?;
        let client_order_id = *cache.client_order_id(&venue_order_id)?;
        cache.order(&client_order_id).map(|o| o.clone())
    }
}

fn synthesize_filled_order_status_report(
    cmd: &GenerateOrderStatusReport,
    order: &OrderAny,
    fills: &[FillReport],
) -> Option<OrderStatusReport> {
    let venue_order_id = cmd.venue_order_id.or(order.venue_order_id());
    let truncated_client_order_id = truncate_cl_ord_id(&order.client_order_id());

    let mut matched: Vec<&FillReport> = if let Some(venue_order_id) = venue_order_id {
        fills
            .iter()
            .filter(|fill| fill.venue_order_id == venue_order_id)
            .collect()
    } else {
        Vec::new()
    };

    if matched.is_empty() {
        matched = fills
            .iter()
            .filter(|fill| {
                fill.client_order_id == Some(order.client_order_id())
                    || fill
                        .client_order_id
                        .as_ref()
                        .is_some_and(|fill_client_order_id| {
                            fill_client_order_id.as_str() == truncated_client_order_id
                        })
            })
            .collect();
    }

    if matched.is_empty() {
        return None;
    }

    matched.sort_by_key(|fill| fill.ts_event);
    let first_fill = *matched.first()?;
    let last_fill = *matched.last()?;

    let total_filled = matched
        .iter()
        .fold(Decimal::ZERO, |acc, fill| acc + fill.last_qty.as_decimal());
    if total_filled < order.quantity().as_decimal() {
        return None;
    }

    let total_notional = matched.iter().fold(Decimal::ZERO, |acc, fill| {
        acc + fill.last_qty.as_decimal() * fill.last_px.as_decimal()
    });
    let avg_px = if total_filled.is_zero() {
        None
    } else {
        Some(total_notional / total_filled)
    };
    let venue_order_id = venue_order_id.unwrap_or(first_fill.venue_order_id);

    let mut report = OrderStatusReport::new(
        first_fill.account_id,
        order.instrument_id(),
        Some(order.client_order_id()),
        venue_order_id,
        order.order_side().into(),
        order.order_type(),
        order.time_in_force(),
        OrderStatus::Filled,
        order.quantity(),
        order.quantity(),
        first_fill.ts_event,
        last_fill.ts_event,
        last_fill.ts_init,
        None,
    );
    report.order_list_id = order.order_list_id();
    report.venue_position_id = matched.iter().rev().find_map(|fill| fill.venue_position_id);
    report.linked_order_ids = order
        .linked_order_ids()
        .map(|linked_order_ids| linked_order_ids.to_vec());
    report.parent_order_id = order.parent_order_id();
    report.expire_time = order.expire_time();
    report.price = order.price();
    report.trigger_price = order.trigger_price();
    report.trigger_type = order.trigger_type();
    report.avg_px = avg_px;
    report.display_qty = order.display_qty();
    report.post_only = order.is_post_only();
    report.reduce_only = order.is_reduce_only();
    Some(report)
}

#[cfg(test)]
mod tests {
    use nautilus_core::{UUID4, UnixNanos};
    use nautilus_model::{
        enums::{LiquiditySide, OrderSide, OrderType, TimeInForce},
        identifiers::{
            AccountId, ClientOrderId, InstrumentId, StrategyId, TradeId, TraderId, VenueOrderId,
        },
        orders::OrderTestBuilder,
        reports::FillReport,
        types::{Currency, Money, Price, Quantity},
    };
    use rstest::rstest;

    use super::*;

    const TEST_INSTRUMENT_ID: &str = "PF_XBTUSD.KRAKEN";

    #[tokio::test]
    async fn test_cancel_order_for_futures_missing_cached_instrument_returns_canonical_error() {
        let http = KrakenFuturesHttpClient::default();
        let instrument_id = InstrumentId::from(TEST_INSTRUMENT_ID);
        let client_order_id = ClientOrderId::from("C-001");
        let venue_order_id = VenueOrderId::from("V-001");

        let result = cancel_order_for_futures(
            &http,
            AccountId::from("KRAKEN-001"),
            instrument_id,
            Some(client_order_id),
            Some(venue_order_id),
        )
        .await;

        match result {
            Err(CommandFailure::NotSent(reason)) => {
                assert_eq!(
                    reason,
                    InstrumentLookupError::not_found(instrument_id).to_string()
                );
            }
            _ => panic!("Expected local validation failure"),
        }
    }

    #[rstest]
    fn test_batch_cancel_item_for_futures_missing_cached_instrument_returns_canonical_error() {
        let http = KrakenFuturesHttpClient::default();
        let instrument_id = InstrumentId::from(TEST_INSTRUMENT_ID);
        let cancel = CancelOrder::new(
            TraderId::from("TESTER-001"),
            None,
            StrategyId::from("S-001"),
            instrument_id,
            ClientOrderId::from("C-001"),
            Some(VenueOrderId::from("V-001")),
            UUID4::new(),
            UnixNanos::default(),
            None,
            None,
        );

        let result = batch_cancel_item_for_futures(&http, &cancel);

        match result {
            Err(CommandFailure::NotSent(reason)) => {
                assert_eq!(
                    reason,
                    InstrumentLookupError::not_found(instrument_id).to_string()
                );
            }
            _ => panic!("Expected local validation failure"),
        }
    }

    fn make_fill(
        venue_order_id: &str,
        client_order_id: Option<&str>,
        quantity: &str,
        price: &str,
        ts_event: u64,
    ) -> FillReport {
        FillReport::new(
            AccountId::from("KRAKEN-001"),
            InstrumentId::from(TEST_INSTRUMENT_ID),
            VenueOrderId::from(venue_order_id),
            TradeId::from(format!("T-{ts_event}").as_str()),
            OrderSide::Buy,
            Quantity::from(quantity),
            Price::from(price),
            Money::from_decimal(Decimal::ZERO, Currency::USD()).unwrap(),
            LiquiditySide::Taker,
            client_order_id.map(ClientOrderId::from),
            None,
            UnixNanos::from(ts_event),
            UnixNanos::from(ts_event),
            None,
        )
    }

    fn make_cmd(
        client_order_id: Option<&str>,
        venue_order_id: Option<&str>,
    ) -> GenerateOrderStatusReport {
        GenerateOrderStatusReport::new(
            UUID4::new(),
            UnixNanos::default(),
            Some(InstrumentId::from(TEST_INSTRUMENT_ID)),
            client_order_id.map(ClientOrderId::from),
            venue_order_id.map(VenueOrderId::from),
            None,
            None,
        )
    }

    fn make_order(client_order_id: &str) -> OrderAny {
        OrderTestBuilder::new(OrderType::Market)
            .instrument_id(InstrumentId::from(TEST_INSTRUMENT_ID))
            .client_order_id(ClientOrderId::from(client_order_id))
            .side(OrderSide::Buy)
            .quantity(Quantity::from("100"))
            .time_in_force(TimeInForce::Ioc)
            .build()
    }

    #[rstest]
    fn test_synthesize_filled_order_status_report_matches_full_fill_by_venue_order_id() {
        let order = make_order("O-123456");
        let cmd = make_cmd(Some("O-123456"), Some("KRAKEN-789"));
        let fills = vec![
            make_fill("KRAKEN-789", Some("O-123456"), "40", "50000.0", 1),
            make_fill("KRAKEN-789", Some("O-123456"), "60", "50010.0", 2),
            make_fill("KRAKEN-OTHER", Some("O-123456"), "999", "1.0", 3),
        ];

        let report = synthesize_filled_order_status_report(&cmd, &order, &fills)
            .expect("expected a filled report");

        assert_eq!(report.venue_order_id, VenueOrderId::from("KRAKEN-789"));
        assert_eq!(
            report.client_order_id,
            Some(ClientOrderId::from("O-123456"))
        );
        assert_eq!(report.order_status, OrderStatus::Filled);
        assert_eq!(report.order_type, OrderType::Market);
        assert_eq!(report.time_in_force, TimeInForce::Ioc);
        assert_eq!(report.quantity, Quantity::from("100"));
        assert_eq!(report.filled_qty, Quantity::from("100"));
        assert_eq!(
            report.avg_px,
            Some(Decimal::from_str_exact("50006.0").unwrap())
        );
    }

    #[rstest]
    fn test_synthesize_filled_order_status_report_requires_full_fill_size() {
        let order = make_order("O-123457");
        let cmd = make_cmd(Some("O-123457"), Some("KRAKEN-790"));
        let fills = vec![make_fill(
            "KRAKEN-790",
            Some("O-123457"),
            "40",
            "50000.0",
            1,
        )];

        assert!(synthesize_filled_order_status_report(&cmd, &order, &fills).is_none());
    }

    #[rstest]
    fn test_synthesize_filled_order_status_report_matches_truncated_client_order_id() {
        let long_client_order_id = "O202602270023210040011";
        let order = make_order(long_client_order_id);
        let cmd = make_cmd(Some(long_client_order_id), None);
        let fills = vec![make_fill(
            "KRAKEN-791",
            Some(truncate_cl_ord_id(&ClientOrderId::from(long_client_order_id)).as_str()),
            "100",
            "50000.0",
            1,
        )];

        let report = synthesize_filled_order_status_report(&cmd, &order, &fills)
            .expect("expected a filled report");

        assert_eq!(
            report.client_order_id,
            Some(ClientOrderId::from(long_client_order_id))
        );
        assert_eq!(report.venue_order_id, VenueOrderId::from("KRAKEN-791"));
        assert_eq!(report.order_status, OrderStatus::Filled);
    }
}