nautilus_hyperliquid/python/
websocket.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 Hyperliquid WebSocket client.
17
18use std::time::Duration;
19
20use nautilus_common::live::get_runtime;
21use nautilus_core::python::{
22    IntoPyObjectNautilusExt, call_python_threadsafe, to_pyruntime_err, to_pyvalue_err,
23};
24use nautilus_model::{
25    data::{BarType, Data, OrderBookDeltas_API},
26    enums::{OrderSide, OrderType, TimeInForce},
27    identifiers::{AccountId, ClientOrderId, InstrumentId, VenueOrderId},
28    orders::OrderAny,
29    python::{
30        data::data_to_pycapsule, instruments::pyobject_to_instrument_any,
31        orders::pyobject_to_order_any,
32    },
33    types::{Price, Quantity},
34};
35use nautilus_network::websocket::TransportBackend;
36use pyo3::{conversion::IntoPyObjectExt, prelude::*, types::PyList};
37
38use crate::{
39    common::enums::HyperliquidEnvironment,
40    http::client::HyperliquidHttpClient,
41    websocket::{
42        HyperliquidWebSocketClient,
43        messages::{ExecutionReport, NautilusWsMessage},
44    },
45};
46
47fn ws_data_to_pyobject(py: Python<'_>, data: Data) -> PyResult<Py<PyAny>> {
48    match data {
49        Data::Custom(custom) => Py::new(py, custom).map(|obj| obj.into_any()),
50        Data::OptionGreeks(greeks) => Py::new(py, greeks).map(|obj| obj.into_any()),
51        other => Ok(data_to_pycapsule(py, other)),
52    }
53}
54
55#[pymethods]
56#[pyo3_stub_gen::derive::gen_stub_pymethods]
57impl HyperliquidWebSocketClient {
58    /// Hyperliquid WebSocket client following the BitMEX pattern.
59    ///
60    /// Orchestrates WebSocket connection and subscriptions using a command-based architecture,
61    /// where the inner FeedHandler owns the WebSocketClient and handles all I/O.
62    #[new]
63    #[pyo3(signature = (url=None, environment=HyperliquidEnvironment::Mainnet, account_id=None, proxy_url=None))]
64    fn py_new(
65        url: Option<String>,
66        environment: HyperliquidEnvironment,
67        account_id: Option<String>,
68        proxy_url: Option<String>,
69    ) -> Self {
70        let account_id = account_id.map(|s| AccountId::from(s.as_str()));
71        Self::new(
72            url,
73            environment,
74            account_id,
75            TransportBackend::default(),
76            proxy_url,
77        )
78    }
79
80    /// Returns the URL of this WebSocket client.
81    #[getter]
82    #[pyo3(name = "url")]
83    #[must_use]
84    pub fn py_url(&self) -> String {
85        self.url().to_string()
86    }
87
88    /// Returns true if the WebSocket is actively connected.
89    #[pyo3(name = "is_active")]
90    fn py_is_active(&self) -> bool {
91        self.is_active()
92    }
93
94    #[pyo3(name = "is_closed")]
95    fn py_is_closed(&self) -> bool {
96        !self.is_active()
97    }
98
99    /// Sets the timeout for WebSocket post trading requests.
100    #[pyo3(name = "set_post_timeout")]
101    fn py_set_post_timeout(&mut self, timeout_secs: u64) {
102        self.set_post_timeout(Duration::from_secs(timeout_secs));
103    }
104
105    /// Submit an order through the Hyperliquid WebSocket post API.
106    ///
107    /// The HTTP client supplies signing credentials, builder attribution, and
108    /// cached instrument metadata. The action itself is sent over WebSocket.
109    ///
110    /// Returns an `OrderStatusReport` describing the venue's immediate
111    /// response (`Filled` for an atomic IOC fill, `Accepted` for a resting
112    /// order), or `None` when the venue deferred the order without an oid (for
113    /// example a `waitingForFill` trigger child): the order stays `SUBMITTED`
114    /// until the user-events stream delivers the first `OrderAccepted`.
115    #[pyo3(name = "submit_order", signature = (
116        signer,
117        instrument_id,
118        client_order_id,
119        order_side,
120        order_type,
121        quantity,
122        time_in_force,
123        price=None,
124        trigger_price=None,
125        post_only=false,
126        reduce_only=false,
127    ))]
128    #[expect(clippy::too_many_arguments)]
129    fn py_submit_order<'py>(
130        &self,
131        py: Python<'py>,
132        signer: &HyperliquidHttpClient,
133        instrument_id: InstrumentId,
134        client_order_id: ClientOrderId,
135        order_side: OrderSide,
136        order_type: OrderType,
137        quantity: Quantity,
138        time_in_force: TimeInForce,
139        price: Option<Price>,
140        trigger_price: Option<Price>,
141        post_only: bool,
142        reduce_only: bool,
143    ) -> PyResult<Bound<'py, PyAny>> {
144        let client = self.clone();
145        let signer = signer.clone();
146
147        pyo3_async_runtimes::tokio::future_into_py(py, async move {
148            let report = client
149                .submit_order(
150                    &signer,
151                    instrument_id,
152                    client_order_id,
153                    order_side,
154                    order_type,
155                    quantity,
156                    time_in_force,
157                    price,
158                    trigger_price,
159                    post_only,
160                    reduce_only,
161                )
162                .await
163                .map_err(to_pyvalue_err)?;
164
165            Python::attach(|py| match report {
166                Some(r) => Ok(r.into_py_any_unwrap(py)),
167                None => Ok(py.None()),
168            })
169        })
170    }
171
172    /// Submit multiple orders through the Hyperliquid WebSocket post API.
173    ///
174    /// Returns one `OrderStatusReport` per accepted order in submission
175    /// order. Deferred trigger children of a `normalTpsl` bracket are absent
176    /// from the result; they stay `SUBMITTED` until the user-events stream
177    /// delivers an `OrderAccepted` with the real oid.
178    #[pyo3(name = "submit_orders")]
179    fn py_submit_orders<'py>(
180        &self,
181        py: Python<'py>,
182        signer: &HyperliquidHttpClient,
183        orders: Vec<Py<PyAny>>,
184    ) -> PyResult<Bound<'py, PyAny>> {
185        let client = self.clone();
186        let signer = signer.clone();
187
188        pyo3_async_runtimes::tokio::future_into_py(py, async move {
189            let order_anys: Vec<OrderAny> = Python::attach(|py| {
190                orders
191                    .into_iter()
192                    .map(|order| pyobject_to_order_any(py, order))
193                    .collect::<PyResult<Vec<_>>>()
194                    .map_err(to_pyvalue_err)
195            })?;
196            let order_refs: Vec<&OrderAny> = order_anys.iter().collect();
197
198            let reports = client
199                .submit_orders(&signer, &order_refs)
200                .await
201                .map_err(to_pyvalue_err)?;
202
203            Python::attach(|py| {
204                let pylist =
205                    PyList::new(py, reports.into_iter().map(|r| r.into_py_any_unwrap(py)))?;
206                Ok(pylist.into_py_any_unwrap(py))
207            })
208        })
209    }
210
211    /// Cancel an order through the Hyperliquid WebSocket post API.
212    #[pyo3(name = "cancel_order", signature = (
213        signer,
214        instrument_id,
215        client_order_id=None,
216        venue_order_id=None,
217    ))]
218    fn py_cancel_order<'py>(
219        &self,
220        py: Python<'py>,
221        signer: &HyperliquidHttpClient,
222        instrument_id: InstrumentId,
223        client_order_id: Option<ClientOrderId>,
224        venue_order_id: Option<VenueOrderId>,
225    ) -> PyResult<Bound<'py, PyAny>> {
226        let client = self.clone();
227        let signer = signer.clone();
228
229        pyo3_async_runtimes::tokio::future_into_py(py, async move {
230            client
231                .cancel_order(&signer, instrument_id, client_order_id, venue_order_id)
232                .await
233                .map_err(to_pyvalue_err)?;
234            Ok(())
235        })
236    }
237
238    /// Cancel multiple orders through one Hyperliquid WebSocket post action.
239    #[pyo3(name = "cancel_orders")]
240    fn py_cancel_orders<'py>(
241        &self,
242        py: Python<'py>,
243        signer: &HyperliquidHttpClient,
244        cancels: Vec<(InstrumentId, ClientOrderId, Option<VenueOrderId>)>,
245    ) -> PyResult<Bound<'py, PyAny>> {
246        let client = self.clone();
247        let signer = signer.clone();
248
249        pyo3_async_runtimes::tokio::future_into_py(py, async move {
250            client
251                .cancel_orders(&signer, &cancels)
252                .await
253                .map_err(to_pyvalue_err)
254        })
255    }
256
257    /// Modify an order through the Hyperliquid WebSocket post API.
258    #[pyo3(name = "modify_order")]
259    #[expect(clippy::too_many_arguments)]
260    fn py_modify_order<'py>(
261        &self,
262        py: Python<'py>,
263        signer: &HyperliquidHttpClient,
264        instrument_id: InstrumentId,
265        venue_order_id: VenueOrderId,
266        order_side: OrderSide,
267        order_type: OrderType,
268        price: Price,
269        quantity: Quantity,
270        trigger_price: Option<Price>,
271        reduce_only: bool,
272        post_only: bool,
273        time_in_force: TimeInForce,
274        client_order_id: Option<ClientOrderId>,
275    ) -> PyResult<Bound<'py, PyAny>> {
276        let client = self.clone();
277        let signer = signer.clone();
278
279        pyo3_async_runtimes::tokio::future_into_py(py, async move {
280            client
281                .modify_order(
282                    &signer,
283                    instrument_id,
284                    venue_order_id,
285                    order_side,
286                    order_type,
287                    price,
288                    quantity,
289                    trigger_price,
290                    reduce_only,
291                    post_only,
292                    time_in_force,
293                    client_order_id,
294                )
295                .await
296                .map_err(to_pyvalue_err)?;
297            Ok(())
298        })
299    }
300
301    /// Caches spot fill coin mappings for instrument lookup.
302    ///
303    /// Hyperliquid WebSocket fills for spot use `@{pair_index}` format (e.g., `@107`),
304    /// while instruments are identified by full symbols (e.g., `HYPE-USDC-SPOT`).
305    /// This mapping allows the handler to look up instruments from spot fills.
306    #[pyo3(name = "cache_spot_fill_coins")]
307    fn py_cache_spot_fill_coins(&self, mapping: std::collections::HashMap<String, String>) {
308        let ahash_mapping: ahash::AHashMap<ustr::Ustr, ustr::Ustr> = mapping
309            .into_iter()
310            .map(|(k, v)| (ustr::Ustr::from(&k), ustr::Ustr::from(&v)))
311            .collect();
312        self.cache_spot_fill_coins(ahash_mapping);
313    }
314
315    /// Cache the ordered instrument IDs required to normalize `allDexsAssetCtxs`.
316    #[pyo3(name = "cache_all_dex_asset_ctxs_instrument_ids")]
317    fn py_cache_all_dex_asset_ctxs_instrument_ids(
318        &self,
319        mapping: std::collections::HashMap<String, Vec<Option<InstrumentId>>>,
320    ) {
321        let ahash_mapping: ahash::AHashMap<ustr::Ustr, Vec<Option<InstrumentId>>> = mapping
322            .into_iter()
323            .map(|(dex, instrument_ids)| (ustr::Ustr::from(&dex), instrument_ids))
324            .collect();
325        self.cache_all_dex_asset_ctxs_instrument_ids(ahash_mapping);
326    }
327
328    /// Caches a venue CLOID to client_order_id mapping for order/fill resolution.
329    ///
330    /// This mapping allows WebSocket order status and fill reports to be resolved back to
331    /// the original client_order_id.
332    ///
333    /// This writes directly to a shared cache that the handler reads from, avoiding any
334    /// race conditions between caching and WebSocket message processing.
335    #[pyo3(name = "cache_cloid_mapping")]
336    fn py_cache_cloid_mapping(&self, cloid: &str, client_order_id: ClientOrderId) {
337        self.cache_cloid_mapping(ustr::Ustr::from(cloid), client_order_id);
338    }
339
340    /// Removes a cloid mapping from the cache.
341    ///
342    /// Called on terminal order state. The cache is FIFO-bounded so missed
343    /// removals self-evict (see GH-3972 cancel-replace drain).
344    #[pyo3(name = "remove_cloid_mapping")]
345    fn py_remove_cloid_mapping(&self, cloid: &str) {
346        self.remove_cloid_mapping(&ustr::Ustr::from(cloid));
347    }
348
349    /// Clears all cloid mappings from the cache.
350    ///
351    /// Useful for cleanup during reconnection or shutdown.
352    #[pyo3(name = "clear_cloid_cache")]
353    fn py_clear_cloid_cache(&self) {
354        self.clear_cloid_cache();
355    }
356
357    /// Returns the number of cloid mappings in the cache.
358    #[pyo3(name = "cloid_cache_len")]
359    fn py_cloid_cache_len(&self) -> usize {
360        self.cloid_cache_len()
361    }
362
363    /// Looks up a client_order_id by its venue CLOID.
364    ///
365    /// Returns `Some(ClientOrderId)` if the mapping exists, `None` otherwise.
366    #[pyo3(name = "get_cloid_mapping")]
367    fn py_get_cloid_mapping(&self, cloid: &str) -> Option<ClientOrderId> {
368        self.get_cloid_mapping(&ustr::Ustr::from(cloid))
369    }
370
371    /// Establishes WebSocket connection and spawns the message handler.
372    #[pyo3(name = "connect")]
373    #[expect(clippy::needless_pass_by_value)]
374    fn py_connect<'py>(
375        &self,
376        py: Python<'py>,
377        loop_: Py<PyAny>,
378        instruments: Vec<Py<PyAny>>,
379        callback: Py<PyAny>,
380    ) -> PyResult<Bound<'py, PyAny>> {
381        let call_soon: Py<PyAny> = loop_.getattr(py, "call_soon_threadsafe")?;
382
383        for inst in instruments {
384            let inst_any = pyobject_to_instrument_any(py, inst)?;
385            self.cache_instrument(inst_any);
386        }
387
388        let mut client = self.clone();
389
390        pyo3_async_runtimes::tokio::future_into_py(py, async move {
391            client.connect().await.map_err(to_pyruntime_err)?;
392
393            get_runtime().spawn(async move {
394                loop {
395                    let event = client.next_event().await;
396
397                    match event {
398                        Some(msg) => {
399                            log::trace!("Received WebSocket message: {msg:?}");
400
401                            match msg {
402                                NautilusWsMessage::Trades(trade_ticks) => {
403                                    Python::attach(|py| {
404                                        for tick in trade_ticks {
405                                            let py_obj = data_to_pycapsule(py, Data::Trade(tick));
406                                            call_python_threadsafe(py, &call_soon, &callback, py_obj);
407                                        }
408                                    });
409                                }
410                                NautilusWsMessage::Quote(quote_tick) => {
411                                    Python::attach(|py| {
412                                        let py_obj = data_to_pycapsule(py, Data::Quote(quote_tick));
413                                        call_python_threadsafe(py, &call_soon, &callback, py_obj);
414                                    });
415                                }
416                                NautilusWsMessage::Deltas(deltas) => {
417                                    Python::attach(|py| {
418                                        let py_obj = data_to_pycapsule(
419                                            py,
420                                            Data::Deltas(OrderBookDeltas_API::new(deltas)),
421                                        );
422                                        call_python_threadsafe(py, &call_soon, &callback, py_obj);
423                                    });
424                                }
425                                NautilusWsMessage::Depth10(depth) => {
426                                    Python::attach(|py| {
427                                        let py_obj = data_to_pycapsule(py, Data::Depth10(depth));
428                                        call_python_threadsafe(py, &call_soon, &callback, py_obj);
429                                    });
430                                }
431                                NautilusWsMessage::Candle(bar) => {
432                                    Python::attach(|py| {
433                                        let py_obj = data_to_pycapsule(py, Data::Bar(bar));
434                                        call_python_threadsafe(py, &call_soon, &callback, py_obj);
435                                    });
436                                }
437                                NautilusWsMessage::MarkPrice(mark_price) => {
438                                    Python::attach(|py| {
439                                        let py_obj = data_to_pycapsule(
440                                            py,
441                                            Data::MarkPriceUpdate(mark_price),
442                                        );
443                                        call_python_threadsafe(py, &call_soon, &callback, py_obj);
444                                    });
445                                }
446                                NautilusWsMessage::IndexPrice(index_price) => {
447                                    Python::attach(|py| {
448                                        let py_obj = data_to_pycapsule(
449                                            py,
450                                            Data::IndexPriceUpdate(index_price),
451                                        );
452                                        call_python_threadsafe(py, &call_soon, &callback, py_obj);
453                                    });
454                                }
455                                NautilusWsMessage::FundingRate(funding_rate) => {
456                                    Python::attach(|py| {
457                                        if let Ok(py_obj) = funding_rate.into_py_any(py) {
458                                            call_python_threadsafe(py, &call_soon, &callback, py_obj);
459                                        }
460                                    });
461                                }
462                                NautilusWsMessage::CustomData(data) => {
463                                    Python::attach(|py| match ws_data_to_pyobject(py, data) {
464                                        Ok(py_obj) => {
465                                            call_python_threadsafe(py, &call_soon, &callback, py_obj);
466                                        }
467                                        Err(e) => {
468                                            log::error!(
469                                                "Error converting CustomData to Python object: {e}"
470                                            );
471                                        }
472                                    });
473                                }
474                                NautilusWsMessage::ExecutionReports(reports) => {
475                                    Python::attach(|py| {
476                                        for report in reports {
477                                            match report {
478                                                ExecutionReport::Order(order_report) => {
479                                                    log::debug!(
480                                                        "Forwarding order status report: order_id={}, status={:?}",
481                                                        order_report.venue_order_id,
482                                                        order_report.order_status
483                                                    );
484
485                                                    match Py::new(py, order_report) {
486                                                        Ok(py_obj) => {
487                                                            call_python_threadsafe(py, &call_soon, &callback, py_obj.into_any());
488                                                        }
489                                                        Err(e) => {
490                                                            log::error!("Error converting OrderStatusReport to Python: {e}");
491                                                        }
492                                                    }
493                                                }
494                                                ExecutionReport::Fill(fill_report) => {
495                                                    log::debug!(
496                                                        "Forwarding fill report: trade_id={}, side={:?}, qty={}, price={}",
497                                                        fill_report.trade_id,
498                                                        fill_report.order_side,
499                                                        fill_report.last_qty,
500                                                        fill_report.last_px
501                                                    );
502
503                                                    match Py::new(py, fill_report) {
504                                                        Ok(py_obj) => {
505                                                            call_python_threadsafe(py, &call_soon, &callback, py_obj.into_any());
506                                                        }
507                                                        Err(e) => {
508                                                            log::error!("Error converting FillReport to Python: {e}");
509                                                        }
510                                                    }
511                                                }
512                                            }
513                                        }
514                                    });
515                                }
516                                _ => {
517                                    log::debug!("Unhandled message type: {msg:?}");
518                                }
519                            }
520                        }
521                        None => {
522                            log::debug!("WebSocket connection closed");
523                            break;
524                        }
525                    }
526                }
527            });
528
529            Ok(())
530        })
531    }
532
533    #[pyo3(name = "wait_until_active")]
534    fn py_wait_until_active<'py>(
535        &self,
536        py: Python<'py>,
537        timeout_secs: f64,
538    ) -> PyResult<Bound<'py, PyAny>> {
539        let client = self.clone();
540
541        pyo3_async_runtimes::tokio::future_into_py(py, async move {
542            let start = std::time::Instant::now();
543
544            loop {
545                if client.is_active() {
546                    return Ok(());
547                }
548
549                if start.elapsed().as_secs_f64() >= timeout_secs {
550                    return Err(to_pyruntime_err(format!(
551                        "WebSocket connection did not become active within {timeout_secs} seconds"
552                    )));
553                }
554
555                tokio::time::sleep(std::time::Duration::from_millis(100)).await;
556            }
557        })
558    }
559
560    #[pyo3(name = "close")]
561    fn py_close<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
562        let mut client = self.clone();
563
564        pyo3_async_runtimes::tokio::future_into_py(py, async move {
565            if let Err(e) = client.disconnect().await {
566                log::warn!("Error on close: {e}");
567            }
568            Ok(())
569        })
570    }
571
572    /// Subscribe to trades for an instrument.
573    #[pyo3(name = "subscribe_trades")]
574    fn py_subscribe_trades<'py>(
575        &self,
576        py: Python<'py>,
577        instrument_id: InstrumentId,
578    ) -> PyResult<Bound<'py, PyAny>> {
579        let client = self.clone();
580
581        pyo3_async_runtimes::tokio::future_into_py(py, async move {
582            client
583                .subscribe_trades(instrument_id)
584                .await
585                .map_err(to_pyruntime_err)?;
586            Ok(())
587        })
588    }
589
590    /// Unsubscribe from trades for an instrument.
591    #[pyo3(name = "unsubscribe_trades")]
592    fn py_unsubscribe_trades<'py>(
593        &self,
594        py: Python<'py>,
595        instrument_id: InstrumentId,
596    ) -> PyResult<Bound<'py, PyAny>> {
597        let client = self.clone();
598
599        pyo3_async_runtimes::tokio::future_into_py(py, async move {
600            client
601                .unsubscribe_trades(instrument_id)
602                .await
603                .map_err(to_pyruntime_err)?;
604            Ok(())
605        })
606    }
607
608    /// Subscribe to all mid prices across markets.
609    #[pyo3(name = "subscribe_all_mids")]
610    fn py_subscribe_all_mids<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
611        let client = self.clone();
612
613        pyo3_async_runtimes::tokio::future_into_py(py, async move {
614            client
615                .subscribe_all_mids()
616                .await
617                .map_err(to_pyruntime_err)?;
618            Ok(())
619        })
620    }
621
622    /// Subscribe to aggregate asset contexts across all perp dexes.
623    #[pyo3(name = "subscribe_all_dexs_asset_ctxs")]
624    fn py_subscribe_all_dexs_asset_ctxs<'py>(
625        &self,
626        py: Python<'py>,
627    ) -> PyResult<Bound<'py, PyAny>> {
628        let client = self.clone();
629
630        pyo3_async_runtimes::tokio::future_into_py(py, async move {
631            client
632                .subscribe_all_dexs_asset_ctxs()
633                .await
634                .map_err(to_pyruntime_err)?;
635            Ok(())
636        })
637    }
638
639    /// Subscribe to all mid prices across markets, optionally scoped to a specific dex.
640    #[pyo3(name = "subscribe_all_mids_with_dex")]
641    fn py_subscribe_all_mids_with_dex<'py>(
642        &self,
643        py: Python<'py>,
644        dex: Option<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                .subscribe_all_mids_with_dex(dex.as_deref())
651                .await
652                .map_err(to_pyruntime_err)?;
653            Ok(())
654        })
655    }
656
657    /// Subscribe to L2 order book for an instrument.
658    #[pyo3(name = "subscribe_book")]
659    fn py_subscribe_book<'py>(
660        &self,
661        py: Python<'py>,
662        instrument_id: InstrumentId,
663    ) -> PyResult<Bound<'py, PyAny>> {
664        let client = self.clone();
665
666        pyo3_async_runtimes::tokio::future_into_py(py, async move {
667            client
668                .subscribe_book(instrument_id)
669                .await
670                .map_err(to_pyruntime_err)?;
671            Ok(())
672        })
673    }
674
675    /// Unsubscribe from L2 order book for an instrument.
676    #[pyo3(name = "unsubscribe_book")]
677    fn py_unsubscribe_book<'py>(
678        &self,
679        py: Python<'py>,
680        instrument_id: InstrumentId,
681    ) -> PyResult<Bound<'py, PyAny>> {
682        let client = self.clone();
683
684        pyo3_async_runtimes::tokio::future_into_py(py, async move {
685            client
686                .unsubscribe_book(instrument_id)
687                .await
688                .map_err(to_pyruntime_err)?;
689            Ok(())
690        })
691    }
692
693    #[pyo3(name = "subscribe_book_deltas")]
694    fn py_subscribe_book_deltas<'py>(
695        &self,
696        py: Python<'py>,
697        instrument_id: InstrumentId,
698        _book_type: u8,
699        _depth: u64,
700    ) -> PyResult<Bound<'py, PyAny>> {
701        let client = self.clone();
702
703        pyo3_async_runtimes::tokio::future_into_py(py, async move {
704            client
705                .subscribe_book(instrument_id)
706                .await
707                .map_err(to_pyruntime_err)?;
708            Ok(())
709        })
710    }
711
712    #[pyo3(name = "unsubscribe_book_deltas")]
713    fn py_unsubscribe_book_deltas<'py>(
714        &self,
715        py: Python<'py>,
716        instrument_id: InstrumentId,
717    ) -> PyResult<Bound<'py, PyAny>> {
718        let client = self.clone();
719
720        pyo3_async_runtimes::tokio::future_into_py(py, async move {
721            client
722                .unsubscribe_book(instrument_id)
723                .await
724                .map_err(to_pyruntime_err)?;
725            Ok(())
726        })
727    }
728
729    #[pyo3(name = "subscribe_book_snapshots")]
730    fn py_subscribe_book_snapshots<'py>(
731        &self,
732        py: Python<'py>,
733        instrument_id: InstrumentId,
734        _book_type: u8,
735        _depth: u64,
736    ) -> PyResult<Bound<'py, PyAny>> {
737        let client = self.clone();
738
739        pyo3_async_runtimes::tokio::future_into_py(py, async move {
740            client
741                .subscribe_book(instrument_id)
742                .await
743                .map_err(to_pyruntime_err)?;
744            Ok(())
745        })
746    }
747
748    /// Subscribe to best bid/offer (BBO) quotes for an instrument.
749    #[pyo3(name = "subscribe_quotes")]
750    fn py_subscribe_quotes<'py>(
751        &self,
752        py: Python<'py>,
753        instrument_id: InstrumentId,
754    ) -> PyResult<Bound<'py, PyAny>> {
755        let client = self.clone();
756
757        pyo3_async_runtimes::tokio::future_into_py(py, async move {
758            client
759                .subscribe_quotes(instrument_id)
760                .await
761                .map_err(to_pyruntime_err)?;
762            Ok(())
763        })
764    }
765
766    /// Unsubscribe from quote ticks for an instrument.
767    #[pyo3(name = "unsubscribe_quotes")]
768    fn py_unsubscribe_quotes<'py>(
769        &self,
770        py: Python<'py>,
771        instrument_id: InstrumentId,
772    ) -> PyResult<Bound<'py, PyAny>> {
773        let client = self.clone();
774
775        pyo3_async_runtimes::tokio::future_into_py(py, async move {
776            client
777                .unsubscribe_quotes(instrument_id)
778                .await
779                .map_err(to_pyruntime_err)?;
780            Ok(())
781        })
782    }
783
784    /// Subscribe to candle/bar data for a specific coin and interval.
785    #[pyo3(name = "subscribe_bars")]
786    fn py_subscribe_bars<'py>(
787        &self,
788        py: Python<'py>,
789        bar_type: BarType,
790    ) -> PyResult<Bound<'py, PyAny>> {
791        let client = self.clone();
792
793        pyo3_async_runtimes::tokio::future_into_py(py, async move {
794            client
795                .subscribe_bars(bar_type)
796                .await
797                .map_err(to_pyruntime_err)?;
798            Ok(())
799        })
800    }
801
802    /// Unsubscribe from candle/bar data.
803    #[pyo3(name = "unsubscribe_bars")]
804    fn py_unsubscribe_bars<'py>(
805        &self,
806        py: Python<'py>,
807        bar_type: BarType,
808    ) -> PyResult<Bound<'py, PyAny>> {
809        let client = self.clone();
810
811        pyo3_async_runtimes::tokio::future_into_py(py, async move {
812            client
813                .unsubscribe_bars(bar_type)
814                .await
815                .map_err(to_pyruntime_err)?;
816            Ok(())
817        })
818    }
819
820    /// Unsubscribe from all mid prices across markets.
821    #[pyo3(name = "unsubscribe_all_mids")]
822    fn py_unsubscribe_all_mids<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
823        let client = self.clone();
824
825        pyo3_async_runtimes::tokio::future_into_py(py, async move {
826            client
827                .unsubscribe_all_mids()
828                .await
829                .map_err(to_pyruntime_err)?;
830            Ok(())
831        })
832    }
833
834    /// Unsubscribe from aggregate asset contexts across all perp dexes.
835    #[pyo3(name = "unsubscribe_all_dexs_asset_ctxs")]
836    fn py_unsubscribe_all_dexs_asset_ctxs<'py>(
837        &self,
838        py: Python<'py>,
839    ) -> PyResult<Bound<'py, PyAny>> {
840        let client = self.clone();
841
842        pyo3_async_runtimes::tokio::future_into_py(py, async move {
843            client
844                .unsubscribe_all_dexs_asset_ctxs()
845                .await
846                .map_err(to_pyruntime_err)?;
847            Ok(())
848        })
849    }
850
851    /// Unsubscribe from all mid prices across markets, optionally scoped to a specific dex.
852    #[pyo3(name = "unsubscribe_all_mids_with_dex")]
853    fn py_unsubscribe_all_mids_with_dex<'py>(
854        &self,
855        py: Python<'py>,
856        dex: Option<String>,
857    ) -> PyResult<Bound<'py, PyAny>> {
858        let client = self.clone();
859
860        pyo3_async_runtimes::tokio::future_into_py(py, async move {
861            client
862                .unsubscribe_all_mids_with_dex(dex.as_deref())
863                .await
864                .map_err(to_pyruntime_err)?;
865            Ok(())
866        })
867    }
868
869    /// Subscribe to order updates for a specific user address.
870    #[pyo3(name = "subscribe_order_updates")]
871    fn py_subscribe_order_updates<'py>(
872        &self,
873        py: Python<'py>,
874        user: String,
875    ) -> PyResult<Bound<'py, PyAny>> {
876        let client = self.clone();
877
878        pyo3_async_runtimes::tokio::future_into_py(py, async move {
879            client
880                .subscribe_order_updates(&user)
881                .await
882                .map_err(to_pyruntime_err)?;
883            Ok(())
884        })
885    }
886
887    /// Subscribe to user events (fills, funding, liquidations) for a specific user address.
888    #[pyo3(name = "subscribe_user_events")]
889    fn py_subscribe_user_events<'py>(
890        &self,
891        py: Python<'py>,
892        user: String,
893    ) -> PyResult<Bound<'py, PyAny>> {
894        let client = self.clone();
895
896        pyo3_async_runtimes::tokio::future_into_py(py, async move {
897            client
898                .subscribe_user_events(&user)
899                .await
900                .map_err(to_pyruntime_err)?;
901            Ok(())
902        })
903    }
904
905    /// Subscribe to user fills for a specific user address.
906    ///
907    /// Note: This channel is redundant with `userEvents` which already includes fills.
908    /// Prefer using `subscribe_user_events` or `subscribe_all_user_channels` instead.
909    #[pyo3(name = "subscribe_user_fills")]
910    fn py_subscribe_user_fills<'py>(
911        &self,
912        py: Python<'py>,
913        user: String,
914    ) -> PyResult<Bound<'py, PyAny>> {
915        let client = self.clone();
916
917        pyo3_async_runtimes::tokio::future_into_py(py, async move {
918            client
919                .subscribe_user_fills(&user)
920                .await
921                .map_err(to_pyruntime_err)?;
922            Ok(())
923        })
924    }
925
926    /// Subscribe to mark price updates for an instrument.
927    #[pyo3(name = "subscribe_mark_prices")]
928    fn py_subscribe_mark_prices<'py>(
929        &self,
930        py: Python<'py>,
931        instrument_id: InstrumentId,
932    ) -> PyResult<Bound<'py, PyAny>> {
933        let client = self.clone();
934
935        pyo3_async_runtimes::tokio::future_into_py(py, async move {
936            client
937                .subscribe_mark_prices(instrument_id)
938                .await
939                .map_err(to_pyruntime_err)?;
940            Ok(())
941        })
942    }
943
944    /// Unsubscribe from mark price updates for an instrument.
945    #[pyo3(name = "unsubscribe_mark_prices")]
946    fn py_unsubscribe_mark_prices<'py>(
947        &self,
948        py: Python<'py>,
949        instrument_id: InstrumentId,
950    ) -> PyResult<Bound<'py, PyAny>> {
951        let client = self.clone();
952
953        pyo3_async_runtimes::tokio::future_into_py(py, async move {
954            client
955                .unsubscribe_mark_prices(instrument_id)
956                .await
957                .map_err(to_pyruntime_err)?;
958            Ok(())
959        })
960    }
961
962    /// Subscribe to index/oracle price updates for an instrument.
963    #[pyo3(name = "subscribe_index_prices")]
964    fn py_subscribe_index_prices<'py>(
965        &self,
966        py: Python<'py>,
967        instrument_id: InstrumentId,
968    ) -> PyResult<Bound<'py, PyAny>> {
969        let client = self.clone();
970
971        pyo3_async_runtimes::tokio::future_into_py(py, async move {
972            client
973                .subscribe_index_prices(instrument_id)
974                .await
975                .map_err(to_pyruntime_err)?;
976            Ok(())
977        })
978    }
979
980    /// Unsubscribe from index/oracle price updates for an instrument.
981    #[pyo3(name = "unsubscribe_index_prices")]
982    fn py_unsubscribe_index_prices<'py>(
983        &self,
984        py: Python<'py>,
985        instrument_id: InstrumentId,
986    ) -> PyResult<Bound<'py, PyAny>> {
987        let client = self.clone();
988
989        pyo3_async_runtimes::tokio::future_into_py(py, async move {
990            client
991                .unsubscribe_index_prices(instrument_id)
992                .await
993                .map_err(to_pyruntime_err)?;
994            Ok(())
995        })
996    }
997
998    /// Subscribe to funding rate updates for an instrument.
999    #[pyo3(name = "subscribe_funding_rates")]
1000    fn py_subscribe_funding_rates<'py>(
1001        &self,
1002        py: Python<'py>,
1003        instrument_id: InstrumentId,
1004    ) -> PyResult<Bound<'py, PyAny>> {
1005        let client = self.clone();
1006
1007        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1008            client
1009                .subscribe_funding_rates(instrument_id)
1010                .await
1011                .map_err(to_pyruntime_err)?;
1012            Ok(())
1013        })
1014    }
1015
1016    /// Subscribe to open interest updates for an instrument.
1017    #[pyo3(name = "subscribe_open_interest")]
1018    fn py_subscribe_open_interest<'py>(
1019        &self,
1020        py: Python<'py>,
1021        instrument_id: InstrumentId,
1022    ) -> PyResult<Bound<'py, PyAny>> {
1023        let client = self.clone();
1024
1025        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1026            client
1027                .subscribe_open_interest(instrument_id)
1028                .await
1029                .map_err(to_pyruntime_err)?;
1030            Ok(())
1031        })
1032    }
1033
1034    /// Unsubscribe from funding rate updates for an instrument.
1035    #[pyo3(name = "unsubscribe_funding_rates")]
1036    fn py_unsubscribe_funding_rates<'py>(
1037        &self,
1038        py: Python<'py>,
1039        instrument_id: InstrumentId,
1040    ) -> PyResult<Bound<'py, PyAny>> {
1041        let client = self.clone();
1042
1043        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1044            client
1045                .unsubscribe_funding_rates(instrument_id)
1046                .await
1047                .map_err(to_pyruntime_err)?;
1048            Ok(())
1049        })
1050    }
1051
1052    /// Unsubscribe from open interest updates for an instrument.
1053    #[pyo3(name = "unsubscribe_open_interest")]
1054    fn py_unsubscribe_open_interest<'py>(
1055        &self,
1056        py: Python<'py>,
1057        instrument_id: InstrumentId,
1058    ) -> PyResult<Bound<'py, PyAny>> {
1059        let client = self.clone();
1060
1061        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1062            client
1063                .unsubscribe_open_interest(instrument_id)
1064                .await
1065                .map_err(to_pyruntime_err)?;
1066            Ok(())
1067        })
1068    }
1069}
nautilus_hyperliquid/python/websocket.rs

nautilus_hyperliquid/python/
websocket.rs
