nautilus_dydx/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 dYdX WebSocket client.
17
18use std::{
19    str::FromStr,
20    sync::atomic::Ordering,
21    time::{Duration, Instant},
22};
23
24use ahash::AHashMap;
25use dashmap::DashMap;
26use nautilus_common::live::get_runtime;
27use nautilus_core::{
28    UUID4,
29    python::{call_python_threadsafe, to_pyvalue_err},
30    time::get_atomic_clock_realtime,
31};
32use nautilus_model::{
33    data::{
34        Bar, BarType, Data, FundingRateUpdate, IndexPriceUpdate, InstrumentStatus, MarkPriceUpdate,
35        OrderBookDeltas, OrderBookDeltas_API,
36    },
37    enums::{AccountType, MarketStatusAction, OrderSide, OrderStatus, OrderType},
38    events::{AccountState, OrderAccepted, OrderCanceled},
39    identifiers::{
40        AccountId, ClientOrderId, InstrumentId, StrategyId, Symbol, TraderId, VenueOrderId,
41    },
42    instruments::{Instrument, InstrumentAny},
43    python::{data::data_to_pycapsule, instruments::pyobject_to_instrument_any},
44    types::{AccountBalance, Currency, Money},
45};
46use nautilus_network::mode::ConnectionMode;
47use pyo3::{IntoPyObjectExt, prelude::*, types::PyDict};
48use rust_decimal::Decimal;
49use ustr::Ustr;
50
51use crate::{
52    common::{
53        consts::DYDX_VENUE,
54        credential::DydxCredential,
55        enums::{DydxCandleResolution, DydxMarketStatus},
56        parse::{extract_raw_symbol, parse_price},
57    },
58    execution::types::OrderContext,
59    http::{client::DydxHttpClient, parse::parse_account_state},
60    python::encoder::PyDydxClientOrderIdEncoder,
61    websocket::{
62        DydxWsDispatchState, OrderIdentity,
63        client::{DydxWebSocketClient, candle_ids_from_topics},
64        enums::DydxWsOutputMessage,
65        fill_report_to_order_filled, parse as ws_parse,
66        parse::{parse_ws_fill_report, parse_ws_order_report, parse_ws_position_report},
67    },
68};
69
70#[pymethods]
71#[pyo3_stub_gen::derive::gen_stub_pymethods]
72impl DydxWebSocketClient {
73    /// Creates a new public WebSocket client for market data.
74    ///
75    /// This creates a new independent instrument cache. To share a cache with
76    /// the HTTP client, use `Self.new_public_with_cache` instead.
77    #[staticmethod]
78    #[pyo3(name = "new_public")]
79    #[pyo3(signature = (url, heartbeat=None, proxy_url=None))]
80    fn py_new_public(url: String, heartbeat: Option<u64>, proxy_url: Option<String>) -> Self {
81        Self::new_public(url, heartbeat, proxy_url)
82    }
83
84    /// Creates a new private WebSocket client for account updates.
85    ///
86    /// This creates a new independent instrument cache. To share a cache with
87    /// the HTTP client, use `Self.new_private_with_cache` instead.
88    #[staticmethod]
89    #[pyo3(name = "new_private")]
90    #[pyo3(signature = (url, private_key, authenticator_ids, account_id, heartbeat=None, proxy_url=None))]
91    fn py_new_private(
92        url: String,
93        private_key: &str,
94        authenticator_ids: Vec<u64>,
95        account_id: AccountId,
96        heartbeat: Option<u64>,
97        proxy_url: Option<String>,
98    ) -> PyResult<Self> {
99        let credential = DydxCredential::from_private_key(private_key, authenticator_ids)
100            .map_err(to_pyvalue_err)?;
101        Ok(Self::new_private(
102            url, credential, account_id, heartbeat, proxy_url,
103        ))
104    }
105
106    /// Returns `true` when any connection in the pool is connected.
107    #[pyo3(name = "is_connected")]
108    fn py_is_connected(&self) -> bool {
109        self.is_connected()
110    }
111
112    /// Sets the account ID for account message parsing.
113    #[pyo3(name = "set_account_id")]
114    fn py_set_account_id(&mut self, account_id: AccountId) {
115        self.set_account_id(account_id);
116    }
117
118    /// Sets whether bar timestamps use the close time.
119    #[pyo3(name = "set_bars_timestamp_on_close")]
120    fn py_set_bars_timestamp_on_close(&self, value: bool) {
121        self.set_bars_timestamp_on_close(value);
122    }
123
124    /// Shares the HTTP client's instrument cache with this WebSocket client.
125    ///
126    /// The HTTP client's cache includes CLOB pair ID and market ticker indices
127    /// needed for parsing SubaccountsChannelData into typed execution events.
128    /// Must be called before `connect()`.
129    #[pyo3(name = "share_instrument_cache")]
130    fn py_share_instrument_cache(&mut self, http_client: &DydxHttpClient) {
131        self.set_instrument_cache(http_client.instrument_cache().clone());
132    }
133
134    #[pyo3(name = "register_order_identity")]
135    fn py_register_order_identity(
136        &self,
137        client_order_id: ClientOrderId,
138        instrument_id: InstrumentId,
139        strategy_id: StrategyId,
140        order_side: OrderSide,
141        order_type: OrderType,
142    ) {
143        self.ws_dispatch_state().order_identities.insert(
144            client_order_id,
145            OrderIdentity {
146                instrument_id,
147                strategy_id,
148                order_side,
149                order_type,
150            },
151        );
152    }
153
154    #[pyo3(name = "remove_order_identity")]
155    fn py_remove_order_identity(&self, client_order_id: ClientOrderId) {
156        self.ws_dispatch_state()
157            .order_identities
158            .remove(&client_order_id);
159    }
160
161    /// Returns the account ID if set.
162    #[pyo3(name = "account_id")]
163    fn py_account_id(&self) -> Option<AccountId> {
164        self.account_id()
165    }
166
167    /// Returns a reference to the shared client order ID encoder.
168    #[pyo3(name = "encoder")]
169    fn py_encoder(&self) -> PyDydxClientOrderIdEncoder {
170        PyDydxClientOrderIdEncoder::from_arc(self.encoder().clone())
171    }
172
173    /// Returns the URL of this WebSocket client.
174    #[getter]
175    fn py_url(&self) -> String {
176        self.url().to_string()
177    }
178
179    /// Connects the websocket client and opens the primary pool slot.
180    ///
181    /// Additional slots are spawned lazily by `subscribe_*` methods once the
182    /// per-channel limit is reached on every existing slot.
183    #[pyo3(name = "connect")]
184    #[pyo3(signature = (loop_, instruments, callback, trader_id=None))]
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        trader_id: Option<TraderId>,
193    ) -> PyResult<Bound<'py, PyAny>> {
194        let call_soon = loop_.getattr(py, "call_soon_threadsafe")?;
195
196        let mut instruments_any = Vec::new();
197
198        for inst in instruments {
199            let inst_any = pyobject_to_instrument_any(py, inst)?;
200            instruments_any.push(inst_any);
201        }
202
203        self.cache_instruments(instruments_any);
204
205        let mut client = self.clone();
206        let bar_types = self.bar_types().clone();
207        let dispatch_state = self.ws_dispatch_state().clone();
208        let trader_id = trader_id.unwrap_or(TraderId::from("TRADER-000"));
209
210        pyo3_async_runtimes::tokio::future_into_py(py, async move {
211            client.connect().await.map_err(to_pyvalue_err)?;
212
213            if let Some(mut rx) = client.take_receiver() {
214                get_runtime().spawn(async move {
215                    let _client = client; // Keep client alive in spawned task
216                    let clock = get_atomic_clock_realtime();
217                    let order_contexts: DashMap<u32, OrderContext> = DashMap::new();
218                    let order_id_map: DashMap<String, (u32, u32)> = DashMap::new();
219                    let bars_timestamp_on_close = _client.bars_timestamp_on_close();
220                    let mut pending_bars: AHashMap<String, Bar> = AHashMap::new();
221                    let mut seen_tickers: ahash::AHashSet<Ustr> = ahash::AHashSet::new();
222
223                    while let Some(msg) = rx.recv().await {
224                        let ts_init = clock.get_time_ns();
225
226                        match msg {
227                            DydxWsOutputMessage::Trades { id, contents } => {
228                                let Some(instrument) = _client.instrument_cache().get_by_market(&id) else {
229                                    log::warn!("No instrument cached for market {id}");
230                                    continue;
231                                };
232                                let instrument_id = instrument.id();
233
234                                match ws_parse::parse_trade_ticks(instrument_id, &instrument, &contents, ts_init) {
235                                    Ok(items) => {
236                                        Python::attach(|py| {
237                                            for data in items {
238                                                let py_obj = data_to_pycapsule(py, data);
239                                                call_python_threadsafe(py, &call_soon, &callback, py_obj);
240                                            }
241                                        });
242                                    }
243                                    Err(e) => log::error!("Failed to parse trade ticks for {id}: {e}"),
244                                }
245                            }
246                            DydxWsOutputMessage::OrderbookSnapshot { id, contents } => {
247                                let Some(instrument) = _client.instrument_cache().get_by_market(&id) else {
248                                    log::warn!("No instrument cached for market {id}");
249                                    continue;
250                                };
251                                let instrument_id = instrument.id();
252                                let price_precision = instrument.price_precision();
253                                let size_precision = instrument.size_precision();
254
255                                match ws_parse::parse_orderbook_snapshot(
256                                    &instrument_id,
257                                    &contents,
258                                    price_precision,
259                                    size_precision,
260                                    ts_init,
261                                ) {
262                                    Ok(deltas) => {
263                                        Python::attach(|py| {
264                                            let data = Data::Deltas(OrderBookDeltas_API::new(deltas));
265                                            let py_obj = data_to_pycapsule(py, data);
266                                            call_python_threadsafe(py, &call_soon, &callback, py_obj);
267                                        });
268                                    }
269                                    Err(e) => log::error!("Failed to parse orderbook snapshot for {id}: {e}"),
270                                }
271                            }
272                            DydxWsOutputMessage::OrderbookUpdate { id, contents } => {
273                                let Some(instrument) = _client.instrument_cache().get_by_market(&id) else {
274                                    log::warn!("No instrument cached for market {id}");
275                                    continue;
276                                };
277                                let instrument_id = instrument.id();
278                                let price_precision = instrument.price_precision();
279                                let size_precision = instrument.size_precision();
280
281                                match ws_parse::parse_orderbook_deltas(
282                                    &instrument_id,
283                                    &contents,
284                                    price_precision,
285                                    size_precision,
286                                    ts_init,
287                                ) {
288                                    Ok(deltas) => {
289                                        Python::attach(|py| {
290                                            let data = Data::Deltas(OrderBookDeltas_API::new(deltas));
291                                            let py_obj = data_to_pycapsule(py, data);
292                                            call_python_threadsafe(py, &call_soon, &callback, py_obj);
293                                        });
294                                    }
295                                    Err(e) => log::error!("Failed to parse orderbook deltas for {id}: {e}"),
296                                }
297                            }
298                            DydxWsOutputMessage::OrderbookBatch { id, updates } => {
299                                let Some(instrument) = _client.instrument_cache().get_by_market(&id) else {
300                                    log::warn!("No instrument cached for market {id}");
301                                    continue;
302                                };
303                                let instrument_id = instrument.id();
304                                let price_precision = instrument.price_precision();
305                                let size_precision = instrument.size_precision();
306
307                                let mut all_deltas = Vec::new();
308                                let last_idx = updates.len().saturating_sub(1);
309                                let mut parse_ok = true;
310
311                                for (idx, update) in updates.iter().enumerate() {
312                                    if idx < last_idx {
313                                        match ws_parse::parse_orderbook_deltas_with_flag(
314                                            &instrument_id,
315                                            update,
316                                            price_precision,
317                                            size_precision,
318                                            ts_init,
319                                            false,
320                                        ) {
321                                            Ok(deltas) => all_deltas.extend(deltas),
322                                            Err(e) => {
323                                                log::error!("Failed to parse batch orderbook deltas for {id}: {e}");
324                                                parse_ok = false;
325                                                break;
326                                            }
327                                        }
328                                    } else {
329                                        match ws_parse::parse_orderbook_deltas(
330                                            &instrument_id,
331                                            update,
332                                            price_precision,
333                                            size_precision,
334                                            ts_init,
335                                        ) {
336                                            Ok(last_deltas) => all_deltas.extend(last_deltas.deltas),
337                                            Err(e) => {
338                                                log::error!("Failed to parse batch orderbook deltas for {id}: {e}");
339                                                parse_ok = false;
340                                                break;
341                                            }
342                                        }
343                                    }
344                                }
345
346                                if parse_ok && !all_deltas.is_empty() {
347                                    let combined = OrderBookDeltas::new(instrument_id, all_deltas);
348                                    Python::attach(|py| {
349                                        let data = Data::Deltas(OrderBookDeltas_API::new(combined));
350                                        let py_obj = data_to_pycapsule(py, data);
351                                        call_python_threadsafe(py, &call_soon, &callback, py_obj);
352                                    });
353                                }
354                            }
355                            DydxWsOutputMessage::Candles { id, contents } => {
356                                let ticker = id.split('/').next().unwrap_or(&id);
357
358                                let Some(bar_type) = bar_types.get(&id).map(|r| *r) else {
359                                    log::debug!("No bar type registered for candle topic {id}");
360                                    continue;
361                                };
362
363                                let Some(instrument) = _client.instrument_cache().get_by_market(ticker) else {
364                                    log::warn!("No instrument cached for market {ticker}");
365                                    continue;
366                                };
367
368                                match ws_parse::parse_candle_bar(
369                                    bar_type,
370                                    &instrument,
371                                    &contents,
372                                    bars_timestamp_on_close,
373                                    ts_init,
374                                ) {
375                                    Ok(bar) => {
376                                        if let Some(prev_bar) = pending_bars.get(&id) {
377                                            if bar.ts_event == prev_bar.ts_event {
378                                                pending_bars.insert(id, bar);
379                                            } else {
380                                                let emit_bar = *prev_bar;
381                                                pending_bars.insert(id.clone(), bar);
382                                                Python::attach(|py| {
383                                                    let py_obj = data_to_pycapsule(py, Data::Bar(emit_bar));
384                                                    call_python_threadsafe(py, &call_soon, &callback, py_obj);
385                                                });
386                                            }
387                                        } else {
388                                            pending_bars.insert(id, bar);
389                                        }
390                                    }
391                                    Err(e) => log::error!("Failed to parse candle bar for {id}: {e}"),
392                                }
393                            }
394                            DydxWsOutputMessage::Markets(contents) => {
395                                if let Some(ref oracle_prices) = contents.oracle_prices {
396                                    for (ticker, oracle_data) in oracle_prices {
397                                        let Some(instrument) = _client.instrument_cache().get_by_market(ticker) else {
398                                            continue;
399                                        };
400                                        let instrument_id = instrument.id();
401
402                                        let Ok(price) = parse_price(&oracle_data.oracle_price, "oracle_price") else {
403                                            log::warn!("Failed to parse oracle price for {ticker}");
404                                            continue;
405                                        };
406
407                                        let mark_price = MarkPriceUpdate::new(
408                                            instrument_id,
409                                            price,
410                                            ts_init,
411                                            ts_init,
412                                        );
413                                        Python::attach(|py| {
414                                            match mark_price.into_py_any(py) {
415                                                Ok(py_obj) => {
416                                                    call_python_threadsafe(py, &call_soon, &callback, py_obj);
417                                                }
418                                                Err(e) => log::error!("Failed to convert MarkPriceUpdate to Python: {e}"),
419                                            }
420                                        });
421
422                                        let index_price = IndexPriceUpdate::new(
423                                            instrument_id,
424                                            price,
425                                            ts_init,
426                                            ts_init,
427                                        );
428                                        Python::attach(|py| {
429                                            match index_price.into_py_any(py) {
430                                                Ok(py_obj) => {
431                                                    call_python_threadsafe(py, &call_soon, &callback, py_obj);
432                                                }
433                                                Err(e) => log::error!("Failed to convert IndexPriceUpdate to Python: {e}"),
434                                            }
435                                        });
436                                    }
437                                }
438
439                                handle_markets_trading_data(
440                                    contents.trading.as_ref(),
441                                    _client.instrument_cache(),
442                                    &mut seen_tickers,
443                                    &call_soon,
444                                    &callback,
445                                    ts_init,
446                                );
447                                handle_markets_trading_data(
448                                    contents.markets.as_ref(),
449                                    _client.instrument_cache(),
450                                    &mut seen_tickers,
451                                    &call_soon,
452                                    &callback,
453                                    ts_init,
454                                );
455
456                                // Parse oracle prices from initial snapshot markets entries
457                                if let Some(ref markets_map) = contents.markets {
458                                    for (ticker, update) in markets_map {
459                                        if let Some(ref oracle_price_str) = update.oracle_price {
460                                            let Some(instrument) = _client.instrument_cache().get_by_market(ticker) else {
461                                                continue;
462                                            };
463                                            let instrument_id = instrument.id();
464                                            let Ok(price) = parse_price(oracle_price_str, "oracle_price") else {
465                                                log::warn!("Failed to parse oracle price for {ticker}");
466                                                continue;
467                                            };
468
469                                            let mark_price = MarkPriceUpdate::new(
470                                                instrument_id,
471                                                price,
472                                                ts_init,
473                                                ts_init,
474                                            );
475                                            Python::attach(|py| {
476                                                match mark_price.into_py_any(py) {
477                                                    Ok(py_obj) => {
478                                                        call_python_threadsafe(py, &call_soon, &callback, py_obj);
479                                                    }
480                                                    Err(e) => log::error!("Failed to convert MarkPriceUpdate to Python: {e}"),
481                                                }
482                                            });
483
484                                            let index_price = IndexPriceUpdate::new(
485                                                instrument_id,
486                                                price,
487                                                ts_init,
488                                                ts_init,
489                                            );
490                                            Python::attach(|py| {
491                                                match index_price.into_py_any(py) {
492                                                    Ok(py_obj) => {
493                                                        call_python_threadsafe(py, &call_soon, &callback, py_obj);
494                                                    }
495                                                    Err(e) => log::error!("Failed to convert IndexPriceUpdate to Python: {e}"),
496                                                }
497                                            });
498                                        }
499                                    }
500                                }
501                            }
502                            DydxWsOutputMessage::SubaccountSubscribed(data) => {
503                                let Some(account_id) = _client.account_id() else {
504                                    log::warn!("Cannot parse subaccount subscription: account_id not set");
505                                    continue;
506                                };
507
508                                let instrument_cache = _client.instrument_cache();
509
510                                let inst_map = instrument_cache.to_instrument_id_map();
511                                let oracle_map = instrument_cache.to_oracle_prices_map();
512
513                                if let Some(ref subaccount) = data.contents.subaccount {
514                                    match parse_account_state(
515                                        subaccount,
516                                        account_id,
517                                        &inst_map,
518                                        &oracle_map,
519                                        ts_init,
520                                        ts_init,
521                                    ) {
522                                        Ok(account_state) => {
523                                            Python::attach(|py| {
524                                                match account_state.into_py_any(py) {
525                                                    Ok(py_obj) => {
526                                                        call_python_threadsafe(py, &call_soon, &callback, py_obj);
527                                                    }
528                                                    Err(e) => log::error!("Failed to convert AccountState to Python: {e}"),
529                                                }
530                                            });
531                                        }
532                                    Err(e) => log::error!("Failed to parse account state: {e}"),
533                                }
534
535                                if let Some(ref positions) = subaccount.open_perpetual_positions {
536                                    for (market, ws_position) in positions {
537                                        match parse_ws_position_report(
538                                            ws_position,
539                                            instrument_cache,
540                                            account_id,
541                                            ts_init,
542                                        ) {
543                                            Ok(report) => {
544                                                Python::attach(|py| {
545                                                    match pyo3::Py::new(py, report) {
546                                                        Ok(py_obj) => {
547                                                            call_python_threadsafe(py, &call_soon, &callback, py_obj.into_any());
548                                                        }
549                                                        Err(e) => log::error!("Failed to convert PositionStatusReport to Python: {e}"),
550                                                    }
551                                                });
552                                            }
553                                            Err(e) => log::error!("Failed to parse position for {market}: {e}"),
554                                        }
555                                    }
556                                }
557                                } else {
558                                    log::warn!("Subaccount subscription without initial state (new/empty subaccount)");
559
560                                    // Emit zero-balance account state so account gets registered
561                                    let currency = Currency::get_or_create_crypto_with_context("USDC", None);
562                                    let zero = Money::zero(currency);
563                                    let balance = AccountBalance::new_checked(zero, zero, zero)
564                                        .expect("zero balance should always be valid");
565                                    let account_state = AccountState::new(
566                                        account_id,
567                                        AccountType::Margin,
568                                        vec![balance],
569                                        vec![],
570                                        true,
571                                        UUID4::new(),
572                                        ts_init,
573                                        ts_init,
574                                        None,
575                                    );
576                                    Python::attach(|py| {
577                                        match account_state.into_py_any(py) {
578                                            Ok(py_obj) => {
579                                                call_python_threadsafe(py, &call_soon, &callback, py_obj);
580                                            }
581                                            Err(e) => log::error!("Failed to convert AccountState to Python: {e}"),
582                                        }
583                                    });
584                                }
585                            }
586                            DydxWsOutputMessage::SubaccountsChannelData(data) => {
587                                let Some(account_id) = _client.account_id() else {
588                                    log::warn!("Cannot parse SubaccountsChannelData: account_id not set");
589                                    continue;
590                                };
591
592                                let instrument_cache = _client.instrument_cache();
593                                let encoder = _client.encoder();
594
595                                let mut terminal_orders: Vec<(u32, u32, String)> = Vec::new();
596                                let mut cum_fill_totals: AHashMap<VenueOrderId, (Decimal, Decimal)> = AHashMap::new();
597
598                                // Phase 1: Parse orders and build order_id_map (needed for fill correlation)
599                                let mut pending_order_reports = Vec::new();
600
601                                if let Some(ref orders) = data.contents.orders {
602                                    for ws_order in orders {
603                                        if let Ok(client_id_u32) = ws_order.client_id.parse::<u32>() {
604                                            let client_meta = ws_order.client_metadata
605                                                .as_ref()
606                                                .and_then(|s| s.parse::<u32>().ok())
607                                                .unwrap_or(crate::grpc::DEFAULT_RUST_CLIENT_METADATA);
608                                            order_id_map.insert(ws_order.id.clone(), (client_id_u32, client_meta));
609                                        }
610
611                                        match parse_ws_order_report(
612                                            ws_order,
613                                            instrument_cache,
614                                            &order_contexts,
615                                            encoder,
616                                            account_id,
617                                            ts_init,
618                                        ) {
619                                            Ok(report) => {
620                                                if !report.order_status.is_open()
621                                                    && let Ok(cid) = ws_order.client_id.parse::<u32>()
622                                                {
623                                                    let meta = ws_order.client_metadata
624                                                        .as_ref()
625                                                        .and_then(|s| s.parse::<u32>().ok())
626                                                        .unwrap_or(crate::grpc::DEFAULT_RUST_CLIENT_METADATA);
627                                                    terminal_orders.push((cid, meta, ws_order.id.clone()));
628                                                }
629                                                pending_order_reports.push(report);
630                                            }
631                                            Err(e) => log::error!("Failed to parse WS order: {e}"),
632                                        }
633                                    }
634                                }
635
636                                // Phase 2: Process fills (tracked get OrderFilled, untracked get FillReport)
637                                if let Some(ref fills) = data.contents.fills {
638                                    for ws_fill in fills {
639                                        match parse_ws_fill_report(
640                                            ws_fill,
641                                            instrument_cache,
642                                            &order_id_map,
643                                            &order_contexts,
644                                            encoder,
645                                            account_id,
646                                            ts_init,
647                                        ) {
648                                            Ok(report) => {
649                                                let identity = report.client_order_id.and_then(|cid| {
650                                                    dispatch_state.order_identities.get(&cid).map(|r| (cid, r.clone()))
651                                                });
652
653                                                if let Some((cid, ident)) = identity {
654                                                    ensure_accepted_to_python(
655                                                        cid,
656                                                        account_id,
657                                                        report.venue_order_id,
658                                                        &ident,
659                                                        &dispatch_state,
660                                                        trader_id,
661                                                        ts_init,
662                                                        &call_soon,
663                                                        &callback,
664                                                    );
665                                                    dispatch_state.insert_filled(cid);
666                                                    let quote_currency = instrument_cache
667                                                        .get(&report.instrument_id)
668                                                        .map_or_else(Currency::USD, |i: InstrumentAny| i.quote_currency());
669                                                    let filled = fill_report_to_order_filled(
670                                                        &report, trader_id, &ident, quote_currency,
671                                                    );
672                                                    send_to_python(filled, &call_soon, &callback);
673                                                } else {
674                                                    let entry = cum_fill_totals
675                                                        .entry(report.venue_order_id)
676                                                        .or_default();
677                                                    let qty = report.last_qty.as_decimal();
678                                                    entry.0 += report.last_px.as_decimal() * qty;
679                                                    entry.1 += qty;
680                                                    send_to_python(report, &call_soon, &callback);
681                                                }
682                                            }
683                                            Err(e) => log::error!("Failed to parse WS fill: {e}"),
684                                        }
685                                    }
686                                }
687
688                                // Phase 3: Process order status updates
689                                for report in &mut pending_order_reports {
690                                    if let Some((notional, total_qty)) =
691                                        cum_fill_totals.get(&report.venue_order_id)
692                                        && !total_qty.is_zero()
693                                    {
694                                        report.avg_px = Some(notional / total_qty);
695                                    }
696                                }
697
698                                for report in pending_order_reports {
699                                    let identity = report.client_order_id.and_then(|cid| {
700                                        dispatch_state.order_identities.get(&cid).map(|r| (cid, r.clone()))
701                                    });
702
703                                    if let Some((cid, ident)) = identity {
704                                        match report.order_status {
705                                            OrderStatus::Accepted => {
706                                                if dispatch_state.emitted_accepted.contains(&cid)
707                                                    || dispatch_state.filled_orders.contains(&cid)
708                                                {
709                                                    log::debug!("Skipping duplicate Accepted for {cid}");
710                                                    continue;
711                                                }
712                                                dispatch_state.insert_accepted(cid);
713                                                let accepted = OrderAccepted::new(
714                                                    trader_id,
715                                                    ident.strategy_id,
716                                                    ident.instrument_id,
717                                                    cid,
718                                                    report.venue_order_id,
719                                                    account_id,
720                                                    UUID4::new(),
721                                                    report.ts_last,
722                                                    ts_init,
723                                                    false,
724                                                );
725                                                send_to_python(accepted, &call_soon, &callback);
726                                            }
727                                            OrderStatus::Canceled => {
728                                                ensure_accepted_to_python(
729                                                    cid,
730                                                    account_id,
731                                                    report.venue_order_id,
732                                                    &ident,
733                                                    &dispatch_state,
734                                                    trader_id,
735                                                    ts_init,
736                                                    &call_soon,
737                                                    &callback,
738                                                );
739                                                let canceled = OrderCanceled::new(
740                                                    trader_id,
741                                                    ident.strategy_id,
742                                                    ident.instrument_id,
743                                                    cid,
744                                                    UUID4::new(),
745                                                    report.ts_last,
746                                                    ts_init,
747                                                    false,
748                                                    Some(report.venue_order_id),
749                                                    Some(account_id),
750                                                );
751                                                send_to_python(canceled, &call_soon, &callback);
752                                                dispatch_state.cleanup_terminal(&cid);
753                                            }
754                                            OrderStatus::Filled => {
755                                                dispatch_state.cleanup_terminal(&cid);
756                                            }
757                                            _ => {
758                                                send_to_python(report, &call_soon, &callback);
759                                            }
760                                        }
761                                    } else {
762                                        send_to_python(report, &call_soon, &callback);
763                                    }
764                                }
765
766                                // Deferred cleanup after fills are correlated
767                                for (client_id, client_metadata, order_id) in terminal_orders {
768                                    order_contexts.remove(&client_id);
769                                    encoder.remove(client_id, client_metadata);
770                                    order_id_map.remove(&order_id);
771                                }
772                            }
773                            DydxWsOutputMessage::BlockHeight { height, time } => {
774                                Python::attach(|py| {
775                                    let dict = PyDict::new(py);
776                                    let _ = dict.set_item("type", "block_height");
777                                    let _ = dict.set_item("height", height);
778                                    let _ = dict.set_item("time", time.to_rfc3339());
779                                    if let Ok(py_obj) = dict.into_py_any(py) {
780                                        call_python_threadsafe(py, &call_soon, &callback, py_obj);
781                                    }
782                                });
783                            }
784                            DydxWsOutputMessage::Error(err) => {
785                                log::warn!("dYdX WebSocket error: {err}");
786                            }
787                            DydxWsOutputMessage::Reconnected { topics } => {
788                                log::info!("dYdX WebSocket reconnected");
789                                let reconnected_candles = candle_ids_from_topics(&topics);
790                                pending_bars.retain(|id, _| !reconnected_candles.contains(id));
791                            }
792                        }
793                    }
794                });
795            }
796
797            Ok(())
798        })
799    }
800
801    /// Disconnects all websocket connections in the pool.
802    ///
803    /// # Errors
804    ///
805    /// Returns an error if the underlying clients cannot be accessed.
806    #[pyo3(name = "disconnect")]
807    fn py_disconnect<'py>(&mut self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
808        let mut client = self.clone();
809        pyo3_async_runtimes::tokio::future_into_py(py, async move {
810            client.disconnect().await.map_err(to_pyvalue_err)?;
811            Ok(())
812        })
813    }
814
815    #[pyo3(name = "wait_until_active")]
816    fn py_wait_until_active<'py>(
817        &self,
818        py: Python<'py>,
819        timeout_secs: f64,
820    ) -> PyResult<Bound<'py, PyAny>> {
821        let connection_mode = self.connection_mode_atomic();
822
823        pyo3_async_runtimes::tokio::future_into_py(py, async move {
824            let timeout = Duration::from_secs_f64(timeout_secs);
825            let start = Instant::now();
826
827            loop {
828                let mode = connection_mode.load();
829                let mode_u8 = mode.load(Ordering::Relaxed);
830                let is_connected = matches!(
831                    mode_u8,
832                    x if x == ConnectionMode::Active as u8 || x == ConnectionMode::Reconnect as u8
833                );
834
835                if is_connected {
836                    break;
837                }
838
839                if start.elapsed() > timeout {
840                    return Err(to_pyvalue_err(std::io::Error::new(
841                        std::io::ErrorKind::TimedOut,
842                        format!("Client did not become active within {timeout_secs}s"),
843                    )));
844                }
845                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
846            }
847
848            Ok(())
849        })
850    }
851
852    /// Caches a single instrument.
853    ///
854    /// Any existing instrument with the same ID will be replaced.
855    #[pyo3(name = "cache_instrument")]
856    fn py_cache_instrument(&self, instrument: Py<PyAny>, py: Python<'_>) -> PyResult<()> {
857        let inst_any = pyobject_to_instrument_any(py, instrument)?;
858        self.cache_instrument(inst_any);
859        Ok(())
860    }
861
862    /// Caches multiple instruments.
863    ///
864    /// Any existing instruments with the same IDs will be replaced.
865    #[pyo3(name = "cache_instruments")]
866    fn py_cache_instruments(&self, instruments: Vec<Py<PyAny>>, py: Python<'_>) -> PyResult<()> {
867        let mut instruments_any = Vec::new();
868
869        for inst in instruments {
870            let inst_any = pyobject_to_instrument_any(py, inst)?;
871            instruments_any.push(inst_any);
872        }
873        self.cache_instruments(instruments_any);
874        Ok(())
875    }
876
877    #[pyo3(name = "is_closed")]
878    fn py_is_closed(&self) -> bool {
879        !self.is_connected()
880    }
881
882    /// Subscribes to public trade updates for a specific instrument.
883    ///
884    /// # References
885    ///
886    /// <https://docs.dydx.trade/developers/indexer/websockets#trades-channel>
887    #[pyo3(name = "subscribe_trades")]
888    fn py_subscribe_trades<'py>(
889        &self,
890        py: Python<'py>,
891        instrument_id: InstrumentId,
892    ) -> PyResult<Bound<'py, PyAny>> {
893        let client = self.clone();
894        pyo3_async_runtimes::tokio::future_into_py(py, async move {
895            client
896                .subscribe_trades(instrument_id)
897                .await
898                .map_err(to_pyvalue_err)?;
899            Ok(())
900        })
901    }
902
903    /// Unsubscribes from public trade updates for a specific instrument.
904    #[pyo3(name = "unsubscribe_trades")]
905    fn py_unsubscribe_trades<'py>(
906        &self,
907        py: Python<'py>,
908        instrument_id: InstrumentId,
909    ) -> PyResult<Bound<'py, PyAny>> {
910        let client = self.clone();
911        pyo3_async_runtimes::tokio::future_into_py(py, async move {
912            client
913                .unsubscribe_trades(instrument_id)
914                .await
915                .map_err(to_pyvalue_err)?;
916            Ok(())
917        })
918    }
919
920    /// Subscribes to orderbook updates for a specific instrument.
921    ///
922    /// # References
923    ///
924    /// <https://docs.dydx.trade/developers/indexer/websockets#orderbook-channel>
925    #[pyo3(name = "subscribe_orderbook")]
926    fn py_subscribe_orderbook<'py>(
927        &self,
928        py: Python<'py>,
929        instrument_id: InstrumentId,
930    ) -> PyResult<Bound<'py, PyAny>> {
931        let client = self.clone();
932        pyo3_async_runtimes::tokio::future_into_py(py, async move {
933            client
934                .subscribe_orderbook(instrument_id)
935                .await
936                .map_err(to_pyvalue_err)?;
937            Ok(())
938        })
939    }
940
941    /// Unsubscribes from orderbook updates for a specific instrument.
942    #[pyo3(name = "unsubscribe_orderbook")]
943    fn py_unsubscribe_orderbook<'py>(
944        &self,
945        py: Python<'py>,
946        instrument_id: InstrumentId,
947    ) -> PyResult<Bound<'py, PyAny>> {
948        let client = self.clone();
949        pyo3_async_runtimes::tokio::future_into_py(py, async move {
950            client
951                .unsubscribe_orderbook(instrument_id)
952                .await
953                .map_err(to_pyvalue_err)?;
954            Ok(())
955        })
956    }
957
958    #[pyo3(name = "subscribe_bars")]
959    fn py_subscribe_bars<'py>(
960        &self,
961        py: Python<'py>,
962        bar_type: BarType,
963    ) -> PyResult<Bound<'py, PyAny>> {
964        let spec = bar_type.spec();
965        let resolution = DydxCandleResolution::from_bar_spec(&spec).map_err(to_pyvalue_err)?;
966        let resolution = resolution.to_string();
967
968        let client = self.clone();
969        let instrument_id = bar_type.instrument_id();
970        let bar_types = self.bar_types().clone();
971
972        // Build topic for bar type registration (e.g., "ETH-USD/1MIN")
973        let ticker = extract_raw_symbol(instrument_id.symbol.as_str());
974        let topic = format!("{ticker}/{resolution}");
975
976        pyo3_async_runtimes::tokio::future_into_py(py, async move {
977            bar_types.insert(topic, bar_type);
978
979            client
980                .subscribe_candles(instrument_id, &resolution)
981                .await
982                .map_err(to_pyvalue_err)?;
983            Ok(())
984        })
985    }
986
987    #[pyo3(name = "unsubscribe_bars")]
988    fn py_unsubscribe_bars<'py>(
989        &self,
990        py: Python<'py>,
991        bar_type: BarType,
992    ) -> PyResult<Bound<'py, PyAny>> {
993        let spec = bar_type.spec();
994        let resolution = DydxCandleResolution::from_bar_spec(&spec).map_err(to_pyvalue_err)?;
995        let resolution = resolution.to_string();
996
997        let client = self.clone();
998        let instrument_id = bar_type.instrument_id();
999        let bar_types = self.bar_types().clone();
1000
1001        // Build topic for unregistration
1002        let ticker = extract_raw_symbol(instrument_id.symbol.as_str());
1003        let topic = format!("{ticker}/{resolution}");
1004
1005        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1006            client
1007                .unsubscribe_candles(instrument_id, &resolution)
1008                .await
1009                .map_err(to_pyvalue_err)?;
1010
1011            bar_types.remove(&topic);
1012
1013            Ok(())
1014        })
1015    }
1016
1017    /// Subscribes to market updates for all instruments.
1018    ///
1019    /// # Errors
1020    ///
1021    /// Returns an error if the subscription request fails.
1022    ///
1023    /// # References
1024    ///
1025    /// <https://docs.dydx.trade/developers/indexer/websockets#markets-channel>
1026    #[pyo3(name = "subscribe_markets")]
1027    fn py_subscribe_markets<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
1028        let client = self.clone();
1029        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1030            client.subscribe_markets().await.map_err(to_pyvalue_err)?;
1031            Ok(())
1032        })
1033    }
1034
1035    /// Unsubscribes from market updates.
1036    ///
1037    /// # Errors
1038    ///
1039    /// Returns an error if the unsubscription request fails.
1040    #[pyo3(name = "unsubscribe_markets")]
1041    fn py_unsubscribe_markets<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
1042        let client = self.clone();
1043        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1044            client.unsubscribe_markets().await.map_err(to_pyvalue_err)?;
1045            Ok(())
1046        })
1047    }
1048
1049    /// Subscribes to subaccount updates (orders, fills, positions, balances).
1050    ///
1051    /// This requires authentication and will only work for private WebSocket clients
1052    /// created with `Self.new_private`. Subaccount streams stay pinned to the
1053    /// primary slot: the Indexer caps them at 256 per connection, which is well
1054    /// above realistic per-process usage and keeps related fill/position events
1055    /// on a single in-order stream.
1056    ///
1057    /// # References
1058    ///
1059    /// <https://docs.dydx.trade/developers/indexer/websockets#subaccounts-channel>
1060    #[pyo3(name = "subscribe_subaccount")]
1061    fn py_subscribe_subaccount<'py>(
1062        &self,
1063        py: Python<'py>,
1064        address: String,
1065        subaccount_number: u32,
1066    ) -> PyResult<Bound<'py, PyAny>> {
1067        let client = self.clone();
1068        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1069            client
1070                .subscribe_subaccount(&address, subaccount_number)
1071                .await
1072                .map_err(to_pyvalue_err)?;
1073            Ok(())
1074        })
1075    }
1076
1077    /// Unsubscribes from subaccount updates.
1078    #[pyo3(name = "unsubscribe_subaccount")]
1079    fn py_unsubscribe_subaccount<'py>(
1080        &self,
1081        py: Python<'py>,
1082        address: String,
1083        subaccount_number: u32,
1084    ) -> PyResult<Bound<'py, PyAny>> {
1085        let client = self.clone();
1086        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1087            client
1088                .unsubscribe_subaccount(&address, subaccount_number)
1089                .await
1090                .map_err(to_pyvalue_err)?;
1091            Ok(())
1092        })
1093    }
1094
1095    /// Subscribes to block height updates.
1096    ///
1097    /// # Errors
1098    ///
1099    /// Returns an error if the subscription request fails.
1100    ///
1101    /// # References
1102    ///
1103    /// <https://docs.dydx.trade/developers/indexer/websockets#block-height-channel>
1104    #[pyo3(name = "subscribe_block_height")]
1105    fn py_subscribe_block_height<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
1106        let client = self.clone();
1107        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1108            client
1109                .subscribe_block_height()
1110                .await
1111                .map_err(to_pyvalue_err)?;
1112            Ok(())
1113        })
1114    }
1115
1116    /// Unsubscribes from block height updates.
1117    ///
1118    /// # Errors
1119    ///
1120    /// Returns an error if the unsubscription request fails.
1121    #[pyo3(name = "unsubscribe_block_height")]
1122    fn py_unsubscribe_block_height<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
1123        let client = self.clone();
1124        pyo3_async_runtimes::tokio::future_into_py(py, async move {
1125            client
1126                .unsubscribe_block_height()
1127                .await
1128                .map_err(to_pyvalue_err)?;
1129            Ok(())
1130        })
1131    }
1132}
1133
1134fn instrument_id_from_ticker(ticker: &str) -> InstrumentId {
1135    let symbol = format!("{ticker}-PERP");
1136    InstrumentId::new(Symbol::new(&symbol), *DYDX_VENUE)
1137}
1138
1139fn handle_markets_trading_data(
1140    trading: Option<
1141        &std::collections::HashMap<String, crate::websocket::messages::DydxMarketTradingUpdate>,
1142    >,
1143    instrument_cache: &std::sync::Arc<crate::common::instrument_cache::InstrumentCache>,
1144    seen_tickers: &mut ahash::AHashSet<Ustr>,
1145    call_soon: &Py<PyAny>,
1146    callback: &Py<PyAny>,
1147    ts_init: nautilus_core::UnixNanos,
1148) {
1149    let Some(trading_map) = trading else {
1150        return;
1151    };
1152
1153    for (ticker, update) in trading_map {
1154        let instrument_id = instrument_id_from_ticker(ticker);
1155
1156        if let Some(status) = &update.status {
1157            if *status == DydxMarketStatus::Unknown {
1158                log::warn!("Skipping unmodeled dYdX market status for {instrument_id}");
1159            } else {
1160                let action = MarketStatusAction::from(*status);
1161                let is_trading = matches!(status, DydxMarketStatus::Active);
1162
1163                let instrument_status = InstrumentStatus::new(
1164                    instrument_id,
1165                    action,
1166                    ts_init,
1167                    ts_init,
1168                    None,
1169                    None,
1170                    Some(is_trading),
1171                    None,
1172                    None,
1173                );
1174
1175                if instrument_cache.get_by_market(ticker).is_some() {
1176                    Python::attach(|py| match instrument_status.into_py_any(py) {
1177                        Ok(py_obj) => {
1178                            call_python_threadsafe(py, call_soon, callback, py_obj);
1179                        }
1180                        Err(e) => {
1181                            log::error!("Failed to convert InstrumentStatus to Python: {e}");
1182                        }
1183                    });
1184                }
1185            }
1186        }
1187
1188        let ticker_ustr = Ustr::from(ticker.as_str());
1189        if !seen_tickers.contains(&ticker_ustr) {
1190            let is_active = update
1191                .status
1192                .as_ref()
1193                .is_none_or(|s| matches!(s, crate::common::enums::DydxMarketStatus::Active));
1194            if instrument_cache.get_by_market(ticker).is_some() {
1195                seen_tickers.insert(ticker_ustr);
1196            } else if is_active {
1197                seen_tickers.insert(ticker_ustr);
1198                log::debug!("New instrument discovered via WebSocket: {ticker}");
1199                Python::attach(|py| {
1200                    let dict = PyDict::new(py);
1201                    let _ = dict.set_item("type", "new_instrument_discovered");
1202                    let _ = dict.set_item("ticker", ticker);
1203                    if let Ok(py_obj) = dict.into_py_any(py) {
1204                        call_python_threadsafe(py, call_soon, callback, py_obj);
1205                    }
1206                });
1207            }
1208        }
1209
1210        if let Some(ref rate_str) = update.next_funding_rate {
1211            if let Ok(rate) = Decimal::from_str(rate_str) {
1212                let funding_rate = FundingRateUpdate {
1213                    instrument_id,
1214                    rate,
1215                    interval: Some(60),
1216                    next_funding_ns: None,
1217                    ts_event: ts_init,
1218                    ts_init,
1219                };
1220                Python::attach(|py| match funding_rate.into_py_any(py) {
1221                    Ok(py_obj) => {
1222                        call_python_threadsafe(py, call_soon, callback, py_obj);
1223                    }
1224                    Err(e) => log::error!("Failed to convert FundingRateUpdate to Python: {e}"),
1225                });
1226            } else {
1227                log::warn!("Failed to parse next_funding_rate for {ticker}: {rate_str}");
1228            }
1229        }
1230    }
1231}
1232
1233#[expect(clippy::too_many_arguments)]
1234fn ensure_accepted_to_python(
1235    client_order_id: ClientOrderId,
1236    account_id: AccountId,
1237    venue_order_id: VenueOrderId,
1238    identity: &OrderIdentity,
1239    state: &DydxWsDispatchState,
1240    trader_id: TraderId,
1241    ts_init: nautilus_core::UnixNanos,
1242    call_soon: &Py<PyAny>,
1243    callback: &Py<PyAny>,
1244) {
1245    if state.emitted_accepted.contains(&client_order_id) {
1246        return;
1247    }
1248    state.insert_accepted(client_order_id);
1249    let accepted = OrderAccepted::new(
1250        trader_id,
1251        identity.strategy_id,
1252        identity.instrument_id,
1253        client_order_id,
1254        venue_order_id,
1255        account_id,
1256        UUID4::new(),
1257        ts_init,
1258        ts_init,
1259        false,
1260    );
1261    send_to_python(accepted, call_soon, callback);
1262}
1263
1264fn send_to_python<T: for<'py> IntoPyObjectExt<'py>>(
1265    value: T,
1266    call_soon: &Py<PyAny>,
1267    callback: &Py<PyAny>,
1268) {
1269    Python::attach(|py| match value.into_py_any(py) {
1270        Ok(py_obj) => {
1271            call_python_threadsafe(py, call_soon, callback, py_obj);
1272        }
1273        Err(e) => log::error!("Failed to convert to Python: {e}"),
1274    });
1275}
nautilus_dydx/python/websocket.rs

nautilus_dydx/python/
websocket.rs
