nautilus_kraken/python/
websocket_futures.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Python bindings for the Kraken Futures WebSocket client.
17
18use std::sync::{
19    Arc, RwLock,
20    atomic::{AtomicU64, Ordering},
21};
22
23use ahash::AHashMap;
24use nautilus_common::live::get_runtime;
25use nautilus_core::{
26    AtomicMap, UnixNanos,
27    python::{call_python_threadsafe, to_pyruntime_err},
28    time::get_atomic_clock_realtime,
29};
30use nautilus_model::{
31    data::{Data, OrderBookDeltas, OrderBookDeltas_API, QuoteTick},
32    enums::{BookType, OrderSide, OrderStatus, OrderType, TimeInForce},
33    identifiers::{AccountId, ClientOrderId, InstrumentId, StrategyId, TraderId, VenueOrderId},
34    instruments::{Instrument, InstrumentAny},
35    orderbook::OrderBook,
36    python::{data::data_to_pycapsule, instruments::pyobject_to_instrument_any},
37    reports::{FillReport, OrderStatusReport},
38    types::Quantity,
39};
40use nautilus_network::websocket::{SubscriptionState, TransportBackend};
41use pyo3::{IntoPyObjectExt, prelude::*};
42
43use crate::{
44    common::{
45        credential::KrakenCredential,
46        enums::{KrakenEnvironment, KrakenProductType},
47        lookup_instrument_in_snapshot,
48        urls::get_kraken_ws_public_url,
49    },
50    websocket::futures::{
51        client::KrakenFuturesWebSocketClient,
52        messages::{
53            KrakenFuturesBookDelta, KrakenFuturesBookSnapshot, KrakenFuturesFillsDelta,
54            KrakenFuturesOpenOrdersCancel, KrakenFuturesOpenOrdersDelta, KrakenFuturesTickerData,
55            KrakenFuturesTradeData, KrakenFuturesWsMessage,
56        },
57        parse::{
58            parse_futures_ws_book_delta, parse_futures_ws_book_snapshot_deltas,
59            parse_futures_ws_fill_report, parse_futures_ws_funding_rate,
60            parse_futures_ws_index_price, parse_futures_ws_mark_price,
61            parse_futures_ws_order_status_report, parse_futures_ws_trade_tick,
62        },
63    },
64};
65
66#[pymethods]
67#[pyo3_stub_gen::derive::gen_stub_pymethods]
68impl KrakenFuturesWebSocketClient {
69    /// WebSocket client for the Kraken Futures v1 streaming API.
70    #[new]
71    #[pyo3(signature = (environment=None, base_url=None, heartbeat_secs=60, api_key=None, api_secret=None, proxy_url=None))]
72    fn py_new(
73        environment: Option<KrakenEnvironment>,
74        base_url: Option<String>,
75        heartbeat_secs: u64,
76        api_key: Option<String>,
77        api_secret: Option<String>,
78        proxy_url: Option<String>,
79    ) -> Self {
80        let env = environment.unwrap_or(KrakenEnvironment::Live);
81        let demo = env == KrakenEnvironment::Demo;
82        let url = base_url.unwrap_or_else(|| {
83            get_kraken_ws_public_url(KrakenProductType::Futures, env).to_string()
84        });
85        let credential = KrakenCredential::resolve_futures(api_key, api_secret, demo);
86
87        Self::with_credentials(
88            url,
89            heartbeat_secs,
90            credential,
91            TransportBackend::default(),
92            proxy_url,
93        )
94    }
95
96    /// Returns true if the client has API credentials set.
97    #[getter]
98    #[pyo3(name = "has_credentials")]
99    #[must_use]
100    pub fn py_has_credentials(&self) -> bool {
101        self.has_credentials()
102    }
103
104    /// Returns the WebSocket URL.
105    #[getter]
106    #[pyo3(name = "url")]
107    #[must_use]
108    pub fn py_url(&self) -> &str {
109        self.url()
110    }
111
112    /// Returns true if the connection is closed.
113    #[pyo3(name = "is_closed")]
114    fn py_is_closed(&self) -> bool {
115        self.is_closed()
116    }
117
118    /// Returns true if the connection is active.
119    #[pyo3(name = "is_active")]
120    fn py_is_active(&self) -> bool {
121        self.is_active()
122    }
123
124    /// Waits until the WebSocket connection is active or timeout.
125    #[pyo3(name = "wait_until_active")]
126    fn py_wait_until_active<'py>(
127        &self,
128        py: Python<'py>,
129        timeout_secs: f64,
130    ) -> PyResult<Bound<'py, PyAny>> {
131        let client = self.clone();
132
133        pyo3_async_runtimes::tokio::future_into_py(py, async move {
134            client
135                .wait_until_active(timeout_secs)
136                .await
137                .map_err(to_pyruntime_err)?;
138            Ok(())
139        })
140    }
141
142    /// Returns true if the WebSocket is authenticated for private feeds.
143    #[pyo3(name = "is_authenticated")]
144    fn py_is_authenticated(&self) -> bool {
145        self.is_authenticated()
146    }
147
148    /// Waits until the WebSocket is authenticated or the timeout elapses.
149    ///
150    /// Returns an error on timeout or explicit auth failure.
151    #[pyo3(name = "wait_until_authenticated")]
152    fn py_wait_until_authenticated<'py>(
153        &self,
154        py: Python<'py>,
155        timeout_secs: f64,
156    ) -> PyResult<Bound<'py, PyAny>> {
157        let client = self.clone();
158
159        pyo3_async_runtimes::tokio::future_into_py(py, async move {
160            client
161                .wait_until_authenticated(timeout_secs)
162                .await
163                .map_err(to_pyruntime_err)?;
164            Ok(())
165        })
166    }
167
168    /// Authenticates the WebSocket connection for private feeds.
169    ///
170    /// Sends a challenge request and waits for the handler to parse the response,
171    /// sign it, and mark the `AuthTracker` successful. Private subscriptions gate
172    /// on the stored challenge / signed-challenge pair.
173    #[pyo3(name = "authenticate")]
174    fn py_authenticate<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
175        let client = self.clone();
176
177        pyo3_async_runtimes::tokio::future_into_py(py, async move {
178            client.authenticate().await.map_err(to_pyruntime_err)?;
179            Ok(())
180        })
181    }
182
183    /// Connects to the WebSocket server.
184    #[pyo3(name = "connect")]
185    #[expect(clippy::needless_pass_by_value)]
186    fn py_connect<'py>(
187        &mut self,
188        py: Python<'py>,
189        loop_: Py<PyAny>,
190        instruments: Vec<Py<PyAny>>,
191        callback: Py<PyAny>,
192    ) -> PyResult<Bound<'py, PyAny>> {
193        let call_soon: Py<PyAny> = loop_.getattr(py, "call_soon_threadsafe")?;
194
195        for inst in instruments {
196            let inst_any = pyobject_to_instrument_any(py, inst)?;
197            self.cache_instrument(inst_any);
198        }
199
200        let instruments_map = self.instruments_shared().clone();
201        let subscriptions = self.subscriptions().clone();
202        let account_id = self.account_id_shared().clone();
203        let truncated_id_map = self.truncated_id_map().clone();
204        let order_instrument_map = self.order_instrument_map().clone();
205        let mut client = self.clone();
206
207        pyo3_async_runtimes::tokio::future_into_py(py, async move {
208            client.connect().await.map_err(to_pyruntime_err)?;
209
210            if let Some(mut rx) = client.take_output_rx() {
211                let clock = get_atomic_clock_realtime();
212                let book_sequence = Arc::new(AtomicU64::new(0));
213
214                get_runtime().spawn(async move {
215                    let mut order_books: AHashMap<InstrumentId, OrderBook> = AHashMap::new();
216                    let mut last_quotes: AHashMap<InstrumentId, QuoteTick> = AHashMap::new();
217                    let venue_client_map: Arc<AtomicMap<String, ClientOrderId>> =
218                        Arc::new(AtomicMap::new());
219                    let venue_order_qty: Arc<AtomicMap<String, Quantity>> =
220                        Arc::new(AtomicMap::new());
221
222                    while let Some(msg) = rx.recv().await {
223                        let ts_init = clock.get_time_ns();
224
225                        match msg {
226                            KrakenFuturesWsMessage::OpenOrdersDelta(delta) => {
227                                handle_open_orders_delta(
228                                    &delta,
229                                    &instruments_map,
230                                    &account_id,
231                                    &truncated_id_map,
232                                    &order_instrument_map,
233                                    &venue_client_map,
234                                    &venue_order_qty,
235                                    ts_init,
236                                    &call_soon,
237                                    &callback,
238                                );
239                            }
240                            KrakenFuturesWsMessage::OpenOrdersCancel(cancel) => {
241                                handle_open_orders_cancel(
242                                    &cancel,
243                                    &account_id,
244                                    &truncated_id_map,
245                                    &order_instrument_map,
246                                    &venue_client_map,
247                                    &venue_order_qty,
248                                    ts_init,
249                                    &call_soon,
250                                    &callback,
251                                );
252                            }
253                            KrakenFuturesWsMessage::FillsDelta(fills_delta) => {
254                                handle_fills_delta(
255                                    &fills_delta,
256                                    &instruments_map,
257                                    &account_id,
258                                    &truncated_id_map,
259                                    ts_init,
260                                    &call_soon,
261                                    &callback,
262                                );
263                            }
264                            KrakenFuturesWsMessage::Ticker(ref ticker) => {
265                                handle_ticker(
266                                    ticker,
267                                    &instruments_map,
268                                    ts_init,
269                                    &call_soon,
270                                    &callback,
271                                );
272                            }
273                            KrakenFuturesWsMessage::Trade(ref trade) => {
274                                handle_trade(
275                                    trade,
276                                    &instruments_map,
277                                    ts_init,
278                                    &call_soon,
279                                    &callback,
280                                );
281                            }
282                            KrakenFuturesWsMessage::BookSnapshot(ref snapshot) => {
283                                handle_book_snapshot(
284                                    snapshot,
285                                    &instruments_map,
286                                    &subscriptions,
287                                    &mut order_books,
288                                    &mut last_quotes,
289                                    &book_sequence,
290                                    ts_init,
291                                    &call_soon,
292                                    &callback,
293                                );
294                            }
295                            KrakenFuturesWsMessage::BookDelta(ref delta) => {
296                                handle_book_delta(
297                                    delta,
298                                    &instruments_map,
299                                    &subscriptions,
300                                    &mut order_books,
301                                    &mut last_quotes,
302                                    &book_sequence,
303                                    ts_init,
304                                    &call_soon,
305                                    &callback,
306                                );
307                            }
308                            KrakenFuturesWsMessage::Challenge(_)
309                            | KrakenFuturesWsMessage::Reconnected => {}
310                        }
311                    }
312                });
313            }
314
315            Ok(())
316        })
317    }
318
319    /// Sets the account ID for execution report parsing.
320    #[pyo3(name = "set_account_id")]
321    fn py_set_account_id(&self, account_id: AccountId) {
322        self.set_account_id(account_id);
323    }
324
325    /// Caches an instrument for execution report parsing.
326    #[pyo3(name = "cache_instrument")]
327    fn py_cache_instrument(&self, py: Python, instrument: Py<PyAny>) -> PyResult<()> {
328        let inst_any = pyobject_to_instrument_any(py, instrument)?;
329        self.cache_instrument(inst_any);
330        Ok(())
331    }
332
333    /// Caches multiple instruments for execution report parsing.
334    #[pyo3(name = "cache_instruments")]
335    fn py_cache_instruments(&self, py: Python, instruments: Vec<Py<PyAny>>) -> PyResult<()> {
336        let mut inst_vec = Vec::with_capacity(instruments.len());
337        for inst in instruments {
338            inst_vec.push(pyobject_to_instrument_any(py, inst)?);
339        }
340        self.cache_instruments(&inst_vec);
341        Ok(())
342    }
343
344    /// Caches a client order for truncated ID resolution and instrument lookup.
345    ///
346    /// Kraken Futures limits client order IDs to 18 characters, so orders with
347    /// longer IDs are truncated. This method stores the mapping from truncated
348    /// to full ID, and from venue order ID to instrument ID for cancel messages.
349    #[pyo3(name = "cache_client_order")]
350    fn py_cache_client_order(
351        &self,
352        client_order_id: ClientOrderId,
353        venue_order_id: Option<VenueOrderId>,
354        instrument_id: InstrumentId,
355        trader_id: TraderId,
356        strategy_id: StrategyId,
357    ) {
358        self.cache_client_order(
359            client_order_id,
360            venue_order_id,
361            instrument_id,
362            trader_id,
363            strategy_id,
364        );
365    }
366
367    /// Disconnects from the WebSocket server.
368    #[pyo3(name = "disconnect")]
369    fn py_disconnect<'py>(&mut self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
370        let mut client = self.clone();
371
372        pyo3_async_runtimes::tokio::future_into_py(py, async move {
373            client.disconnect().await.map_err(to_pyruntime_err)?;
374            Ok(())
375        })
376    }
377
378    /// Closes the WebSocket connection.
379    #[pyo3(name = "close")]
380    fn py_close<'py>(&mut self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
381        let mut client = self.clone();
382
383        pyo3_async_runtimes::tokio::future_into_py(py, async move {
384            client.close().await.map_err(to_pyruntime_err)?;
385            Ok(())
386        })
387    }
388
389    /// Subscribes to order book updates for the given instrument.
390    ///
391    /// Note: The `depth` parameter is accepted for API compatibility with spot client but is
392    /// not used by Kraken Futures (full book is always returned).
393    #[pyo3(name = "subscribe_book")]
394    #[pyo3(signature = (instrument_id, depth=None))]
395    fn py_subscribe_book<'py>(
396        &self,
397        py: Python<'py>,
398        instrument_id: InstrumentId,
399        depth: Option<u32>,
400    ) -> PyResult<Bound<'py, PyAny>> {
401        let client = self.clone();
402
403        pyo3_async_runtimes::tokio::future_into_py(py, async move {
404            client
405                .subscribe_book(instrument_id, depth)
406                .await
407                .map_err(to_pyruntime_err)?;
408            Ok(())
409        })
410    }
411
412    /// Subscribes to quote updates for the given instrument.
413    ///
414    /// Uses the order book channel for low-latency top-of-book quotes.
415    #[pyo3(name = "subscribe_quotes")]
416    fn py_subscribe_quotes<'py>(
417        &self,
418        py: Python<'py>,
419        instrument_id: InstrumentId,
420    ) -> PyResult<Bound<'py, PyAny>> {
421        let client = self.clone();
422
423        pyo3_async_runtimes::tokio::future_into_py(py, async move {
424            client
425                .subscribe_quotes(instrument_id)
426                .await
427                .map_err(to_pyruntime_err)?;
428            Ok(())
429        })
430    }
431
432    /// Subscribes to trade updates for the given instrument.
433    #[pyo3(name = "subscribe_trades")]
434    fn py_subscribe_trades<'py>(
435        &self,
436        py: Python<'py>,
437        instrument_id: InstrumentId,
438    ) -> PyResult<Bound<'py, PyAny>> {
439        let client = self.clone();
440
441        pyo3_async_runtimes::tokio::future_into_py(py, async move {
442            client
443                .subscribe_trades(instrument_id)
444                .await
445                .map_err(to_pyruntime_err)?;
446            Ok(())
447        })
448    }
449
450    /// Subscribes to mark price updates for the given instrument.
451    #[pyo3(name = "subscribe_mark_price")]
452    fn py_subscribe_mark_price<'py>(
453        &self,
454        py: Python<'py>,
455        instrument_id: InstrumentId,
456    ) -> PyResult<Bound<'py, PyAny>> {
457        let client = self.clone();
458
459        pyo3_async_runtimes::tokio::future_into_py(py, async move {
460            client
461                .subscribe_mark_price(instrument_id)
462                .await
463                .map_err(to_pyruntime_err)?;
464            Ok(())
465        })
466    }
467
468    /// Subscribes to index price updates for the given instrument.
469    #[pyo3(name = "subscribe_index_price")]
470    fn py_subscribe_index_price<'py>(
471        &self,
472        py: Python<'py>,
473        instrument_id: InstrumentId,
474    ) -> PyResult<Bound<'py, PyAny>> {
475        let client = self.clone();
476
477        pyo3_async_runtimes::tokio::future_into_py(py, async move {
478            client
479                .subscribe_index_price(instrument_id)
480                .await
481                .map_err(to_pyruntime_err)?;
482            Ok(())
483        })
484    }
485
486    /// Subscribes to funding rate updates for the given instrument.
487    #[pyo3(name = "subscribe_funding_rate")]
488    fn py_subscribe_funding_rate<'py>(
489        &self,
490        py: Python<'py>,
491        instrument_id: InstrumentId,
492    ) -> PyResult<Bound<'py, PyAny>> {
493        let client = self.clone();
494
495        pyo3_async_runtimes::tokio::future_into_py(py, async move {
496            client
497                .subscribe_funding_rate(instrument_id)
498                .await
499                .map_err(to_pyruntime_err)?;
500            Ok(())
501        })
502    }
503
504    /// Unsubscribes from order book updates for the given instrument.
505    #[pyo3(name = "unsubscribe_book")]
506    fn py_unsubscribe_book<'py>(
507        &self,
508        py: Python<'py>,
509        instrument_id: InstrumentId,
510    ) -> PyResult<Bound<'py, PyAny>> {
511        let client = self.clone();
512
513        pyo3_async_runtimes::tokio::future_into_py(py, async move {
514            client
515                .unsubscribe_book(instrument_id)
516                .await
517                .map_err(to_pyruntime_err)?;
518            Ok(())
519        })
520    }
521
522    /// Unsubscribes from quote updates for the given instrument.
523    #[pyo3(name = "unsubscribe_quotes")]
524    fn py_unsubscribe_quotes<'py>(
525        &self,
526        py: Python<'py>,
527        instrument_id: InstrumentId,
528    ) -> PyResult<Bound<'py, PyAny>> {
529        let client = self.clone();
530
531        pyo3_async_runtimes::tokio::future_into_py(py, async move {
532            client
533                .unsubscribe_quotes(instrument_id)
534                .await
535                .map_err(to_pyruntime_err)?;
536            Ok(())
537        })
538    }
539
540    /// Unsubscribes from trade updates for the given instrument.
541    #[pyo3(name = "unsubscribe_trades")]
542    fn py_unsubscribe_trades<'py>(
543        &self,
544        py: Python<'py>,
545        instrument_id: InstrumentId,
546    ) -> PyResult<Bound<'py, PyAny>> {
547        let client = self.clone();
548
549        pyo3_async_runtimes::tokio::future_into_py(py, async move {
550            client
551                .unsubscribe_trades(instrument_id)
552                .await
553                .map_err(to_pyruntime_err)?;
554            Ok(())
555        })
556    }
557
558    /// Unsubscribes from mark price updates for the given instrument.
559    #[pyo3(name = "unsubscribe_mark_price")]
560    fn py_unsubscribe_mark_price<'py>(
561        &self,
562        py: Python<'py>,
563        instrument_id: InstrumentId,
564    ) -> PyResult<Bound<'py, PyAny>> {
565        let client = self.clone();
566
567        pyo3_async_runtimes::tokio::future_into_py(py, async move {
568            client
569                .unsubscribe_mark_price(instrument_id)
570                .await
571                .map_err(to_pyruntime_err)?;
572            Ok(())
573        })
574    }
575
576    /// Unsubscribes from index price updates for the given instrument.
577    #[pyo3(name = "unsubscribe_index_price")]
578    fn py_unsubscribe_index_price<'py>(
579        &self,
580        py: Python<'py>,
581        instrument_id: InstrumentId,
582    ) -> PyResult<Bound<'py, PyAny>> {
583        let client = self.clone();
584
585        pyo3_async_runtimes::tokio::future_into_py(py, async move {
586            client
587                .unsubscribe_index_price(instrument_id)
588                .await
589                .map_err(to_pyruntime_err)?;
590            Ok(())
591        })
592    }
593
594    /// Unsubscribes from funding rate updates for the given instrument.
595    #[pyo3(name = "unsubscribe_funding_rate")]
596    fn py_unsubscribe_funding_rate<'py>(
597        &self,
598        py: Python<'py>,
599        instrument_id: InstrumentId,
600    ) -> PyResult<Bound<'py, PyAny>> {
601        let client = self.clone();
602
603        pyo3_async_runtimes::tokio::future_into_py(py, async move {
604            client
605                .unsubscribe_funding_rate(instrument_id)
606                .await
607                .map_err(to_pyruntime_err)?;
608            Ok(())
609        })
610    }
611
612    /// Sign a challenge with the API credentials.
613    ///
614    /// Returns the signed challenge on success.
615    #[pyo3(name = "sign_challenge")]
616    fn py_sign_challenge(&self, challenge: &str) -> PyResult<String> {
617        self.sign_challenge(challenge).map_err(to_pyruntime_err)
618    }
619
620    /// Complete authentication with a received challenge.
621    #[pyo3(name = "authenticate_with_challenge")]
622    fn py_authenticate_with_challenge<'py>(
623        &self,
624        py: Python<'py>,
625        challenge: String,
626    ) -> PyResult<Bound<'py, PyAny>> {
627        let client = self.clone();
628
629        pyo3_async_runtimes::tokio::future_into_py(py, async move {
630            client
631                .authenticate_with_challenge(&challenge)
632                .await
633                .map_err(to_pyruntime_err)?;
634            Ok(())
635        })
636    }
637
638    /// Set authentication credentials directly (for when challenge is obtained externally).
639    #[pyo3(name = "set_auth_credentials")]
640    fn py_set_auth_credentials<'py>(
641        &self,
642        py: Python<'py>,
643        original_challenge: String,
644        signed_challenge: String,
645    ) -> PyResult<Bound<'py, PyAny>> {
646        let client = self.clone();
647
648        pyo3_async_runtimes::tokio::future_into_py(py, async move {
649            client
650                .set_auth_credentials(original_challenge, signed_challenge)
651                .await
652                .map_err(to_pyruntime_err)?;
653            Ok(())
654        })
655    }
656
657    /// Subscribes to open orders feed (private, requires authentication).
658    #[pyo3(name = "subscribe_open_orders")]
659    fn py_subscribe_open_orders<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
660        let client = self.clone();
661
662        pyo3_async_runtimes::tokio::future_into_py(py, async move {
663            client
664                .subscribe_open_orders()
665                .await
666                .map_err(to_pyruntime_err)?;
667            Ok(())
668        })
669    }
670
671    /// Subscribes to fills feed (private, requires authentication).
672    #[pyo3(name = "subscribe_fills")]
673    fn py_subscribe_fills<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
674        let client = self.clone();
675
676        pyo3_async_runtimes::tokio::future_into_py(py, async move {
677            client.subscribe_fills().await.map_err(to_pyruntime_err)?;
678            Ok(())
679        })
680    }
681
682    /// Subscribes to both open orders and fills (convenience method).
683    #[pyo3(name = "subscribe_executions")]
684    fn py_subscribe_executions<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
685        let client = self.clone();
686
687        pyo3_async_runtimes::tokio::future_into_py(py, async move {
688            client
689                .subscribe_executions()
690                .await
691                .map_err(to_pyruntime_err)?;
692            Ok(())
693        })
694    }
695}
696
697fn resolve_client_order_id(
698    truncated: &str,
699    truncated_id_map: &Arc<AtomicMap<String, ClientOrderId>>,
700) -> ClientOrderId {
701    truncated_id_map
702        .load()
703        .get(truncated)
704        .copied()
705        .unwrap_or_else(|| ClientOrderId::new(truncated))
706}
707
708fn dispatch_report_to_python(
709    report: OrderStatusReport,
710    call_soon: &Py<PyAny>,
711    callback: &Py<PyAny>,
712) {
713    Python::attach(|py| match report.into_py_any(py) {
714        Ok(py_obj) => call_python_threadsafe(py, call_soon, callback, py_obj),
715        Err(e) => log::error!("Failed to convert OrderStatusReport to Python: {e}"),
716    });
717}
718
719fn dispatch_fill_to_python(report: FillReport, call_soon: &Py<PyAny>, callback: &Py<PyAny>) {
720    Python::attach(|py| match report.into_py_any(py) {
721        Ok(py_obj) => call_python_threadsafe(py, call_soon, callback, py_obj),
722        Err(e) => log::error!("Failed to convert FillReport to Python: {e}"),
723    });
724}
725
726#[expect(clippy::too_many_arguments)]
727fn handle_open_orders_delta(
728    delta: &KrakenFuturesOpenOrdersDelta,
729    instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
730    account_id: &Arc<RwLock<Option<AccountId>>>,
731    truncated_id_map: &Arc<AtomicMap<String, ClientOrderId>>,
732    order_instrument_map: &Arc<AtomicMap<String, InstrumentId>>,
733    venue_client_map: &Arc<AtomicMap<String, ClientOrderId>>,
734    venue_order_qty: &Arc<AtomicMap<String, Quantity>>,
735    ts_init: UnixNanos,
736    call_soon: &Py<PyAny>,
737    callback: &Py<PyAny>,
738) {
739    // The fills delta carries the real fill; skip the cancel-shaped delta
740    // Kraken emits when an order leaves the book because it filled.
741    if delta.is_fill_driven_cancel() {
742        log::debug!(
743            "Skipping fill-driven open_orders delta: order_id={}, reason={:?}",
744            delta.order.order_id,
745            delta.reason,
746        );
747        return;
748    }
749
750    let product_id = delta.order.instrument.as_str();
751
752    let instruments = instruments.load();
753    let Some(instrument) = lookup_instrument_in_snapshot(&instruments, product_id) else {
754        log::warn!("No instrument for product_id: {product_id}");
755        return;
756    };
757
758    let Some(acct_id) = account_id.read().ok().and_then(|g| *g) else {
759        log::warn!("Account ID not set, cannot process order delta");
760        return;
761    };
762
763    order_instrument_map.insert(delta.order.order_id.clone(), instrument.id());
764
765    let qty = Quantity::new(delta.order.qty, instrument.size_precision());
766    venue_order_qty.insert(delta.order.order_id.clone(), qty);
767
768    match parse_futures_ws_order_status_report(
769        &delta.order,
770        delta.is_cancel,
771        delta.reason.as_deref(),
772        instrument,
773        acct_id,
774        ts_init,
775    ) {
776        Ok(mut report) => {
777            if let Some(ref cl_ord_id) = delta.order.cli_ord_id {
778                let full_id = resolve_client_order_id(cl_ord_id, truncated_id_map);
779                report = report.with_client_order_id(full_id);
780
781                venue_client_map.insert(delta.order.order_id.clone(), full_id);
782            }
783            dispatch_report_to_python(report, call_soon, callback);
784        }
785        Err(e) => log::error!("Failed to parse futures order status report: {e}"),
786    }
787}
788
789#[expect(clippy::too_many_arguments)]
790fn handle_open_orders_cancel(
791    cancel: &KrakenFuturesOpenOrdersCancel,
792    account_id: &Arc<RwLock<Option<AccountId>>>,
793    truncated_id_map: &Arc<AtomicMap<String, ClientOrderId>>,
794    order_instrument_map: &Arc<AtomicMap<String, InstrumentId>>,
795    venue_client_map: &Arc<AtomicMap<String, ClientOrderId>>,
796    venue_order_qty: &Arc<AtomicMap<String, Quantity>>,
797    ts_init: UnixNanos,
798    call_soon: &Py<PyAny>,
799    callback: &Py<PyAny>,
800) {
801    if let Some(ref reason) = cancel.reason
802        && (reason == "full_fill" || reason == "partial_fill")
803    {
804        log::debug!(
805            "Skipping fill-driven cancel: order_id={}, reason={reason}",
806            cancel.order_id,
807        );
808        return;
809    }
810
811    let Some(acct_id) = account_id.read().ok().and_then(|g| *g) else {
812        log::warn!("Account ID not set, cannot process order cancel");
813        return;
814    };
815
816    let venue_order_id = VenueOrderId::new(&cancel.order_id);
817
818    let instrument_id = order_instrument_map.load().get(&cancel.order_id).copied();
819
820    let Some(instrument_id) = instrument_id else {
821        log::warn!(
822            "Cannot resolve instrument for cancel: order_id={}, \
823             order not seen in previous delta",
824            cancel.order_id
825        );
826        return;
827    };
828
829    let client_order_id = cancel
830        .cli_ord_id
831        .as_ref()
832        .map(|id| resolve_client_order_id(id, truncated_id_map))
833        .or_else(|| venue_client_map.load().get(&cancel.order_id).copied());
834
835    let Some(quantity) = venue_order_qty.load().get(&cancel.order_id).copied() else {
836        log::warn!(
837            "Cannot resolve quantity for cancel: order_id={}, skipping",
838            cancel.order_id
839        );
840        return;
841    };
842
843    let report = OrderStatusReport::new(
844        acct_id,
845        instrument_id,
846        client_order_id,
847        venue_order_id,
848        OrderSide::NoOrderSide,
849        OrderType::Limit,
850        TimeInForce::Gtc,
851        OrderStatus::Canceled,
852        quantity,
853        Quantity::zero(0),
854        ts_init,
855        ts_init,
856        ts_init,
857        None,
858    );
859
860    let report = if let Some(ref reason) = cancel.reason
861        && !reason.is_empty()
862    {
863        report.with_cancel_reason(reason.clone())
864    } else {
865        report
866    };
867
868    dispatch_report_to_python(report, call_soon, callback);
869}
870
871fn handle_fills_delta(
872    fills_delta: &KrakenFuturesFillsDelta,
873    instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
874    account_id: &Arc<RwLock<Option<AccountId>>>,
875    truncated_id_map: &Arc<AtomicMap<String, ClientOrderId>>,
876    ts_init: UnixNanos,
877    call_soon: &Py<PyAny>,
878    callback: &Py<PyAny>,
879) {
880    let Some(acct_id) = account_id.read().ok().and_then(|g| *g) else {
881        log::warn!("Account ID not set, cannot process fills");
882        return;
883    };
884
885    let instruments = instruments.load();
886
887    for fill in &fills_delta.fills {
888        let product_id = match &fill.instrument {
889            Some(id) => id.as_str(),
890            None => {
891                log::warn!("Fill missing instrument field: fill_id={}", fill.fill_id);
892                continue;
893            }
894        };
895
896        let Some(instrument) = lookup_instrument_in_snapshot(&instruments, product_id) else {
897            log::warn!("No instrument for product_id: {product_id}");
898            continue;
899        };
900
901        match parse_futures_ws_fill_report(fill, instrument, acct_id, ts_init) {
902            Ok(mut report) => {
903                if let Some(ref cl_ord_id) = fill.cli_ord_id {
904                    let full_id = resolve_client_order_id(cl_ord_id, truncated_id_map);
905                    report.client_order_id = Some(full_id);
906                }
907                dispatch_fill_to_python(report, call_soon, callback);
908            }
909            Err(e) => log::error!("Failed to parse futures fill report: {e}"),
910        }
911    }
912}
913
914fn handle_ticker(
915    ticker: &KrakenFuturesTickerData,
916    instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
917    ts_init: UnixNanos,
918    call_soon: &Py<PyAny>,
919    callback: &Py<PyAny>,
920) {
921    let instruments = instruments.load();
922    let Some(instrument) = lookup_instrument_in_snapshot(&instruments, ticker.product_id.as_str())
923    else {
924        return;
925    };
926
927    if let Some(mark_price) = parse_futures_ws_mark_price(ticker, instrument, ts_init) {
928        Python::attach(|py| {
929            let py_obj = data_to_pycapsule(py, Data::MarkPriceUpdate(mark_price));
930            call_python_threadsafe(py, call_soon, callback, py_obj);
931        });
932    }
933
934    if let Some(index_price) = parse_futures_ws_index_price(ticker, instrument, ts_init) {
935        Python::attach(|py| {
936            let py_obj = data_to_pycapsule(py, Data::IndexPriceUpdate(index_price));
937            call_python_threadsafe(py, call_soon, callback, py_obj);
938        });
939    }
940
941    if let Some(funding_rate) = parse_futures_ws_funding_rate(ticker, instrument, ts_init) {
942        Python::attach(|py| match funding_rate.into_py_any(py) {
943            Ok(py_obj) => call_python_threadsafe(py, call_soon, callback, py_obj),
944            Err(e) => log::error!("Failed to convert FundingRateUpdate to Python: {e}"),
945        });
946    }
947}
948
949fn handle_trade(
950    trade: &KrakenFuturesTradeData,
951    instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
952    ts_init: UnixNanos,
953    call_soon: &Py<PyAny>,
954    callback: &Py<PyAny>,
955) {
956    let instruments = instruments.load();
957    let Some(instrument) = lookup_instrument_in_snapshot(&instruments, trade.product_id.as_str())
958    else {
959        return;
960    };
961
962    match parse_futures_ws_trade_tick(trade, instrument, ts_init) {
963        Ok(tick) => {
964            Python::attach(|py| {
965                let py_obj = data_to_pycapsule(py, Data::Trade(tick));
966                call_python_threadsafe(py, call_soon, callback, py_obj);
967            });
968        }
969        Err(e) => log::error!("Failed to parse futures trade tick: {e}"),
970    }
971}
972
973#[expect(clippy::too_many_arguments)]
974fn handle_book_snapshot(
975    snapshot: &KrakenFuturesBookSnapshot,
976    instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
977    subscriptions: &SubscriptionState,
978    order_books: &mut AHashMap<InstrumentId, OrderBook>,
979    last_quotes: &mut AHashMap<InstrumentId, QuoteTick>,
980    book_sequence: &Arc<AtomicU64>,
981    ts_init: UnixNanos,
982    call_soon: &Py<PyAny>,
983    callback: &Py<PyAny>,
984) {
985    let instruments = instruments.load();
986    let Some(instrument) =
987        lookup_instrument_in_snapshot(&instruments, snapshot.product_id.as_str())
988    else {
989        return;
990    };
991    let instrument_id = instrument.id();
992
993    let sequence = book_sequence.fetch_add(
994        (snapshot.bids.len() + snapshot.asks.len() + 1) as u64,
995        Ordering::Relaxed,
996    );
997
998    match parse_futures_ws_book_snapshot_deltas(snapshot, instrument, sequence, ts_init) {
999        Ok(delta_vec) => {
1000            if delta_vec.is_empty() {
1001                return;
1002            }
1003            let deltas = OrderBookDeltas::new(instrument_id, delta_vec);
1004
1005            let quotes_key = format!("quotes:{}", snapshot.product_id);
1006            if subscriptions.get_reference_count(&quotes_key) > 0 {
1007                let book = order_books
1008                    .entry(instrument_id)
1009                    .or_insert_with(|| OrderBook::new(instrument_id, BookType::L2_MBP));
1010
1011                if let Err(e) = book.apply_deltas(&deltas) {
1012                    log::error!("Failed to apply snapshot deltas to order book: {e}");
1013                } else {
1014                    maybe_emit_quote(
1015                        book,
1016                        instrument_id,
1017                        last_quotes,
1018                        ts_init,
1019                        call_soon,
1020                        callback,
1021                    );
1022                }
1023            }
1024
1025            let deltas_key = format!("deltas:{}", snapshot.product_id);
1026            if subscriptions.get_reference_count(&deltas_key) > 0 {
1027                Python::attach(|py| {
1028                    let py_obj =
1029                        data_to_pycapsule(py, Data::Deltas(OrderBookDeltas_API::new(deltas)));
1030                    call_python_threadsafe(py, call_soon, callback, py_obj);
1031                });
1032            }
1033        }
1034        Err(e) => log::error!("Failed to parse futures book snapshot: {e}"),
1035    }
1036}
1037
1038#[expect(clippy::too_many_arguments)]
1039fn handle_book_delta(
1040    delta: &KrakenFuturesBookDelta,
1041    instruments: &Arc<AtomicMap<InstrumentId, InstrumentAny>>,
1042    subscriptions: &SubscriptionState,
1043    order_books: &mut AHashMap<InstrumentId, OrderBook>,
1044    last_quotes: &mut AHashMap<InstrumentId, QuoteTick>,
1045    book_sequence: &Arc<AtomicU64>,
1046    ts_init: UnixNanos,
1047    call_soon: &Py<PyAny>,
1048    callback: &Py<PyAny>,
1049) {
1050    let instruments = instruments.load();
1051    let Some(instrument) = lookup_instrument_in_snapshot(&instruments, delta.product_id.as_str())
1052    else {
1053        return;
1054    };
1055    let instrument_id = instrument.id();
1056
1057    let sequence = book_sequence.fetch_add(1, Ordering::Relaxed);
1058
1059    match parse_futures_ws_book_delta(delta, instrument, sequence, ts_init) {
1060        Ok(book_delta) => {
1061            let deltas = OrderBookDeltas::new(instrument_id, vec![book_delta]);
1062
1063            let quotes_key = format!("quotes:{}", delta.product_id);
1064            if subscriptions.get_reference_count(&quotes_key) > 0
1065                && let Some(book) = order_books.get_mut(&instrument_id)
1066            {
1067                if let Err(e) = book.apply_deltas(&deltas) {
1068                    log::error!("Failed to apply delta to order book: {e}");
1069                } else {
1070                    maybe_emit_quote(
1071                        book,
1072                        instrument_id,
1073                        last_quotes,
1074                        ts_init,
1075                        call_soon,
1076                        callback,
1077                    );
1078                }
1079            }
1080
1081            let deltas_key = format!("deltas:{}", delta.product_id);
1082            if subscriptions.get_reference_count(&deltas_key) > 0 {
1083                Python::attach(|py| {
1084                    let py_obj =
1085                        data_to_pycapsule(py, Data::Deltas(OrderBookDeltas_API::new(deltas)));
1086                    call_python_threadsafe(py, call_soon, callback, py_obj);
1087                });
1088            }
1089        }
1090        Err(e) => log::error!("Failed to parse futures book delta: {e}"),
1091    }
1092}
1093
1094fn maybe_emit_quote(
1095    book: &OrderBook,
1096    instrument_id: InstrumentId,
1097    last_quotes: &mut AHashMap<InstrumentId, QuoteTick>,
1098    ts_init: UnixNanos,
1099    call_soon: &Py<PyAny>,
1100    callback: &Py<PyAny>,
1101) {
1102    let (Some(bid_price), Some(ask_price)) = (book.best_bid_price(), book.best_ask_price()) else {
1103        return;
1104    };
1105    let (Some(bid_size), Some(ask_size)) = (book.best_bid_size(), book.best_ask_size()) else {
1106        return;
1107    };
1108
1109    let bid = bid_price.as_f64();
1110    let ask = ask_price.as_f64();
1111    if bid > 0.0 && (ask - bid) / bid > 0.25 {
1112        log::debug!("Filtered quote with wide spread: bid={bid}, ask={ask}");
1113        return;
1114    }
1115
1116    let quote = QuoteTick::new(
1117        instrument_id,
1118        bid_price,
1119        ask_price,
1120        bid_size,
1121        ask_size,
1122        ts_init,
1123        ts_init,
1124    );
1125
1126    if matches!(last_quotes.get(&instrument_id), Some(prev) if *prev == quote) {
1127        return;
1128    }
1129
1130    last_quotes.insert(instrument_id, quote);
1131
1132    Python::attach(|py| {
1133        let py_obj = data_to_pycapsule(py, Data::Quote(quote));
1134        call_python_threadsafe(py, call_soon, callback, py_obj);
1135    });
1136}
nautilus_kraken/python/websocket_futures.rs

nautilus_kraken/python/
websocket_futures.rs
