nautilus_kraken/data/

spot.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 Spot data client implementation.

use std::{
    future::Future,
    sync::{
        Arc,
        atomic::{AtomicBool, AtomicU64, Ordering},
    },
    time::Duration,
};

use ahash::AHashMap;
use anyhow::Context;
use async_trait::async_trait;
use futures_util::StreamExt;
use nautilus_common::{
    clients::DataClient,
    live::{get_data_event_sender, sender::EventSender},
    messages::{
        DataEvent,
        data::{
            BarsResponse, BookResponse, DataResponse, InstrumentResponse, InstrumentsResponse,
            RequestBars, RequestBookSnapshot, RequestInstrument, RequestInstruments, RequestTrades,
            SubscribeBars, SubscribeBookDeltas, SubscribeIndexPrices, SubscribeInstrument,
            SubscribeInstrumentStatus, SubscribeInstruments, SubscribeMarkPrices, SubscribeQuotes,
            SubscribeTrades, TradesResponse, UnsubscribeBars, UnsubscribeBookDeltas,
            UnsubscribeIndexPrices, UnsubscribeInstrumentStatus, UnsubscribeMarkPrices,
            UnsubscribeQuotes, UnsubscribeTrades,
        },
    },
};
use nautilus_core::{
    AtomicMap, UnixNanos,
    datetime::datetime_to_unix_nanos,
    time::{AtomicTime, get_atomic_clock_realtime},
};
use nautilus_live::{
    SocketControlFactory,
    task::{TaskGroup, TaskRef},
};
use nautilus_model::{
    data::{Bar, Data, OrderBookDeltas},
    enums::{AggregationSource, BookType},
    identifiers::{ClientId, InstrumentId, Venue},
    instruments::{Instrument, InstrumentAny},
};
use parking_lot::Mutex;
use tokio_util::sync::CancellationToken;
use ustr::Ustr;

use crate::{
    common::{consts::KRAKEN_VENUE, lookup_instrument_in_snapshot},
    config::KrakenDataClientConfig,
    http::{KrakenSpotHttpClient, spot::client::KRAKEN_SPOT_DEFAULT_RATE_LIMIT_PER_SECOND},
    websocket::spot_v2::{
        client::KrakenSpotWebSocketClient,
        level_2::{L2BookState, L2Depths},
        level_3::{
            BookOrderIdHasher, KrakenL3WsMessage,
            resync::retry_l3_resync,
            runtime::{L3Sink, L3State, process_l3_message},
        },
        messages::KrakenSpotWsMessage,
        parse::{parse_quote_tick, parse_trade_tick, parse_ws_bar},
    },
};

/// Kraken Spot data client.
///
/// Provides real-time market data from Kraken Spot markets through WebSocket v2.
#[allow(dead_code)]
#[derive(Debug)]
pub struct KrakenSpotDataClient {
    clock: &'static AtomicTime,
    client_id: ClientId,
    config: KrakenDataClientConfig,
    http: KrakenSpotHttpClient,
    ws: KrakenSpotWebSocketClient,
    ws_l3: Option<KrakenSpotWebSocketClient>,
    socket_factory: SocketControlFactory,
    l3_handler_task: Option<TaskRef>,
    is_connected: AtomicBool,
    cancellation_token: CancellationToken,
    session_tasks: TaskGroup,
    command_tasks: TaskGroup,
    instruments: Arc<AtomicMap<InstrumentId, InstrumentAny>>,
    data_sender: EventSender<DataEvent>,
}

impl KrakenSpotDataClient {
    /// Creates a new [`KrakenSpotDataClient`] instance.
    pub fn new(client_id: ClientId, config: KrakenDataClientConfig) -> anyhow::Result<Self> {
        let session_tasks = TaskGroup::new();
        let cancellation_token = session_tasks.cancellation_token();
        let command_tasks = TaskGroup::new();
        let socket_factory = SocketControlFactory::new(client_id, Some(*KRAKEN_VENUE));
        let proxy_url = config
            .proxy_url
            .as_ref()
            .map(|value| value.expose_secret().to_owned());

        let max_requests_per_second = config
            .max_requests_per_second
            .unwrap_or(KRAKEN_SPOT_DEFAULT_RATE_LIMIT_PER_SECOND);
        let http = match (&config.api_key, &config.api_secret) {
            (Some(api_key), Some(api_secret)) => KrakenSpotHttpClient::with_credentials(
                api_key.expose_secret().to_owned(),
                api_secret.expose_secret().to_owned(),
                config.environment,
                config.base_url.clone(),
                config.timeout_secs,
                None,
                None,
                None,
                proxy_url.clone(),
                max_requests_per_second,
            )?,
            _ => KrakenSpotHttpClient::new(
                config.environment,
                config.base_url.clone(),
                config.timeout_secs,
                None,
                None,
                None,
                proxy_url.clone(),
                max_requests_per_second,
            )?,
        };

        let ws =
            KrakenSpotWebSocketClient::new(config.clone(), cancellation_token.clone(), proxy_url)
                .with_socket_control(socket_factory.control("kraken-spot-data-streams"));

        Ok(Self {
            clock: get_atomic_clock_realtime(),
            client_id,
            config,
            http,
            ws,
            ws_l3: None,
            socket_factory,
            l3_handler_task: None,
            is_connected: AtomicBool::new(false),
            cancellation_token,
            session_tasks,
            command_tasks,
            instruments: Arc::new(AtomicMap::new()),
            data_sender: get_data_event_sender(),
        })
    }

    /// Returns the cached instruments.
    #[must_use]
    pub fn instruments(&self) -> Vec<InstrumentAny> {
        self.instruments.load().values().cloned().collect()
    }

    /// Returns a cached instrument by ID.
    #[must_use]
    pub fn get_instrument(&self, instrument_id: &InstrumentId) -> Option<InstrumentAny> {
        self.instruments.load().get(instrument_id).cloned()
    }

    async fn load_instruments(&self) -> anyhow::Result<Vec<InstrumentAny>> {
        let instruments = self
            .http
            .request_instruments(None)
            .await
            .context("Failed to load spot instruments")?;

        self.instruments.rcu(|m| {
            for instrument in &instruments {
                m.insert(instrument.id(), instrument.clone());
            }
        });

        self.http.cache_instruments(&instruments);

        log::debug!(
            "Loaded instruments: client_id={}, count={}",
            self.client_id,
            instruments.len()
        );

        Ok(instruments)
    }

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

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

    fn spawn_command<F>(&self, future: F)
    where
        F: Future<Output = ()> + Send + 'static,
    {
        if let Err(e) = self.command_tasks.spawn(future) {
            log::warn!("Skipping Kraken Spot data command after shutdown began: {e}");
        }
    }

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

    async fn prepare_task_groups(&mut self) -> anyhow::Result<()> {
        if !self.session_tasks.is_open() || !self.command_tasks.is_open() {
            self.session_tasks.begin_shutdown();
            self.command_tasks.begin_shutdown();
            self.ws
                .close()
                .await
                .context("failed to close prior Kraken Spot WebSocket")?;

            if let Some(ws_l3) = self.ws_l3.as_mut() {
                ws_l3
                    .close()
                    .await
                    .context("failed to close prior Kraken Spot L3 WebSocket")?;
                self.ws_l3 = None;
                self.l3_handler_task = None;
            }
            self.finish_tasks().await?;
            self.session_tasks
                .start_generation()
                .context("failed to start Kraken Spot data session task generation")?;
            self.command_tasks
                .start_generation()
                .context("failed to start Kraken Spot data command task generation")?;
            self.cancellation_token = self.session_tasks.cancellation_token();
            self.ws = KrakenSpotWebSocketClient::new(
                self.config.clone(),
                self.cancellation_token.clone(),
                self.config
                    .proxy_url
                    .as_ref()
                    .map(|value| value.expose_secret().to_owned()),
            )
            .with_socket_control(self.socket_factory.control("kraken-spot-data-streams"));
        }
        Ok(())
    }

    async fn teardown_partial_connect(&mut self) -> anyhow::Result<()> {
        self.session_tasks.begin_shutdown();
        self.command_tasks.begin_shutdown();
        let ws_result = self.ws.close().await;
        let ws_l3_result = if let Some(ws_l3) = self.ws_l3.as_mut() {
            ws_l3.close().await
        } else {
            Ok(())
        };

        if ws_l3_result.is_ok() {
            self.ws_l3 = None;
            self.l3_handler_task = None;
        }
        let tasks_result = self.finish_tasks().await;
        self.is_connected.store(false, Ordering::Release);
        tasks_result?;
        ws_result?;
        Ok(ws_l3_result?)
    }

    fn subscribe_l3_book(&mut self, cmd: &SubscribeBookDeltas) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let symbol_ustr = instrument_id.symbol.inner();
        let depth = cmd.depth.map_or(1000, |d| d.get() as u32);

        if !matches!(depth, 10 | 100 | 1000) {
            anyhow::bail!("Invalid L3 depth {depth} for Kraken Spot, valid values: 10, 100, 1000");
        }

        if !self.config.has_api_credentials() {
            anyhow::bail!(
                "L3 order book requires API credentials; configure api_key and api_secret"
            );
        }

        let handler_finished = self
            .l3_handler_task
            .as_ref()
            .is_none_or(TaskRef::is_finished);

        if self.ws_l3.is_none() {
            let ws_l3 = KrakenSpotWebSocketClient::l3(
                self.config.clone(),
                self.cancellation_token.clone(),
                self.config
                    .proxy_url
                    .as_ref()
                    .map(|value| value.expose_secret().to_owned()),
            )
            .with_socket_control(self.socket_factory.control("kraken-spot-l3-data-streams"));

            self.l3_handler_task = self.spawn_l3_handler_task(ws_l3.clone(), false);
            self.ws_l3 = Some(ws_l3);
        } else if handler_finished && let Some(ws_l3) = self.ws_l3.as_ref() {
            let ws_l3 = ws_l3.clone();
            self.l3_handler_task = self.spawn_l3_handler_task(ws_l3, true);
        }

        let ws_l3 = self
            .ws_l3
            .as_ref()
            .expect("ws_l3 initialized above")
            .clone();

        self.spawn_ws(
            async move {
                ws_l3
                    .wait_until_active(10.0)
                    .await
                    .map_err(|e| anyhow::anyhow!("L3 WebSocket failed to become active: {e}"))?;
                ws_l3
                    .wait_until_authenticated(10.0)
                    .await
                    .map_err(|e| anyhow::anyhow!("L3 WebSocket failed to authenticate: {e}"))?;
                ws_l3
                    .subscribe_book_l3(symbol_ustr, depth)
                    .await
                    .map_err(|e| anyhow::anyhow!("{e}"))
            },
            "subscribe l3 book",
        );

        Ok(())
    }

    fn spawn_l3_handler_task(
        &self,
        handler_client: KrakenSpotWebSocketClient,
        restart: bool,
    ) -> Option<TaskRef> {
        let data_sender = self.data_sender.clone();
        let instruments = self.instruments.clone();
        let cancellation_token = self.cancellation_token.clone();
        let clock = self.clock;
        let session_spawner = match self.session_tasks.spawner() {
            Ok(spawner) => spawner,
            Err(e) => {
                log::warn!("Skipping Kraken L3 handler after shutdown began: {e}");
                return None;
            }
        };

        let future = async move {
            let mut handler_client = handler_client;

            if restart && let Err(e) = handler_client.close().await {
                log::error!("Failed to close prior L3 WebSocket generation: {e}");
                return;
            }

            if let Err(e) = handler_client.connect().await {
                log::error!("L3 WebSocket connect failed: {e}");
                return;
            }

            if let Err(e) = handler_client.wait_until_active(10.0).await {
                log::error!("L3 WebSocket failed to become active: {e}");
                return;
            }

            if let Err(e) = handler_client.authenticate().await {
                log::error!("L3 WebSocket authentication failed: {e}");
                return;
            }

            let stream = match handler_client.stream() {
                Ok(s) => s,
                Err(e) => {
                    log::error!("L3 stream() failed: {e}");
                    return;
                }
            };
            tokio::pin!(stream);

            let mut states: AHashMap<String, L3State> = AHashMap::new();
            let hasher = BookOrderIdHasher::new();
            let l3_depths = handler_client.l3_depths_handle();
            let validate_checksum = handler_client.validate_l3_checksum();
            let resync_client = handler_client.clone();

            loop {
                tokio::select! {
                    () = cancellation_token.cancelled() => break,
                    msg = stream.next() => {
                        let Some(msg) = msg else { break };
                        let ts_init = clock.get_time_ns();

                        let runtime_msg = match msg {
                            KrakenSpotWsMessage::L3Snapshot(snap) => {
                                KrakenL3WsMessage::Snapshot(snap)
                            }
                            KrakenSpotWsMessage::L3Update(update) => {
                                KrakenL3WsMessage::Update(update)
                            }
                            KrakenSpotWsMessage::Reconnected => {
                                log::info!("L3 WebSocket reconnected");

                                for state in states.values_mut() {
                                    state.open_orders.clear();
                                    state.awaiting_snapshot = true;
                                }
                                continue;
                            }
                            _ => continue,
                        };

                        let mut sink = DataEventSink { sender: &data_sender };
                        let resync = process_l3_message(
                            runtime_msg,
                            &mut sink,
                            &instruments,
                            &l3_depths,
                            &mut states,
                            &hasher,
                            validate_checksum,
                            ts_init,
                        );

                        if let Some(request) = resync {
                            log::warn!(
                                "Resyncing Kraken L3 book: symbol={}, depth={}, reason={}",
                                request.symbol,
                                request.depth,
                                request.reason,
                            );
                            let symbol_ustr = Ustr::from(&request.symbol);
                            let client_for_resync = resync_client.clone();

                            if let Err(e) = session_spawner.spawn_named(
                                "kraken-spot-l3-resync",
                                async move {
                                    retry_l3_resync(
                                        &client_for_resync,
                                        symbol_ustr,
                                        request.depth,
                                    )
                                    .await;
                                },
                            ) {
                                log::warn!("Skipping Kraken L3 resync after shutdown began: {e}");
                            }
                        }
                    }
                }
            }
        };

        match self
            .session_tasks
            .spawn_named("kraken-spot-l3-handler", future)
        {
            Ok(task) => Some(task),
            Err(e) => {
                log::warn!("Skipping Kraken L3 handler after shutdown began: {e}");
                None
            }
        }
    }

    fn spawn_message_handler(&mut self) -> anyhow::Result<()> {
        let stream = self.ws.stream().map_err(|e| anyhow::anyhow!("{e}"))?;
        let data_sender = self.data_sender.clone();
        let instruments = self.instruments.clone();
        let book_sequence = Arc::new(AtomicU64::new(0));
        let ohlc_buffer: OhlcBuffer = Arc::new(Mutex::new(AHashMap::new()));
        let l2_depths = self.ws.l2_depths_handle();
        let cancellation_token = self.cancellation_token.clone();
        let clock = self.clock;

        let future = async move {
            tokio::pin!(stream);
            let mut l2_books = L2BookState::default();

            loop {
                tokio::select! {
                    () = cancellation_token.cancelled() => {
                        log::debug!("Spot message handler cancelled");
                        Self::flush_ohlc_buffer(&ohlc_buffer, &data_sender);
                        break;
                    }
                    msg = stream.next() => {
                        match msg {
                            Some(ws_msg) => {
                                let context = SpotMessageContext {
                                    sender: &data_sender,
                                    instruments: &instruments,
                                    book_sequence: &book_sequence,
                                    l2_depths: &l2_depths,
                                    ohlc_buffer: &ohlc_buffer,
                                    clock,
                                };
                                Self::handle_ws_message(ws_msg, &context, &mut l2_books);
                            }
                            None => {
                                log::debug!("Spot WebSocket stream ended");
                                Self::flush_ohlc_buffer(&ohlc_buffer, &data_sender);
                                break;
                            }
                        }
                    }
                }
            }
        };

        self.session_tasks
            .spawn(future)
            .context("failed to register Kraken Spot message handler")
    }

    fn flush_ohlc_buffer(ohlc_buffer: &OhlcBuffer, sender: &EventSender<DataEvent>) {
        let mut buffer = ohlc_buffer.lock();
        let bars: Vec<Bar> = buffer.drain().map(|(_, (bar, _))| bar).collect();
        for bar in bars {
            if let Err(e) = sender.send(DataEvent::Data(Data::Bar(bar))) {
                log::error!("Failed to send buffered bar: {e}");
            }
        }
    }

    fn handle_ws_message(
        msg: KrakenSpotWsMessage,
        context: &SpotMessageContext,
        l2_books: &mut L2BookState,
    ) {
        let ts_init = context.clock.get_time_ns();

        match msg {
            KrakenSpotWsMessage::Ticker(tickers) => {
                let instruments = context.instruments.load();

                for ticker in &tickers {
                    let Some(instrument) =
                        lookup_instrument_in_snapshot(&instruments, ticker.symbol.as_str())
                    else {
                        log::warn!("No instrument for symbol: {}", ticker.symbol);
                        continue;
                    };

                    match parse_quote_tick(ticker, instrument, ts_init) {
                        Ok(quote) => {
                            if let Err(e) = context.sender.send(DataEvent::Data(Data::Quote(quote)))
                            {
                                log::error!("Failed to send quote: {e}");
                            }
                        }
                        Err(e) => log::error!("Failed to parse quote tick: {e}"),
                    }
                }
            }
            KrakenSpotWsMessage::Trade(trades) => {
                let instruments = context.instruments.load();

                for trade in &trades {
                    let Some(instrument) =
                        lookup_instrument_in_snapshot(&instruments, trade.symbol.as_str())
                    else {
                        log::warn!("No instrument for symbol: {}", trade.symbol);
                        continue;
                    };

                    match parse_trade_tick(trade, instrument, ts_init) {
                        Ok(tick) => {
                            if let Err(e) = context.sender.send(DataEvent::Data(Data::Trade(tick)))
                            {
                                log::error!("Failed to send trade: {e}");
                            }
                        }
                        Err(e) => log::error!("Failed to parse trade tick: {e}"),
                    }
                }
            }
            KrakenSpotWsMessage::Book { data, is_snapshot } => {
                let instruments = context.instruments.load();

                for book in &data {
                    let Some(instrument) =
                        lookup_instrument_in_snapshot(&instruments, book.symbol.as_str())
                    else {
                        log::warn!("No instrument for symbol: {}", book.symbol);
                        continue;
                    };
                    let sequence = context.book_sequence.load(Ordering::Relaxed);
                    let depth = context.l2_depths.get(book.symbol.as_str());
                    match l2_books.process_book(
                        book,
                        instrument,
                        sequence,
                        is_snapshot,
                        depth,
                        ts_init,
                    ) {
                        Ok(Some((deltas, next_sequence))) => {
                            context
                                .book_sequence
                                .store(next_sequence, Ordering::Relaxed);

                            if let Err(e) = context
                                .sender
                                .send(DataEvent::Data(Data::BookDeltas(Box::new(deltas))))
                            {
                                log::error!("Failed to send deltas: {e}");
                            }
                        }
                        Ok(None) => {}
                        Err(e) => log::error!("Failed to parse book deltas: {e}"),
                    }
                }
            }
            KrakenSpotWsMessage::Ohlc(ohlc_data) => {
                let mut buffer = context.ohlc_buffer.lock();

                let instruments = context.instruments.load();

                for ohlc in &ohlc_data {
                    let Some(instrument) =
                        lookup_instrument_in_snapshot(&instruments, ohlc.symbol.as_str())
                    else {
                        log::warn!("No instrument for symbol: {}", ohlc.symbol);
                        continue;
                    };

                    match parse_ws_bar(ohlc, instrument, ts_init) {
                        Ok(new_bar) => {
                            let key: (Ustr, u32) = (ohlc.symbol, ohlc.interval);
                            let new_interval_begin = UnixNanos::from(
                                u64::try_from(ohlc.interval_begin.as_nanosecond()).unwrap_or(0),
                            );

                            if let Some((buffered_bar, buffered_begin)) = buffer.get(&key)
                                && new_interval_begin != *buffered_begin
                                && let Err(e) = context
                                    .sender
                                    .send(DataEvent::Data(Data::Bar(*buffered_bar)))
                            {
                                log::error!("Failed to send bar: {e}");
                            }

                            buffer.insert(key, (new_bar, new_interval_begin));
                        }
                        Err(e) => log::error!("Failed to parse bar: {e}"),
                    }
                }
            }
            KrakenSpotWsMessage::Execution(_) => {}
            KrakenSpotWsMessage::OrderResponse(_) => {}
            KrakenSpotWsMessage::L3Snapshot(_) => {}
            KrakenSpotWsMessage::L3Update(_) => {}
            KrakenSpotWsMessage::Reconnected => {
                log::info!("Spot WebSocket reconnected");
            }
        }
    }
}

#[async_trait(?Send)]
impl DataClient for KrakenSpotDataClient {
    fn client_id(&self) -> ClientId {
        self.client_id
    }

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

    fn start(&mut self) -> anyhow::Result<()> {
        log::info!(
            "Starting Spot data client: client_id={}, environment={:?}",
            self.client_id,
            self.config.environment
        );
        Ok(())
    }

    fn stop(&mut self) -> anyhow::Result<()> {
        log::info!("Stopping Spot data client: {}", self.client_id);
        self.session_tasks.begin_shutdown();
        self.command_tasks.begin_shutdown();
        self.ws.begin_shutdown();
        self.is_connected.store(false, Ordering::Relaxed);
        Ok(())
    }

    fn reset(&mut self) -> anyhow::Result<()> {
        log::info!("Resetting Spot data client: {}", self.client_id);
        self.session_tasks.begin_shutdown();
        self.command_tasks.begin_shutdown();
        self.ws.begin_shutdown();
        self.is_connected.store(false, Ordering::Relaxed);

        self.instruments.store(ahash::AHashMap::new());
        Ok(())
    }

    fn dispose(&mut self) -> anyhow::Result<()> {
        log::debug!("Disposing Spot data client: {}", self.client_id);
        self.stop()
    }

    fn is_connected(&self) -> bool {
        self.is_connected.load(Ordering::SeqCst)
    }

    fn is_disconnected(&self) -> bool {
        !self.is_connected()
    }

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

        self.prepare_task_groups().await?;

        let instruments = self.load_instruments().await?;

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

            self.spawn_message_handler()?;

            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 Spot data startup teardown failed: {teardown_error}"
                )));
            }
            return Err(e);
        }

        for instrument in instruments {
            if let Err(e) = self.data_sender.send(DataEvent::Instrument(instrument)) {
                log::error!("Failed to send instrument: {e}");
            }
        }

        self.is_connected.store(true, Ordering::Release);
        log::info!("Connected: client_id={}, product_type=Spot", self.client_id);
        Ok(())
    }

    async fn disconnect(&mut self) -> anyhow::Result<()> {
        self.teardown_partial_connect().await?;
        self.is_connected.store(false, Ordering::Relaxed);

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

    fn subscribe_instruments(&mut self, _cmd: SubscribeInstruments) -> anyhow::Result<()> {
        log::debug!("subscribe_instruments: Kraken instruments are fetched via HTTP on connect");
        Ok(())
    }

    fn subscribe_instrument(&mut self, _cmd: SubscribeInstrument) -> anyhow::Result<()> {
        log::debug!("subscribe_instrument: Kraken instruments are fetched via HTTP on connect");
        Ok(())
    }

    fn subscribe_book_deltas(&mut self, cmd: SubscribeBookDeltas) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let depth = cmd.depth;

        match cmd.book_type {
            BookType::L2_MBP => {}
            BookType::L3_MBO => return self.subscribe_l3_book(&cmd),
            other => {
                log::warn!("Unsupported BookType {other:?} for Kraken Spot, skipping");
                return Ok(());
            }
        }

        if let Some(d) = depth {
            let d_val = d.get();
            if !matches!(d_val, 10 | 25 | 100 | 500 | 1000) {
                log::warn!("Invalid depth {d_val} for Kraken Spot, valid: 10, 25, 100, 500, 1000");
                return Ok(());
            }
        }

        let ws = self.ws.clone();
        self.spawn_ws(
            async move {
                ws.subscribe_book(instrument_id, depth.map(|d| d.get() as u32))
                    .await
                    .map_err(|e| anyhow::anyhow!("{e}"))
            },
            "subscribe book",
        );

        Ok(())
    }

    fn subscribe_quotes(&mut self, cmd: SubscribeQuotes) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let ws = self.ws.clone();

        self.spawn_ws(
            async move {
                ws.subscribe_quotes(instrument_id)
                    .await
                    .map_err(|e| anyhow::anyhow!("{e}"))
            },
            "subscribe quotes",
        );

        Ok(())
    }

    fn subscribe_trades(&mut self, cmd: SubscribeTrades) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let ws = self.ws.clone();

        self.spawn_ws(
            async move {
                ws.subscribe_trades(instrument_id)
                    .await
                    .map_err(|e| anyhow::anyhow!("{e}"))
            },
            "subscribe trades",
        );

        Ok(())
    }

    fn subscribe_mark_prices(&mut self, cmd: SubscribeMarkPrices) -> anyhow::Result<()> {
        log::warn!(
            "Mark price subscription not supported for Spot instrument {}",
            cmd.instrument_id
        );
        Ok(())
    }

    fn subscribe_index_prices(&mut self, cmd: SubscribeIndexPrices) -> anyhow::Result<()> {
        log::warn!(
            "Index price subscription not supported for Spot instrument {}",
            cmd.instrument_id
        );
        Ok(())
    }

    fn subscribe_bars(&mut self, cmd: SubscribeBars) -> anyhow::Result<()> {
        let bar_type = cmd.bar_type;

        if bar_type.aggregation_source() != AggregationSource::External {
            log::warn!("Cannot subscribe to {bar_type} bars: only EXTERNAL bars supported");
            return Ok(());
        }

        if !bar_type.spec().is_time_aggregated() {
            log::warn!("Cannot subscribe to {bar_type} bars: only time-based bars supported");
            return Ok(());
        }

        let ws = self.ws.clone();
        self.spawn_ws(
            async move {
                ws.subscribe_bars(bar_type)
                    .await
                    .map_err(|e| anyhow::anyhow!("{e}"))
            },
            "subscribe bars",
        );

        Ok(())
    }

    fn subscribe_instrument_status(
        &mut self,
        cmd: SubscribeInstrumentStatus,
    ) -> anyhow::Result<()> {
        log::debug!(
            "subscribe_instrument_status: {} (status changes detected via periodic instrument polling)",
            cmd.instrument_id,
        );
        Ok(())
    }

    fn unsubscribe_book_deltas(&mut self, cmd: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;

        if self.ws_l3.as_ref().is_some_and(|ws| {
            ws.subscriptions_contains(&format!("level3:{}", instrument_id.symbol))
        }) {
            let symbol_ustr = instrument_id.symbol.inner();

            if let Some(ws_l3) = self.ws_l3.clone() {
                self.spawn_ws(
                    async move {
                        ws_l3
                            .unsubscribe_book_l3(symbol_ustr)
                            .await
                            .map_err(|e| anyhow::anyhow!("{e}"))?;
                        Ok(())
                    },
                    "unsubscribe l3 book",
                );
            }
            return Ok(());
        }

        let ws = self.ws.clone();
        self.spawn_ws(
            async move {
                ws.unsubscribe_book(instrument_id)
                    .await
                    .map_err(|e| anyhow::anyhow!("{e}"))
            },
            "unsubscribe book",
        );

        Ok(())
    }

    fn unsubscribe_quotes(&mut self, cmd: &UnsubscribeQuotes) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let ws = self.ws.clone();

        self.spawn_ws(
            async move {
                ws.unsubscribe_quotes(instrument_id)
                    .await
                    .map_err(|e| anyhow::anyhow!("{e}"))
            },
            "unsubscribe quotes",
        );

        Ok(())
    }

    fn unsubscribe_trades(&mut self, cmd: &UnsubscribeTrades) -> anyhow::Result<()> {
        let instrument_id = cmd.instrument_id;
        let ws = self.ws.clone();

        self.spawn_ws(
            async move {
                ws.unsubscribe_trades(instrument_id)
                    .await
                    .map_err(|e| anyhow::anyhow!("{e}"))
            },
            "unsubscribe trades",
        );

        Ok(())
    }

    fn unsubscribe_mark_prices(&mut self, _cmd: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
        Ok(())
    }

    fn unsubscribe_index_prices(&mut self, _cmd: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
        Ok(())
    }

    fn unsubscribe_bars(&mut self, cmd: &UnsubscribeBars) -> anyhow::Result<()> {
        let bar_type = cmd.bar_type;
        let ws = self.ws.clone();

        self.spawn_ws(
            async move {
                ws.unsubscribe_bars(bar_type)
                    .await
                    .map_err(|e| anyhow::anyhow!("{e}"))
            },
            "unsubscribe bars",
        );

        Ok(())
    }

    fn unsubscribe_instrument_status(
        &mut self,
        _cmd: &UnsubscribeInstrumentStatus,
    ) -> anyhow::Result<()> {
        Ok(())
    }

    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
        let http = self.http.clone();
        let sender = self.data_sender.clone();
        let instruments_cache = self.instruments.clone();
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let venue = *KRAKEN_VENUE;
        let start_nanos = datetime_to_unix_nanos(request.start);
        let end_nanos = datetime_to_unix_nanos(request.end);
        let params = request.params;
        let clock = self.clock;

        self.spawn_command(async move {
            match http.request_instruments(None).await {
                Ok(instruments) => {
                    instruments_cache.rcu(|m| {
                        for instrument in &instruments {
                            m.insert(instrument.id(), instrument.clone());
                        }
                    });
                    http.cache_instruments(&instruments);

                    let response = DataResponse::Instruments(InstrumentsResponse::new(
                        request_id,
                        client_id,
                        venue,
                        instruments,
                        start_nanos,
                        end_nanos,
                        clock.get_time_ns(),
                        params,
                    ));

                    if let Err(e) = sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send instruments response: {e}");
                    }
                }
                Err(e) => log::error!("Instruments request failed: {e:?}"),
            }
        });

        Ok(())
    }

    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
        let http = self.http.clone();
        let sender = self.data_sender.clone();
        let instruments = self.instruments.clone();
        let instrument_id = request.instrument_id;
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let start_nanos = datetime_to_unix_nanos(request.start);
        let end_nanos = datetime_to_unix_nanos(request.end);
        let params = request.params;
        let clock = self.clock;

        self.spawn_command(async move {
            match http.request_instruments(None).await {
                Ok(all_instruments) => {
                    instruments.rcu(|m| {
                        for instrument in &all_instruments {
                            m.insert(instrument.id(), instrument.clone());
                        }
                    });
                    http.cache_instruments(&all_instruments);

                    let instrument = all_instruments
                        .into_iter()
                        .find(|i| i.id() == instrument_id);

                    if let Some(instrument) = instrument {
                        let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
                            request_id,
                            client_id,
                            instrument.id(),
                            instrument,
                            start_nanos,
                            end_nanos,
                            clock.get_time_ns(),
                            params,
                        )));

                        if let Err(e) = sender.send(DataEvent::Response(response)) {
                            log::error!("Failed to send instrument response: {e}");
                        }
                    } else {
                        log::error!("Instrument not found: {instrument_id}");
                    }
                }
                Err(e) => log::error!("Instrument request failed: {e:?}"),
            }
        });

        Ok(())
    }
    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
        let http = self.http.clone();
        let sender = self.data_sender.clone();
        let instrument_id = request.instrument_id;
        let start = request.start;
        let end = request.end;
        let limit = request.limit.map(|n| n.get() as u64);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let params = request.params;
        let clock = self.clock;
        let start_nanos = datetime_to_unix_nanos(start);
        let end_nanos = datetime_to_unix_nanos(end);

        self.spawn_command(async move {
            match http.request_trades(instrument_id, start, end, limit).await {
                Ok(trades) => {
                    let response = DataResponse::Trades(TradesResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        trades,
                        start_nanos,
                        end_nanos,
                        clock.get_time_ns(),
                        params,
                    ));

                    if let Err(e) = sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send trades response: {e}");
                    }
                }
                Err(e) => log::error!("Trades request failed: {e:?}"),
            }
        });

        Ok(())
    }

    fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
        let http = self.http.clone();
        let sender = self.data_sender.clone();
        let bar_type = request.bar_type;
        let start = request.start;
        let end = request.end;
        let limit = request.limit.map(|n| n.get() as u64);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let params = request.params;
        let clock = self.clock;
        let start_nanos = datetime_to_unix_nanos(start);
        let end_nanos = datetime_to_unix_nanos(end);

        self.spawn_command(async move {
            match http.request_bars(bar_type, start, end, limit).await {
                Ok(bars) => {
                    let response = DataResponse::Bars(BarsResponse::new(
                        request_id,
                        client_id,
                        bar_type,
                        bars,
                        start_nanos,
                        end_nanos,
                        clock.get_time_ns(),
                        params,
                    ));

                    if let Err(e) = sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send bars response: {e}");
                    }
                }
                Err(e) => log::error!("Bars request failed: {e:?}"),
            }
        });

        Ok(())
    }

    fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
        let http = self.http.clone();
        let sender = self.data_sender.clone();
        let instrument_id = request.instrument_id;
        let depth = request.depth.map(|n| n.get() as u32);
        let request_id = request.request_id;
        let client_id = request.client_id.unwrap_or(self.client_id);
        let params = request.params;
        let clock = self.clock;

        self.spawn_command(async move {
            match http.request_book_snapshot(instrument_id, depth).await {
                Ok(book) => {
                    let response = DataResponse::Book(BookResponse::new(
                        request_id,
                        client_id,
                        instrument_id,
                        book,
                        None,
                        None,
                        clock.get_time_ns(),
                        params,
                    ));

                    if let Err(e) = sender.send(DataEvent::Response(response)) {
                        log::error!("Failed to send book snapshot response: {e}");
                    }
                }
                Err(e) => log::error!("Book snapshot request failed: {e:?}"),
            }
        });

        Ok(())
    }
}

type OhlcBufferKey = (Ustr, u32);
type OhlcBuffer = Arc<Mutex<AHashMap<OhlcBufferKey, (Bar, UnixNanos)>>>;

struct DataEventSink<'a> {
    sender: &'a EventSender<DataEvent>,
}

impl L3Sink for DataEventSink<'_> {
    fn emit_deltas(&mut self, deltas: OrderBookDeltas) {
        if let Err(e) = self
            .sender
            .send(DataEvent::Data(Data::BookDeltas(Box::new(deltas))))
        {
            log::error!("Failed to send L3 deltas: {e}");
        }
    }
}

struct SpotMessageContext<'a> {
    sender: &'a EventSender<DataEvent>,
    instruments: &'a Arc<AtomicMap<InstrumentId, InstrumentAny>>,
    book_sequence: &'a Arc<AtomicU64>,
    l2_depths: &'a L2Depths,
    ohlc_buffer: &'a OhlcBuffer,
    clock: &'static AtomicTime,
}

#[cfg(test)]
mod tests {
    use nautilus_common::{live::runner::set_data_event_sender, messages::DataEvent};
    use nautilus_model::{
        enums::{BookAction, RecordFlag},
        identifiers::Symbol,
        instruments::{InstrumentAny, currency_pair::CurrencyPair},
        types::{Currency, Price, Quantity},
    };
    use rstest::rstest;
    use rust_decimal::Decimal;
    use rust_decimal_macros::dec;

    use super::*;
    use crate::{
        common::consts::KRAKEN_CLIENT_ID,
        config::KrakenDataClientConfig,
        websocket::spot_v2::{
            level_3::messages::KrakenL3Snapshot,
            messages::{KrakenWsBookData, KrakenWsBookLevel},
        },
    };

    fn setup_test_env() {
        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
        set_data_event_sender(sender);
    }

    fn make_instrument() -> InstrumentAny {
        InstrumentAny::CurrencyPair(
            CurrencyPair::builder()
                .instrument_id(InstrumentId::from("BTC/USD.KRAKEN"))
                .raw_symbol(Symbol::from("BTC/USD"))
                .base_currency(Currency::BTC())
                .quote_currency(Currency::USD())
                .price_precision(1)
                .size_precision(8)
                .price_increment(Price::from("0.1"))
                .size_increment(Quantity::from("0.00000001"))
                .ts_event(UnixNanos::default())
                .ts_init(UnixNanos::default())
                .build()
                .unwrap(),
        )
    }

    #[rstest]
    fn test_spot_data_client_new() {
        setup_test_env();
        let config = KrakenDataClientConfig::default();
        let client = KrakenSpotDataClient::new(*KRAKEN_CLIENT_ID, config);
        assert!(client.is_ok());

        let client = client.unwrap();
        assert_eq!(client.client_id(), *KRAKEN_CLIENT_ID);
        assert_eq!(client.venue(), Some(*KRAKEN_VENUE));
        assert!(!client.is_connected());
        assert!(client.is_disconnected());
        assert!(client.instruments().is_empty());
    }

    #[rstest]
    #[tokio::test]
    async fn test_teardown_clears_l3_client_and_handler_task() {
        setup_test_env();
        let config = KrakenDataClientConfig::default();
        let mut client = KrakenSpotDataClient::new(*KRAKEN_CLIENT_ID, config.clone()).unwrap();
        let cancellation = client.session_tasks.cancellation_token();
        let task = client
            .session_tasks
            .spawn_named("kraken-spot-l3-handler", async move {
                cancellation.cancelled().await;
            })
            .unwrap();
        client.l3_handler_task = Some(task.clone());
        client.ws_l3 = Some(KrakenSpotWebSocketClient::l3(
            config,
            client.cancellation_token.clone(),
            None,
        ));

        client.teardown_partial_connect().await.unwrap();

        assert!(client.ws_l3.is_none());
        assert!(client.l3_handler_task.is_none());
        assert!(task.is_finished());
    }

    #[rstest]
    fn test_l3_snapshot_checksum_mismatch_emits_clear_and_requests_resync() {
        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
        let instruments = Arc::new(AtomicMap::new());
        let instrument = make_instrument();
        instruments.insert(instrument.id(), instrument);

        let depths = Arc::new(Mutex::new(AHashMap::new()));
        depths.lock().insert("BTC/USD".to_string(), 1000);

        let snapshot: KrakenL3Snapshot = serde_json::from_str(
            r#"{
                "symbol": "BTC/USD",
                "bids": [{
                    "order_id": "order-bid-1",
                    "limit_price": 4199.0,
                    "order_qty": 3.00000000,
                    "timestamp": "2024-01-01T00:00:00Z"
                }],
                "asks": [{
                    "order_id": "order-ask-1",
                    "limit_price": 4200.0,
                    "order_qty": 0.01000000,
                    "timestamp": "2024-01-01T00:00:00Z"
                }],
                "checksum": 1,
                "timestamp": "2024-01-01T00:00:00Z"
            }"#,
        )
        .unwrap();

        let mut states = AHashMap::new();
        let hasher = BookOrderIdHasher::new();

        let mut sink = DataEventSink {
            sender: &sender.into(),
        };

        let request = process_l3_message(
            KrakenL3WsMessage::Snapshot(snapshot),
            &mut sink,
            &instruments,
            &depths,
            &mut states,
            &hasher,
            true,
            get_atomic_clock_realtime().get_time_ns(),
        )
        .expect("expected resync request");

        assert_eq!(request.symbol, "BTC/USD");
        assert_eq!(request.depth, 1000);
        assert_eq!(request.reason, "snapshot checksum mismatch");

        let event = receiver.try_recv().expect("expected clear event");
        let DataEvent::Data(Data::BookDeltas(deltas)) = event else {
            panic!("expected deltas event");
        };

        assert_eq!(deltas.deltas.len(), 1);
        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
        assert!(states["BTC/USD"].awaiting_snapshot);
        assert!(states["BTC/USD"].open_orders.is_empty());
        assert!(receiver.try_recv().is_err());
    }

    #[rstest]
    fn test_l2_update_prunes_levels_beyond_subscribed_depth() {
        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel::<DataEvent>();
        let instruments = Arc::new(AtomicMap::new());
        let instrument = make_instrument();
        let instrument_id = instrument.id();
        instruments.insert(instrument_id, instrument);

        let book_sequence = Arc::new(AtomicU64::new(0));
        let l2_depths = L2Depths::default();
        l2_depths.insert("BTC/USD", 10);
        let mut l2_books = L2BookState::default();
        let ohlc_buffer = Arc::new(Mutex::new(AHashMap::new()));
        let context = SpotMessageContext {
            sender: &sender.into(),
            instruments: &instruments,
            book_sequence: &book_sequence,
            l2_depths: &l2_depths,
            ohlc_buffer: &ohlc_buffer,
            clock: get_atomic_clock_realtime(),
        };

        let snapshot = KrakenWsBookData {
            symbol: Ustr::from("BTC/USD"),
            bids: Some(
                (0..10)
                    .map(|i| book_level(Decimal::from(100 - i), Decimal::ONE))
                    .collect(),
            ),
            asks: Some(
                (0..10)
                    .map(|i| book_level(Decimal::from(101 + i), Decimal::ONE))
                    .collect(),
            ),
            checksum: Some(0),
            timestamp: "2024-01-01T00:00:00Z".parse().unwrap(),
        };
        KrakenSpotDataClient::handle_ws_message(
            KrakenSpotWsMessage::Book {
                data: vec![snapshot],
                is_snapshot: true,
            },
            &context,
            &mut l2_books,
        );

        let DataEvent::Data(Data::BookDeltas(snapshot_deltas)) =
            receiver.try_recv().expect("expected snapshot deltas")
        else {
            panic!("expected snapshot deltas");
        };
        assert_eq!(snapshot_deltas.deltas.len(), 21);
        assert_eq!(snapshot_deltas.deltas[0].action, BookAction::Clear);
        assert!(RecordFlag::F_LAST.matches(snapshot_deltas.deltas.last().unwrap().flags));

        let bid_update = KrakenWsBookData {
            symbol: Ustr::from("BTC/USD"),
            bids: Some(vec![book_level(dec!(100.5), Decimal::ONE)]),
            asks: Some(vec![]),
            checksum: Some(0),
            timestamp: "2024-01-01T00:00:01Z".parse().unwrap(),
        };
        KrakenSpotDataClient::handle_ws_message(
            KrakenSpotWsMessage::Book {
                data: vec![bid_update],
                is_snapshot: false,
            },
            &context,
            &mut l2_books,
        );

        let DataEvent::Data(Data::BookDeltas(bid_update_deltas)) =
            receiver.try_recv().expect("expected bid update deltas")
        else {
            panic!("expected bid update deltas");
        };
        assert_eq!(bid_update_deltas.deltas.len(), 2);
        assert_eq!(bid_update_deltas.deltas[0].action, BookAction::Update);
        assert_eq!(bid_update_deltas.deltas[1].action, BookAction::Delete);
        assert_eq!(bid_update_deltas.deltas[1].order.price, Price::from("91.0"));
        assert!(RecordFlag::F_LAST.matches(bid_update_deltas.deltas[1].flags));

        let ask_update = KrakenWsBookData {
            symbol: Ustr::from("BTC/USD"),
            bids: Some(vec![]),
            asks: Some(vec![book_level(dec!(100.6), Decimal::ONE)]),
            checksum: Some(0),
            timestamp: "2024-01-01T00:00:02Z".parse().unwrap(),
        };
        KrakenSpotDataClient::handle_ws_message(
            KrakenSpotWsMessage::Book {
                data: vec![ask_update],
                is_snapshot: false,
            },
            &context,
            &mut l2_books,
        );

        let DataEvent::Data(Data::BookDeltas(ask_update_deltas)) =
            receiver.try_recv().expect("expected ask update deltas")
        else {
            panic!("expected ask update deltas");
        };
        assert_eq!(ask_update_deltas.deltas.len(), 2);
        assert_eq!(ask_update_deltas.deltas[0].action, BookAction::Update);
        assert_eq!(ask_update_deltas.deltas[1].action, BookAction::Delete);
        assert_eq!(
            ask_update_deltas.deltas[1].order.price,
            Price::from("110.0")
        );
        assert!(RecordFlag::F_LAST.matches(ask_update_deltas.deltas[1].flags));

        let book = l2_books
            .books
            .get(&instrument_id)
            .expect("expected shadow book");
        assert_eq!(book.bids(None).count(), 10);
        assert_eq!(book.asks(None).count(), 10);
        assert_eq!(book.best_bid_price(), Some(Price::from("100.5")));
        assert_eq!(book.best_ask_price(), Some(Price::from("100.6")));
        assert!(receiver.try_recv().is_err());
    }

    #[rstest]
    fn test_spot_data_client_start_stop() {
        setup_test_env();
        let config = KrakenDataClientConfig::default();
        let mut client = KrakenSpotDataClient::new(*KRAKEN_CLIENT_ID, config).unwrap();

        assert!(client.start().is_ok());
        assert!(client.stop().is_ok());
        assert!(client.is_disconnected());
    }

    fn book_level(price: Decimal, qty: Decimal) -> KrakenWsBookLevel {
        KrakenWsBookLevel { price, qty }
    }
}