nautilus_bitmex/websocket/
messages.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//! BitMEX WebSocket message structures and supporting types.
17
18use std::{collections::HashMap, fmt::Debug};
19
20use ahash::AHashMap;
21use jiff::Timestamp;
22use nautilus_core::string::secret::REDACTED;
23use rust_decimal::Decimal;
24use serde::{
25    Deserialize, Deserializer, Serialize,
26    de::{self, DeserializeOwned, Error as _},
27};
28use serde_json::{Value, value::RawValue};
29use strum::Display;
30use ustr::Ustr;
31use uuid::Uuid;
32use zeroize::Zeroize;
33
34use super::enums::{
35    BitmexAction, BitmexSide, BitmexTickDirection, BitmexWsAuthAction, BitmexWsOperation,
36};
37use crate::common::{
38    enums::{
39        BitmexContingencyType, BitmexExecInstruction, BitmexExecType, BitmexLiquidityIndicator,
40        BitmexOrderStatus, BitmexOrderType, BitmexPegPriceType, BitmexTimeInForce,
41    },
42    serialization::optional_decimal,
43};
44
45/// Custom deserializer for comma-separated `ExecInstruction` values.
46fn deserialize_exec_instructions<'de, D>(
47    deserializer: D,
48) -> Result<Option<Vec<BitmexExecInstruction>>, D::Error>
49where
50    D: serde::Deserializer<'de>,
51{
52    let s: Option<String> = Option::deserialize(deserializer)?;
53    match s {
54        None => Ok(None),
55        Some(ref s) if s.is_empty() => Ok(None),
56        Some(s) => {
57            let instructions: Result<Vec<BitmexExecInstruction>, _> = s
58                .split(',')
59                .map(|inst| {
60                    let trimmed = inst.trim();
61                    match trimmed {
62                        "ParticipateDoNotInitiate" => {
63                            Ok(BitmexExecInstruction::ParticipateDoNotInitiate)
64                        }
65                        "AllOrNone" => Ok(BitmexExecInstruction::AllOrNone),
66                        "MarkPrice" => Ok(BitmexExecInstruction::MarkPrice),
67                        "IndexPrice" => Ok(BitmexExecInstruction::IndexPrice),
68                        "LastPrice" => Ok(BitmexExecInstruction::LastPrice),
69                        "Close" => Ok(BitmexExecInstruction::Close),
70                        "ReduceOnly" => Ok(BitmexExecInstruction::ReduceOnly),
71                        "Fixed" => Ok(BitmexExecInstruction::Fixed),
72                        "" => Ok(BitmexExecInstruction::Unknown),
73                        _ => Err(format!("Unknown exec instruction: {trimmed}")),
74                    }
75                })
76                .collect();
77            instructions.map(Some).map_err(de::Error::custom)
78        }
79    }
80}
81
82/// BitMEX WebSocket authentication message.
83///
84/// The args array contains [api_key, expires/nonce, signature].
85/// The second element must be a number (not a string) for proper authentication.
86#[derive(Clone, Serialize, Deserialize)]
87pub struct BitmexAuthentication {
88    pub op: BitmexWsAuthAction,
89    pub args: (String, i64, String),
90}
91
92impl Zeroize for BitmexAuthentication {
93    fn zeroize(&mut self) {
94        self.args.0.zeroize();
95        self.args.1.zeroize();
96        self.args.2.zeroize();
97    }
98}
99
100impl Debug for BitmexAuthentication {
101    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
102        f.debug_struct(stringify!(BitmexAuthentication))
103            .field("op", &self.op)
104            .field("api_key", &REDACTED)
105            .field("expires", &self.args.1)
106            .field("signature", &REDACTED)
107            .finish()
108    }
109}
110
111/// BitMEX WebSocket subscription message.
112#[derive(Debug, Clone, Serialize, Deserialize)]
113pub struct BitmexSubscription {
114    pub op: BitmexWsOperation,
115    pub args: Vec<Ustr>,
116}
117
118/// Output message from the BitMEX WebSocket handler.
119///
120/// Contains venue-specific types that consumers parse into Nautilus domain types.
121#[derive(Debug)]
122pub enum BitmexWsMessage {
123    /// Table-based data message from the BitMEX WS stream.
124    Table(BitmexTableMessage),
125    /// Emitted when the underlying WebSocket reconnects.
126    Reconnected,
127    /// Emitted when authentication succeeds.
128    Authenticated,
129}
130
131/// Represents all possible message types from the BitMEX WebSocket API.
132#[derive(Debug, Display, Deserialize)]
133#[serde(untagged)]
134pub(super) enum BitmexWsFrame {
135    /// Table websocket message.
136    #[serde(skip)]
137    Table(BitmexTableMessage),
138    /// Initial welcome message received when connecting to the WebSocket.
139    Welcome {
140        /// Welcome message text.
141        info: String,
142        /// API version string.
143        version: String,
144        /// Server timestamp.
145        timestamp: Timestamp,
146        /// Link to API documentation.
147        docs: String,
148        /// Whether heartbeat is enabled for this connection.
149        #[serde(rename = "heartbeatEnabled")]
150        heartbeat_enabled: bool,
151        /// Rate limit information (absent on some endpoints).
152        limit: Option<BitmexRateLimit>,
153        /// Application name (testnet only).
154        #[serde(rename = "appName")]
155        app_name: Option<String>,
156    },
157    /// Subscription response messages.
158    Subscription {
159        /// Whether the subscription request was successful.
160        success: bool,
161        /// The subscription topic if successful.
162        subscribe: Option<String>,
163        /// Original request metadata (present for subscribe/auth/unsubscribe).
164        request: Option<BitmexHttpRequest>,
165        /// Error message if subscription failed.
166        error: Option<String>,
167    },
168    /// WebSocket error message.
169    Error {
170        status: u16,
171        error: String,
172        meta: HashMap<String, String>,
173        request: BitmexHttpRequest,
174    },
175    /// Indicates a WebSocket reconnection has completed.
176    #[serde(skip)]
177    Reconnected,
178}
179
180#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
181pub struct BitmexHttpRequest {
182    pub op: String,
183    pub args: Vec<Value>,
184}
185
186/// Rate limit information from BitMEX API.
187#[derive(Debug, Deserialize)]
188pub struct BitmexRateLimit {
189    /// Number of requests remaining in the current time window.
190    pub remaining: Option<i32>,
191}
192
193/// Represents table-based messages.
194#[derive(Debug, Display)]
195pub enum BitmexTableMessage {
196    OrderBookL2 {
197        action: BitmexAction,
198        data: Vec<BitmexOrderBookMsg>,
199    },
200    OrderBookL2_25 {
201        action: BitmexAction,
202        data: Vec<BitmexOrderBookMsg>,
203    },
204    OrderBook10 {
205        action: BitmexAction,
206        data: Vec<BitmexOrderBook10Msg>,
207    },
208    Quote {
209        action: BitmexAction,
210        data: Vec<BitmexQuoteMsg>,
211    },
212    Trade {
213        action: BitmexAction,
214        data: Vec<BitmexTradeMsg>,
215    },
216    TradeBin1m {
217        action: BitmexAction,
218        data: Vec<BitmexTradeBinMsg>,
219    },
220    TradeBin5m {
221        action: BitmexAction,
222        data: Vec<BitmexTradeBinMsg>,
223    },
224    TradeBin1h {
225        action: BitmexAction,
226        data: Vec<BitmexTradeBinMsg>,
227    },
228    TradeBin1d {
229        action: BitmexAction,
230        data: Vec<BitmexTradeBinMsg>,
231    },
232    Instrument {
233        action: BitmexAction,
234        data: Vec<BitmexInstrumentMsg>,
235    },
236    Order {
237        action: BitmexAction,
238        data: Vec<OrderData>,
239    },
240    Execution {
241        action: BitmexAction,
242        data: Vec<BitmexExecutionMsg>,
243    },
244    Position {
245        action: BitmexAction,
246        data: Vec<BitmexPositionMsg>,
247    },
248    Wallet {
249        action: BitmexAction,
250        data: Vec<BitmexWalletMsg>,
251    },
252    Margin {
253        action: BitmexAction,
254        data: Vec<BitmexMarginMsg>,
255    },
256    Funding {
257        action: BitmexAction,
258        data: Vec<BitmexFundingMsg>,
259    },
260    Insurance {
261        action: BitmexAction,
262        data: Vec<BitmexInsuranceMsg>,
263    },
264    Liquidation {
265        action: BitmexAction,
266        data: Vec<BitmexLiquidationMsg>,
267    },
268}
269
270#[derive(Deserialize)]
271struct BitmexTableTag {
272    table: Option<String>,
273}
274
275#[derive(Deserialize)]
276struct BitmexTableEnvelope<'a> {
277    table: &'a str,
278    action: BitmexAction,
279    #[serde(borrow)]
280    data: &'a RawValue,
281}
282
283impl BitmexTableMessage {
284    pub(super) fn from_json_if_table(json: &str) -> serde_json::Result<Option<Self>> {
285        let tag: BitmexTableTag = serde_json::from_str(json)?;
286        if tag.table.is_none() {
287            return Ok(None);
288        }
289
290        Self::from_json(json).map(Some)
291    }
292
293    fn from_json(json: &str) -> serde_json::Result<Self> {
294        let envelope: BitmexTableEnvelope = serde_json::from_str(json)?;
295        let action = envelope.action;
296        let data = envelope.data;
297
298        match envelope.table {
299            "orderBookL2" => Ok(Self::OrderBookL2 {
300                action,
301                data: parse_table_data(data)?,
302            }),
303            "orderBookL2_25" => Ok(Self::OrderBookL2_25 {
304                action,
305                data: parse_table_data(data)?,
306            }),
307            "orderBook10" => Ok(Self::OrderBook10 {
308                action,
309                data: parse_table_data(data)?,
310            }),
311            "quote" => Ok(Self::Quote {
312                action,
313                data: parse_table_data(data)?,
314            }),
315            "trade" => Ok(Self::Trade {
316                action,
317                data: parse_table_data(data)?,
318            }),
319            "tradeBin1m" => Ok(Self::TradeBin1m {
320                action,
321                data: parse_table_data(data)?,
322            }),
323            "tradeBin5m" => Ok(Self::TradeBin5m {
324                action,
325                data: parse_table_data(data)?,
326            }),
327            "tradeBin1h" => Ok(Self::TradeBin1h {
328                action,
329                data: parse_table_data(data)?,
330            }),
331            "tradeBin1d" => Ok(Self::TradeBin1d {
332                action,
333                data: parse_table_data(data)?,
334            }),
335            "instrument" => Ok(Self::Instrument {
336                action,
337                data: parse_table_data(data)?,
338            }),
339            "order" => Ok(Self::Order {
340                action,
341                data: parse_order_data(data)?,
342            }),
343            "execution" => Ok(Self::Execution {
344                action,
345                data: parse_table_data(data)?,
346            }),
347            "position" => Ok(Self::Position {
348                action,
349                data: parse_table_data(data)?,
350            }),
351            "wallet" => Ok(Self::Wallet {
352                action,
353                data: parse_table_data(data)?,
354            }),
355            "margin" => Ok(Self::Margin {
356                action,
357                data: parse_table_data(data)?,
358            }),
359            "funding" => Ok(Self::Funding {
360                action,
361                data: parse_table_data(data)?,
362            }),
363            "insurance" => Ok(Self::Insurance {
364                action,
365                data: parse_table_data(data)?,
366            }),
367            "liquidation" => Ok(Self::Liquidation {
368                action,
369                data: parse_table_data(data)?,
370            }),
371            table => Err(serde_json::Error::custom(format!(
372                "unknown BitMEX table `{table}`"
373            ))),
374        }
375    }
376}
377
378impl<'de> Deserialize<'de> for BitmexTableMessage {
379    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
380    where
381        D: Deserializer<'de>,
382    {
383        let raw = Box::<RawValue>::deserialize(deserializer)?;
384        Self::from_json(raw.get()).map_err(D::Error::custom)
385    }
386}
387
388fn parse_table_data<T: DeserializeOwned>(raw: &RawValue) -> serde_json::Result<Vec<T>> {
389    serde_json::from_str(raw.get())
390}
391
392/// Represents a single order book entry in the BitMEX order book.
393#[derive(Clone, Debug, Deserialize)]
394#[serde(rename_all = "camelCase")]
395pub struct BitmexOrderBookMsg {
396    /// The instrument symbol (e.g., "XBTUSD").
397    pub symbol: Ustr,
398    /// Unique order ID.
399    pub id: u64,
400    /// Side of the order ("Buy" or "Sell").
401    pub side: BitmexSide,
402    /// Size of the order, can be None for deletes.
403    pub size: Option<u64>,
404    /// Price level of the order.
405    pub price: f64,
406    /// Timestamp of the update.
407    pub timestamp: Timestamp,
408    /// Timestamp of the transaction.
409    pub transact_time: Timestamp,
410    pub pool: Option<Ustr>,
411}
412
413/// Represents a single order book entry in the BitMEX order book.
414#[derive(Clone, Debug, Deserialize)]
415#[serde(rename_all = "camelCase")]
416pub struct BitmexOrderBook10Msg {
417    /// The instrument symbol (e.g., "XBTUSD").
418    pub symbol: Ustr,
419    /// Array of bid levels, each containing [price, size].
420    pub bids: Vec<[f64; 2]>,
421    /// Array of ask levels, each containing [price, size].
422    pub asks: Vec<[f64; 2]>,
423    /// Timestamp of the orderbook snapshot.
424    pub timestamp: Timestamp,
425    pub pool: Option<Ustr>,
426}
427
428/// Represents a top-of-book quote.
429#[derive(Clone, Debug, Deserialize)]
430#[serde(rename_all = "camelCase")]
431pub struct BitmexQuoteMsg {
432    /// The instrument symbol (e.g., "XBTUSD").
433    pub symbol: Ustr,
434    /// Price of best bid.
435    pub bid_price: Option<f64>,
436    /// Size of best bid.
437    pub bid_size: Option<u64>,
438    /// Price of best ask.
439    pub ask_price: Option<f64>,
440    /// Size of best ask.
441    pub ask_size: Option<u64>,
442    /// Timestamp of the quote.
443    pub timestamp: Timestamp,
444    pub pool: Option<Ustr>,
445}
446
447/// Represents a single trade execution on BitMEX.
448#[derive(Clone, Debug, Deserialize)]
449#[serde(rename_all = "camelCase")]
450pub struct BitmexTradeMsg {
451    /// Timestamp of the trade.
452    pub timestamp: Timestamp,
453    /// The instrument symbol.
454    pub symbol: Ustr,
455    /// Side of the trade ("Buy" or "Sell").
456    pub side: BitmexSide,
457    /// Size of the trade.
458    pub size: u64,
459    /// Price the trade executed at.
460    pub price: f64,
461    /// Direction of the tick ("`PlusTick`", "`MinusTick`", "`ZeroPlusTick`", "`ZeroMinusTick`").
462    pub tick_direction: BitmexTickDirection,
463    /// Unique trade match ID.
464    #[serde(rename = "trdMatchID")]
465    pub trd_match_id: Option<Uuid>,
466    /// Gross value of the trade in satoshis.
467    pub gross_value: Option<i64>,
468    /// Home currency value of the trade.
469    pub home_notional: Option<f64>,
470    /// Foreign currency value of the trade.
471    pub foreign_notional: Option<f64>,
472    /// Trade type.
473    #[serde(rename = "trdType")]
474    pub trade_type: Ustr, // TODO: Add enum
475    pub pool: Option<Ustr>,
476}
477
478#[derive(Clone, Debug, Deserialize)]
479#[serde(rename_all = "camelCase")]
480pub struct BitmexTradeBinMsg {
481    /// Start time of the bin.
482    pub timestamp: Timestamp,
483    /// Trading instrument symbol.
484    pub symbol: Ustr,
485    /// Opening price for the period.
486    pub open: f64,
487    /// Highest price for the period.
488    pub high: f64,
489    /// Lowest price for the period.
490    pub low: f64,
491    /// Closing price for the period.
492    pub close: f64,
493    /// Number of trades in the period.
494    pub trades: i64,
495    /// Volume traded in the period.
496    pub volume: i64,
497    /// Volume weighted average price (None when trades=0).
498    pub vwap: Option<f64>,
499    /// Size of the last trade in the period (None when trades=0).
500    pub last_size: Option<i64>,
501    /// Turnover in satoshis.
502    pub turnover: i64,
503    /// Home currency volume.
504    pub home_notional: f64,
505    /// Foreign currency volume.
506    pub foreign_notional: f64,
507    pub pool: Option<Ustr>,
508}
509
510/// Represents a single order book entry in the BitMEX order book.
511#[derive(Clone, Debug, Deserialize)]
512#[serde(rename_all = "camelCase")]
513pub struct BitmexInstrumentMsg {
514    pub symbol: Ustr,
515    pub root_symbol: Option<Ustr>,
516    pub state: Option<Ustr>,
517    #[serde(rename = "typ")]
518    pub instrument_type: Option<Ustr>,
519    pub listing: Option<Timestamp>,
520    pub front: Option<Timestamp>,
521    pub expiry: Option<Timestamp>,
522    pub settle: Option<Timestamp>,
523    pub listed_settle: Option<Timestamp>,
524    pub position_currency: Option<Ustr>,
525    pub underlying: Option<Ustr>,
526    pub quote_currency: Option<Ustr>,
527    pub underlying_symbol: Option<Ustr>,
528    pub reference: Option<Ustr>,
529    pub reference_symbol: Option<Ustr>,
530    pub max_order_qty: Option<f64>,
531    pub max_price: Option<f64>,
532    pub min_price: Option<f64>,
533    pub lot_size: Option<f64>,
534    pub tick_size: Option<f64>,
535    pub multiplier: Option<f64>,
536    pub settl_currency: Option<Ustr>,
537    pub underlying_to_position_multiplier: Option<f64>,
538    pub underlying_to_settle_multiplier: Option<f64>,
539    pub quote_to_settle_multiplier: Option<f64>,
540    pub is_quanto: Option<bool>,
541    pub is_inverse: Option<bool>,
542    pub init_margin: Option<f64>,
543    pub maint_margin: Option<f64>,
544    pub risk_limit: Option<f64>,
545    pub risk_step: Option<f64>,
546    pub maker_fee: Option<f64>,
547    pub taker_fee: Option<f64>,
548    pub settlement_fee: Option<f64>,
549    pub funding_base_symbol: Option<Ustr>,
550    pub funding_quote_symbol: Option<Ustr>,
551    pub funding_premium_symbol: Option<Ustr>,
552    pub funding_timestamp: Option<Timestamp>,
553    pub funding_interval: Option<Timestamp>,
554    #[serde(default, with = "rust_decimal::serde::float_option")]
555    pub funding_rate: Option<Decimal>,
556    #[serde(default, with = "rust_decimal::serde::float_option")]
557    pub indicative_funding_rate: Option<Decimal>,
558    pub last_price: Option<f64>,
559    pub last_tick_direction: Option<BitmexTickDirection>,
560    pub mark_price: Option<f64>,
561    pub mark_method: Option<Ustr>,
562    pub index_price: Option<f64>,
563    pub indicative_settle_price: Option<f64>,
564    pub indicative_tax_rate: Option<f64>,
565    pub open_interest: Option<i64>,
566    pub open_value: Option<i64>,
567    pub fair_basis: Option<f64>,
568    pub fair_basis_rate: Option<f64>,
569    pub fair_price: Option<f64>,
570    pub timestamp: Timestamp,
571}
572
573impl TryFrom<BitmexInstrumentMsg> for crate::http::models::BitmexInstrument {
574    type Error = anyhow::Error;
575
576    fn try_from(msg: BitmexInstrumentMsg) -> Result<Self, Self::Error> {
577        use crate::common::enums::{BitmexInstrumentState, BitmexInstrumentType};
578
579        // Required fields
580        let root_symbol = msg
581            .root_symbol
582            .ok_or_else(|| anyhow::anyhow!("Missing root_symbol for {}", msg.symbol))?;
583        let underlying = msg
584            .underlying
585            .ok_or_else(|| anyhow::anyhow!("Missing underlying for {}", msg.symbol))?;
586        let quote_currency = msg
587            .quote_currency
588            .ok_or_else(|| anyhow::anyhow!("Missing quote_currency for {}", msg.symbol))?;
589        let tick_size = msg
590            .tick_size
591            .ok_or_else(|| anyhow::anyhow!("Missing tick_size for {}", msg.symbol))?;
592        let multiplier = msg
593            .multiplier
594            .ok_or_else(|| anyhow::anyhow!("Missing multiplier for {}", msg.symbol))?;
595        let is_quanto = msg
596            .is_quanto
597            .ok_or_else(|| anyhow::anyhow!("Missing is_quanto for {}", msg.symbol))?;
598        let is_inverse = msg
599            .is_inverse
600            .ok_or_else(|| anyhow::anyhow!("Missing is_inverse for {}", msg.symbol))?;
601
602        // Parse state - default to Open if not present
603        let state = msg
604            .state
605            .and_then(|s| serde_json::from_str::<BitmexInstrumentState>(&format!("\"{s}\"")).ok())
606            .unwrap_or(BitmexInstrumentState::Open);
607
608        // Parse instrument type - default to PerpetualContract if not present
609        let instrument_type = msg
610            .instrument_type
611            .and_then(|t| serde_json::from_str::<BitmexInstrumentType>(&format!("\"{t}\"")).ok())
612            .unwrap_or(BitmexInstrumentType::PerpetualContract);
613
614        Ok(Self {
615            symbol: msg.symbol,
616            root_symbol,
617            state,
618            instrument_type,
619            listing: msg.listing,
620            front: msg.front,
621            expiry: msg.expiry,
622            settle: msg.settle,
623            listed_settle: msg.listed_settle,
624            position_currency: msg.position_currency,
625            underlying,
626            quote_currency,
627            underlying_symbol: msg.underlying_symbol,
628            reference: msg.reference,
629            reference_symbol: msg.reference_symbol,
630            calc_interval: None,
631            publish_interval: None,
632            publish_time: None,
633            max_order_qty: msg.max_order_qty,
634            max_price: msg.max_price,
635            min_price: msg.min_price,
636            lot_size: msg.lot_size,
637            tick_size,
638            multiplier,
639            settl_currency: msg.settl_currency,
640            underlying_to_position_multiplier: msg.underlying_to_position_multiplier,
641            underlying_to_settle_multiplier: msg.underlying_to_settle_multiplier,
642            quote_to_settle_multiplier: msg.quote_to_settle_multiplier,
643            is_quanto,
644            is_inverse,
645            init_margin: msg.init_margin,
646            maint_margin: msg.maint_margin,
647            risk_limit: msg.risk_limit,
648            risk_step: msg.risk_step,
649            limit: None,
650            taxed: None,
651            deleverage: None,
652            maker_fee: msg.maker_fee,
653            taker_fee: msg.taker_fee,
654            settlement_fee: msg.settlement_fee,
655            funding_base_symbol: msg.funding_base_symbol,
656            funding_quote_symbol: msg.funding_quote_symbol,
657            funding_premium_symbol: msg.funding_premium_symbol,
658            funding_timestamp: msg.funding_timestamp,
659            funding_interval: msg.funding_interval,
660            funding_rate: msg.funding_rate,
661            indicative_funding_rate: msg.indicative_funding_rate,
662            rebalance_timestamp: None,
663            rebalance_interval: None,
664            prev_close_price: None,
665            limit_down_price: None,
666            limit_up_price: None,
667            prev_total_volume: None,
668            total_volume: None,
669            volume: None,
670            volume_24h: None,
671            prev_total_turnover: None,
672            total_turnover: None,
673            turnover: None,
674            turnover_24h: None,
675            home_notional_24h: None,
676            foreign_notional_24h: None,
677            prev_price_24h: None,
678            vwap: None,
679            high_price: None,
680            low_price: None,
681            last_price: msg.last_price,
682            last_price_protected: None,
683            last_tick_direction: None, // WebSocket uses different enum, skip for now
684            last_change_pcnt: None,
685            bid_price: None,
686            mid_price: None,
687            ask_price: None,
688            impact_bid_price: None,
689            impact_mid_price: None,
690            impact_ask_price: None,
691            has_liquidity: None,
692            open_interest: msg.open_interest.map(|v| v as f64),
693            open_value: msg.open_value.map(|v| v as f64),
694            fair_method: None,
695            fair_basis_rate: msg.fair_basis_rate,
696            fair_basis: msg.fair_basis,
697            fair_price: msg.fair_price,
698            mark_method: None,
699            mark_price: msg.mark_price,
700            indicative_settle_price: msg.indicative_settle_price,
701            settled_price_adjustment_rate: None,
702            settled_price: None,
703            instant_pnl: false,
704            min_tick: None,
705            funding_base_rate: None,
706            funding_quote_rate: None,
707            capped: None,
708            opening_timestamp: None,
709            closing_timestamp: None,
710            timestamp: msg.timestamp,
711        })
712    }
713}
714
715/// Represents an order update message with only changed fields.
716/// Used for `update` actions where only modified fields are sent.
717#[derive(Clone, Debug, Deserialize)]
718#[serde(rename_all = "camelCase")]
719pub struct BitmexOrderUpdateMsg {
720    #[serde(rename = "orderID")]
721    pub order_id: Uuid,
722    #[serde(rename = "clOrdID")]
723    pub cl_ord_id: Option<Ustr>,
724    pub account: Option<i64>,
725    pub symbol: Option<Ustr>,
726    pub side: Option<BitmexSide>,
727    #[serde(default)]
728    pub price: FieldUpdate<f64>,
729    pub currency: Option<Ustr>,
730    #[serde(default)]
731    pub text: FieldUpdate<Ustr>,
732    pub transact_time: Option<Timestamp>,
733    pub timestamp: Option<Timestamp>,
734    pub leaves_qty: Option<i64>,
735    pub cum_qty: Option<i64>,
736    #[serde(default, deserialize_with = "deserialize_decimal_update")]
737    pub avg_px: FieldUpdate<Decimal>,
738    pub ord_status: Option<BitmexOrderStatus>,
739}
740
741/// A field in a sparse table update.
742#[derive(Clone, Debug, Default, PartialEq)]
743pub enum FieldUpdate<T> {
744    /// The field was absent and the cached value remains unchanged.
745    #[default]
746    Missing,
747    /// The field was present with a null value.
748    Null,
749    /// The field was present with a value.
750    Value(T),
751}
752
753impl<'de, T> Deserialize<'de> for FieldUpdate<T>
754where
755    T: Deserialize<'de>,
756{
757    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
758    where
759        D: Deserializer<'de>,
760    {
761        Ok(match Option::<T>::deserialize(deserializer)? {
762            Some(value) => Self::Value(value),
763            None => Self::Null,
764        })
765    }
766}
767
768fn deserialize_decimal_update<'de, D>(deserializer: D) -> Result<FieldUpdate<Decimal>, D::Error>
769where
770    D: Deserializer<'de>,
771{
772    Ok(match optional_decimal::deserialize(deserializer)? {
773        Some(value) => FieldUpdate::Value(value),
774        None => FieldUpdate::Null,
775    })
776}
777
778/// Represents a full order message from the WebSocket stream.
779/// Used for `insert` and `partial` actions where all fields are present.
780#[derive(Clone, Debug, Deserialize)]
781#[serde(rename_all = "camelCase")]
782pub struct BitmexOrderMsg {
783    #[serde(rename = "orderID")]
784    pub order_id: Uuid,
785    #[serde(rename = "clOrdID")]
786    pub cl_ord_id: Option<Ustr>,
787    #[serde(rename = "clOrdLinkID")]
788    pub cl_ord_link_id: Option<Ustr>,
789    pub account: i64,
790    pub symbol: Ustr,
791    pub side: BitmexSide,
792    pub order_qty: i64,
793    pub price: Option<f64>,
794    pub display_qty: Option<i64>,
795    pub stop_px: Option<f64>,
796    pub peg_offset_value: Option<f64>,
797    pub peg_price_type: Option<BitmexPegPriceType>,
798    pub currency: Ustr,
799    pub settl_currency: Ustr,
800    pub ord_type: Option<BitmexOrderType>,
801    pub time_in_force: Option<BitmexTimeInForce>,
802    #[serde(default, deserialize_with = "deserialize_exec_instructions")]
803    pub exec_inst: Option<Vec<BitmexExecInstruction>>,
804    pub contingency_type: Option<BitmexContingencyType>,
805    pub ord_status: BitmexOrderStatus,
806    pub triggered: Option<Ustr>,
807    pub working_indicator: bool,
808    pub ord_rej_reason: Option<Ustr>,
809    pub leaves_qty: i64,
810    pub cum_qty: i64,
811    #[serde(default, with = "optional_decimal")]
812    pub avg_px: Option<Decimal>,
813    pub text: Option<Ustr>,
814    pub transact_time: Timestamp,
815    pub timestamp: Timestamp,
816    pub strategy: Option<Ustr>,
817    pub pool: Option<Ustr>,
818}
819
820/// Wrapper enum for order data that can be either full or update messages.
821#[derive(Clone, Debug)]
822pub enum OrderData {
823    Full(BitmexOrderMsg),
824    Update(BitmexOrderUpdateMsg),
825}
826
827#[derive(Debug)]
828pub(crate) enum ResolvedOrderData {
829    Full(BitmexOrderMsg),
830    Update(BitmexOrderUpdateMsg),
831    Terminal(BitmexOrderMsg),
832}
833
834#[derive(Debug, Default)]
835pub(crate) struct OrderRowCache {
836    rows: AHashMap<Uuid, BitmexOrderMsg>,
837}
838
839impl OrderRowCache {
840    pub(crate) fn apply(
841        &mut self,
842        action: BitmexAction,
843        data: Vec<OrderData>,
844    ) -> Vec<ResolvedOrderData> {
845        if action == BitmexAction::Partial {
846            self.clear();
847        }
848
849        data.into_iter()
850            .filter_map(|order_data| self.resolve(order_data))
851            .collect()
852    }
853
854    pub(crate) fn clear(&mut self) {
855        self.rows.clear();
856    }
857
858    fn resolve(&mut self, order_data: OrderData) -> Option<ResolvedOrderData> {
859        match order_data {
860            OrderData::Full(order) => {
861                self.store(&order);
862                Some(ResolvedOrderData::Full(order))
863            }
864            OrderData::Update(mut update) => {
865                let order_id = update.order_id;
866                let Some(mut merged) = self.rows.get(&order_id).cloned() else {
867                    log::warn!("Order update cache miss: order_id={order_id}");
868                    return None;
869                };
870
871                update.apply_to(&mut merged);
872                update.inherit_context(&merged);
873
874                if merged.ord_status.is_terminal() {
875                    self.rows.remove(&order_id);
876                    Some(ResolvedOrderData::Terminal(merged))
877                } else {
878                    self.rows.insert(order_id, merged);
879                    Some(ResolvedOrderData::Update(update))
880                }
881            }
882        }
883    }
884
885    fn store(&mut self, order: &BitmexOrderMsg) {
886        if order.ord_status.is_terminal() {
887            self.rows.remove(&order.order_id);
888        } else {
889            self.rows.insert(order.order_id, order.clone());
890        }
891    }
892}
893
894impl BitmexOrderUpdateMsg {
895    fn apply_to(&self, order: &mut BitmexOrderMsg) {
896        if let Some(cl_ord_id) = self.cl_ord_id {
897            order.cl_ord_id = Some(cl_ord_id);
898        }
899
900        if let Some(account) = self.account {
901            order.account = account;
902        }
903
904        if let Some(symbol) = self.symbol {
905            order.symbol = symbol;
906        }
907
908        if let Some(side) = self.side {
909            order.side = side;
910        }
911        self.price.apply_to(&mut order.price);
912        if let Some(currency) = self.currency {
913            order.currency = currency;
914        }
915        self.text.apply_to(&mut order.text);
916        if let Some(transact_time) = self.transact_time {
917            order.transact_time = transact_time;
918        }
919
920        if let Some(timestamp) = self.timestamp {
921            order.timestamp = timestamp;
922        }
923
924        if let Some(leaves_qty) = self.leaves_qty {
925            order.leaves_qty = leaves_qty;
926        }
927
928        if let Some(cum_qty) = self.cum_qty {
929            order.cum_qty = cum_qty;
930        }
931        self.avg_px.apply_to(&mut order.avg_px);
932
933        if let Some(ord_status) = self.ord_status {
934            order.ord_status = ord_status;
935        }
936    }
937
938    fn inherit_context(&mut self, order: &BitmexOrderMsg) {
939        self.cl_ord_id = self.cl_ord_id.or(order.cl_ord_id);
940        self.account = self.account.or(Some(order.account));
941        self.symbol = self.symbol.or(Some(order.symbol));
942    }
943}
944
945impl<T> FieldUpdate<T> {
946    pub(crate) const fn value(&self) -> Option<&T> {
947        match self {
948            Self::Value(value) => Some(value),
949            Self::Missing | Self::Null => None,
950        }
951    }
952
953    fn apply_to(&self, target: &mut Option<T>)
954    where
955        T: Clone,
956    {
957        match self {
958            Self::Missing => {}
959            Self::Null => *target = None,
960            Self::Value(value) => *target = Some(value.clone()),
961        }
962    }
963}
964
965fn parse_order_data(raw: &RawValue) -> serde_json::Result<Vec<OrderData>> {
966    let raw_values: Vec<Box<RawValue>> = serde_json::from_str(raw.get())?;
967    let mut result = Vec::new();
968
969    for value in raw_values {
970        // Try to deserialize as full message first
971        if let Ok(full_msg) = serde_json::from_str::<BitmexOrderMsg>(value.get()) {
972            result.push(OrderData::Full(full_msg));
973        } else if let Ok(update_msg) = serde_json::from_str::<BitmexOrderUpdateMsg>(value.get()) {
974            result.push(OrderData::Update(update_msg));
975        } else {
976            return Err(serde_json::Error::custom(
977                "Failed to deserialize order data as either full or update message",
978            ));
979        }
980    }
981
982    Ok(result)
983}
984
985/// Raw Order and Balance Data.
986#[derive(Clone, Debug, Deserialize)]
987#[serde(rename_all = "camelCase")]
988pub struct BitmexExecutionMsg {
989    #[serde(rename = "execID")]
990    pub exec_id: Option<Uuid>,
991    #[serde(rename = "orderID")]
992    pub order_id: Option<Uuid>,
993    #[serde(rename = "clOrdID")]
994    pub cl_ord_id: Option<Ustr>,
995    #[serde(rename = "clOrdLinkID")]
996    pub cl_ord_link_id: Option<Ustr>,
997    pub account: Option<i64>,
998    pub symbol: Option<Ustr>,
999    pub side: Option<BitmexSide>,
1000    pub last_qty: Option<i64>,
1001    pub last_px: Option<f64>,
1002    pub underlying_last_px: Option<f64>,
1003    pub last_mkt: Option<Ustr>,
1004    pub last_liquidity_ind: Option<BitmexLiquidityIndicator>,
1005    pub order_qty: Option<i64>,
1006    pub price: Option<f64>,
1007    pub display_qty: Option<i64>,
1008    pub stop_px: Option<f64>,
1009    pub peg_offset_value: Option<f64>,
1010    pub peg_price_type: Option<BitmexPegPriceType>,
1011    pub currency: Option<Ustr>,
1012    pub settl_currency: Option<Ustr>,
1013    pub exec_type: Option<BitmexExecType>,
1014    pub ord_type: Option<BitmexOrderType>,
1015    pub time_in_force: Option<BitmexTimeInForce>,
1016    #[serde(default, deserialize_with = "deserialize_exec_instructions")]
1017    pub exec_inst: Option<Vec<BitmexExecInstruction>>,
1018    pub contingency_type: Option<BitmexContingencyType>,
1019    pub ex_destination: Option<Ustr>,
1020    pub ord_status: Option<BitmexOrderStatus>,
1021    pub triggered: Option<Ustr>,
1022    pub working_indicator: Option<bool>,
1023    pub ord_rej_reason: Option<Ustr>,
1024    pub leaves_qty: Option<i64>,
1025    pub cum_qty: Option<i64>,
1026    pub avg_px: Option<f64>,
1027    pub commission: Option<f64>,
1028    pub trade_publish_indicator: Option<Ustr>,
1029    pub multi_leg_reporting_type: Option<Ustr>,
1030    pub text: Option<Ustr>,
1031    #[serde(rename = "trdMatchID")]
1032    pub trd_match_id: Option<Uuid>,
1033    pub exec_cost: Option<i64>,
1034    pub exec_comm: Option<i64>,
1035    pub home_notional: Option<f64>,
1036    pub foreign_notional: Option<f64>,
1037    pub transact_time: Option<Timestamp>,
1038    pub timestamp: Option<Timestamp>,
1039    pub strategy: Option<Ustr>,
1040    pub pool: Option<Ustr>,
1041    pub exec_comm_ccy: Option<Ustr>,
1042}
1043
1044/// Position status.
1045#[derive(Clone, Debug, Deserialize)]
1046#[serde(rename_all = "camelCase")]
1047pub struct BitmexPositionMsg {
1048    pub account: i64,
1049    pub symbol: Ustr,
1050    pub currency: Option<Ustr>,
1051    pub underlying: Option<Ustr>,
1052    pub quote_currency: Option<Ustr>,
1053    pub commission: Option<f64>,
1054    pub init_margin_req: Option<f64>,
1055    pub maint_margin_req: Option<f64>,
1056    pub risk_limit: Option<i64>,
1057    pub leverage: Option<f64>,
1058    pub cross_margin: Option<bool>,
1059    pub deleverage_percentile: Option<f64>,
1060    pub rebalanced_pnl: Option<i64>,
1061    pub prev_realised_pnl: Option<i64>,
1062    pub prev_unrealised_pnl: Option<i64>,
1063    pub prev_close_price: Option<f64>,
1064    pub opening_timestamp: Option<Timestamp>,
1065    pub opening_qty: Option<i64>,
1066    pub opening_cost: Option<i64>,
1067    pub opening_comm: Option<i64>,
1068    pub open_order_buy_qty: Option<i64>,
1069    pub open_order_buy_cost: Option<i64>,
1070    pub open_order_buy_premium: Option<i64>,
1071    pub open_order_sell_qty: Option<i64>,
1072    pub open_order_sell_cost: Option<i64>,
1073    pub open_order_sell_premium: Option<i64>,
1074    pub exec_buy_qty: Option<i64>,
1075    pub exec_buy_cost: Option<i64>,
1076    pub exec_sell_qty: Option<i64>,
1077    pub exec_sell_cost: Option<i64>,
1078    pub exec_qty: Option<i64>,
1079    pub exec_cost: Option<i64>,
1080    pub exec_comm: Option<i64>,
1081    pub current_timestamp: Option<Timestamp>,
1082    pub current_qty: Option<i64>,
1083    pub current_cost: Option<i64>,
1084    pub current_comm: Option<i64>,
1085    pub realised_cost: Option<i64>,
1086    pub unrealised_cost: Option<i64>,
1087    pub gross_open_cost: Option<i64>,
1088    pub gross_open_premium: Option<i64>,
1089    pub gross_exec_cost: Option<i64>,
1090    pub is_open: Option<bool>,
1091    pub mark_price: Option<f64>,
1092    pub mark_value: Option<i64>,
1093    pub risk_value: Option<i64>,
1094    pub home_notional: Option<f64>,
1095    pub foreign_notional: Option<f64>,
1096    pub pos_state: Option<Ustr>,
1097    pub pos_cost: Option<i64>,
1098    pub pos_cost2: Option<i64>,
1099    pub pos_cross: Option<i64>,
1100    pub pos_init: Option<i64>,
1101    pub pos_comm: Option<i64>,
1102    pub pos_loss: Option<i64>,
1103    pub pos_margin: Option<i64>,
1104    pub pos_maint: Option<i64>,
1105    pub pos_allowance: Option<i64>,
1106    pub taxable_margin: Option<i64>,
1107    pub init_margin: Option<i64>,
1108    pub maint_margin: Option<i64>,
1109    pub session_margin: Option<i64>,
1110    pub target_excess_margin: Option<i64>,
1111    pub var_margin: Option<i64>,
1112    pub realised_gross_pnl: Option<i64>,
1113    pub realised_tax: Option<i64>,
1114    pub realised_pnl: Option<i64>,
1115    pub unrealised_gross_pnl: Option<i64>,
1116    pub long_bankrupt: Option<i64>,
1117    pub short_bankrupt: Option<i64>,
1118    pub tax_base: Option<i64>,
1119    pub indicative_tax_rate: Option<f64>,
1120    pub indicative_tax: Option<i64>,
1121    pub unrealised_tax: Option<i64>,
1122    pub unrealised_pnl: Option<i64>,
1123    pub unrealised_pnl_pcnt: Option<f64>,
1124    pub unrealised_roe_pcnt: Option<f64>,
1125    pub avg_cost_price: Option<f64>,
1126    pub avg_entry_price: Option<f64>,
1127    pub break_even_price: Option<f64>,
1128    pub margin_call_price: Option<f64>,
1129    pub liquidation_price: Option<f64>,
1130    pub bankrupt_price: Option<f64>,
1131    pub timestamp: Option<Timestamp>,
1132    pub last_price: Option<f64>,
1133    pub last_value: Option<i64>,
1134    pub strategy: Option<Ustr>,
1135}
1136
1137#[derive(Clone, Debug, Deserialize)]
1138#[serde(rename_all = "camelCase")]
1139pub struct BitmexWalletMsg {
1140    pub account: i64,
1141    pub currency: Ustr,
1142    pub prev_deposited: Option<i64>,
1143    pub prev_withdrawn: Option<i64>,
1144    pub prev_transfer_in: Option<i64>,
1145    pub prev_transfer_out: Option<i64>,
1146    pub prev_amount: Option<i64>,
1147    pub prev_timestamp: Option<Timestamp>,
1148    pub delta_deposited: Option<i64>,
1149    pub delta_withdrawn: Option<i64>,
1150    pub delta_transfer_in: Option<i64>,
1151    pub delta_transfer_out: Option<i64>,
1152    pub delta_amount: Option<i64>,
1153    pub deposited: Option<i64>,
1154    pub withdrawn: Option<i64>,
1155    pub transfer_in: Option<i64>,
1156    pub transfer_out: Option<i64>,
1157    pub amount: Option<i64>,
1158    pub pending_credit: Option<i64>,
1159    pub pending_debit: Option<i64>,
1160    pub confirmed_debit: Option<i64>,
1161    pub timestamp: Option<Timestamp>,
1162    pub addr: Option<Ustr>,
1163    pub script: Option<Ustr>,
1164    pub withdrawal_lock: Option<Vec<Ustr>>,
1165}
1166
1167/// Represents margin account information
1168#[derive(Clone, Debug, Deserialize)]
1169#[serde(rename_all = "camelCase")]
1170pub struct BitmexMarginMsg {
1171    /// Account identifier
1172    pub account: i64,
1173    /// Currency of the margin account
1174    pub currency: Ustr,
1175    /// Risk limit for the account
1176    pub risk_limit: Option<i64>,
1177    /// Current amount in the account
1178    pub amount: Option<i64>,
1179    /// Previously realized PnL
1180    pub prev_realised_pnl: Option<i64>,
1181    /// Gross commission
1182    pub gross_comm: Option<i64>,
1183    /// Gross open cost
1184    pub gross_open_cost: Option<i64>,
1185    /// Gross open premium
1186    pub gross_open_premium: Option<i64>,
1187    /// Gross execution cost
1188    pub gross_exec_cost: Option<i64>,
1189    /// Gross mark value
1190    pub gross_mark_value: Option<i64>,
1191    /// Risk value
1192    pub risk_value: Option<i64>,
1193    /// Initial margin requirement
1194    pub init_margin: Option<i64>,
1195    /// Maintenance margin requirement
1196    pub maint_margin: Option<i64>,
1197    /// Target excess margin
1198    pub target_excess_margin: Option<i64>,
1199    /// Realized profit and loss
1200    pub realised_pnl: Option<i64>,
1201    /// Unrealized profit and loss
1202    pub unrealised_pnl: Option<i64>,
1203    /// Wallet balance
1204    pub wallet_balance: Option<i64>,
1205    /// Margin balance
1206    pub margin_balance: Option<i64>,
1207    /// Margin leverage
1208    pub margin_leverage: Option<f64>,
1209    /// Margin used percentage
1210    pub margin_used_pcnt: Option<f64>,
1211    /// Excess margin
1212    pub excess_margin: Option<i64>,
1213    /// Available margin
1214    pub available_margin: Option<i64>,
1215    /// Withdrawable margin
1216    pub withdrawable_margin: Option<i64>,
1217    /// Maker fee discount
1218    pub maker_fee_discount: Option<f64>,
1219    /// Taker fee discount
1220    pub taker_fee_discount: Option<f64>,
1221    /// Timestamp of the margin update
1222    pub timestamp: Timestamp,
1223    /// Foreign margin balance
1224    pub foreign_margin_balance: Option<i64>,
1225    /// Foreign margin requirement
1226    pub foreign_requirement: Option<i64>,
1227}
1228
1229/// Represents a funding rate update.
1230#[derive(Clone, Debug, Deserialize)]
1231#[serde(rename_all = "camelCase")]
1232pub struct BitmexFundingMsg {
1233    /// Timestamp of the funding update.
1234    pub timestamp: Timestamp,
1235    /// The instrument symbol the funding applies to.
1236    pub symbol: Ustr,
1237    /// The interval for this funding.
1238    pub funding_interval: Timestamp,
1239    /// The funding rate for this interval.
1240    #[serde(with = "rust_decimal::serde::float")]
1241    pub funding_rate: Decimal,
1242    /// The daily funding rate.
1243    #[serde(with = "rust_decimal::serde::float")]
1244    pub funding_rate_daily: Decimal,
1245}
1246
1247/// Represents an insurance fund update.
1248#[derive(Clone, Debug, Deserialize)]
1249#[serde(rename_all = "camelCase")]
1250pub struct BitmexInsuranceMsg {
1251    /// The currency of the insurance fund.
1252    pub currency: Ustr,
1253    /// Timestamp of the update.
1254    pub timestamp: Timestamp,
1255    /// Current balance of the insurance wallet.
1256    pub wallet_balance: i64,
1257}
1258
1259/// Represents a liquidation order.
1260#[derive(Clone, Debug, Deserialize)]
1261#[serde(rename_all = "camelCase")]
1262pub struct BitmexLiquidationMsg {
1263    /// Unique order ID of the liquidation.
1264    pub order_id: Ustr,
1265    /// The instrument symbol being liquidated.
1266    pub symbol: Ustr,
1267    /// Side of the liquidation ("Buy" or "Sell").
1268    pub side: BitmexSide,
1269    /// Price of the liquidation order.
1270    pub price: f64,
1271    /// Remaining quantity to be executed.
1272    pub leaves_qty: i64,
1273}
1274
1275#[cfg(test)]
1276mod tests {
1277    use rstest::rstest;
1278
1279    use super::*;
1280
1281    #[rstest]
1282    fn test_authentication_redacts_and_zeroizes() {
1283        let mut auth = BitmexAuthentication {
1284            op: BitmexWsAuthAction::AuthKeyExpires,
1285            args: (
1286                "api-key-sentinel".to_string(),
1287                123_456,
1288                "signature-sentinel".to_string(),
1289            ),
1290        };
1291
1292        let debug = format!("{auth:?}");
1293        auth.zeroize();
1294
1295        assert_eq!(debug.matches(REDACTED).count(), 2);
1296        assert!(!debug.contains("api-key-sentinel"));
1297        assert!(!debug.contains("signature-sentinel"));
1298        assert_eq!(auth.args.0, "");
1299        assert_eq!(auth.args.1, 0);
1300        assert_eq!(auth.args.2, "");
1301    }
1302
1303    #[rstest]
1304    fn test_try_from_instrument_msg_with_full_data_success() {
1305        let json_data = r#"{
1306            "symbol": "XBTUSD",
1307            "rootSymbol": "XBT",
1308            "state": "Open",
1309            "typ": "FFWCSX",
1310            "listing": "2016-05-13T12:00:00.000Z",
1311            "front": "2016-05-13T12:00:00.000Z",
1312            "positionCurrency": "USD",
1313            "underlying": "XBT",
1314            "quoteCurrency": "USD",
1315            "underlyingSymbol": "XBT=",
1316            "reference": "BMEX",
1317            "referenceSymbol": ".BXBT",
1318            "maxOrderQty": 10000000,
1319            "maxPrice": 1000000,
1320            "lotSize": 100,
1321            "tickSize": 0.1,
1322            "multiplier": -100000000,
1323            "settlCurrency": "XBt",
1324            "underlyingToSettleMultiplier": -100000000,
1325            "isQuanto": false,
1326            "isInverse": true,
1327            "initMargin": 0.01,
1328            "maintMargin": 0.005,
1329            "riskLimit": 20000000000,
1330            "riskStep": 15000000000,
1331            "taxed": true,
1332            "deleverage": true,
1333            "makerFee": 0.0005,
1334            "takerFee": 0.0005,
1335            "settlementFee": 0,
1336            "fundingBaseSymbol": ".XBTBON8H",
1337            "fundingQuoteSymbol": ".USDBON8H",
1338            "fundingPremiumSymbol": ".XBTUSDPI8H",
1339            "fundingTimestamp": "2024-11-25T04:00:00.000Z",
1340            "fundingInterval": "2000-01-01T08:00:00.000Z",
1341            "fundingRate": 0.00011,
1342            "indicativeFundingRate": 0.000125,
1343            "prevClosePrice": 97409.63,
1344            "limitDownPrice": null,
1345            "limitUpPrice": null,
1346            "prevTotalVolume": 3868480147789,
1347            "totalVolume": 3868507398889,
1348            "volume": 27251100,
1349            "volume24h": 419742700,
1350            "prevTotalTurnover": 37667656761390205,
1351            "totalTurnover": 37667684492745237,
1352            "turnover": 27731355032,
1353            "turnover24h": 431762899194,
1354            "homeNotional24h": 4317.62899194,
1355            "foreignNotional24h": 419742700,
1356            "prevPrice24h": 97655,
1357            "vwap": 97216.6863,
1358            "highPrice": 98743.5,
1359            "lowPrice": 95802.9,
1360            "lastPrice": 97893.7,
1361            "lastPriceProtected": 97912.5054,
1362            "lastTickDirection": "PlusTick",
1363            "lastChangePcnt": 0.0024,
1364            "bidPrice": 97882.5,
1365            "midPrice": 97884.8,
1366            "askPrice": 97887.1,
1367            "impactBidPrice": 97882.7951,
1368            "impactMidPrice": 97884.7,
1369            "impactAskPrice": 97886.6277,
1370            "hasLiquidity": true,
1371            "openInterest": 411647400,
1372            "openValue": 420691293378,
1373            "fairMethod": "FundingRate",
1374            "fairBasisRate": 0.12045,
1375            "fairBasis": 5.99,
1376            "fairPrice": 97849.76,
1377            "markMethod": "FairPrice",
1378            "markPrice": 97849.76,
1379            "indicativeSettlePrice": 97843.77,
1380            "instantPnl": true,
1381            "timestamp": "2024-11-24T23:33:19.034Z",
1382            "minTick": 0.01,
1383            "fundingBaseRate": 0.0003,
1384            "fundingQuoteRate": 0.0006,
1385            "capped": false
1386        }"#;
1387
1388        let ws_msg: BitmexInstrumentMsg =
1389            serde_json::from_str(json_data).expect("Failed to deserialize instrument message");
1390
1391        let result = crate::http::models::BitmexInstrument::try_from(ws_msg);
1392        assert!(
1393            result.is_ok(),
1394            "TryFrom should succeed with full instrument data"
1395        );
1396
1397        let instrument = result.unwrap();
1398        assert_eq!(instrument.symbol.as_str(), "XBTUSD");
1399        assert_eq!(instrument.root_symbol, "XBT");
1400        assert_eq!(instrument.quote_currency, "USD");
1401        assert_eq!(instrument.tick_size, 0.1);
1402    }
1403
1404    #[rstest]
1405    fn test_try_from_instrument_msg_with_partial_data_fails() {
1406        let json_data = r#"{
1407            "symbol": "XBTUSD",
1408            "lastPrice": 95123.5,
1409            "lastTickDirection": "ZeroPlusTick",
1410            "markPrice": 95125.7,
1411            "indexPrice": 95124.3,
1412            "indicativeSettlePrice": 95126.0,
1413            "openInterest": 123456789,
1414            "openValue": 1234567890,
1415            "fairBasis": 1.4,
1416            "fairBasisRate": 0.00001,
1417            "fairPrice": 95125.0,
1418            "markMethod": "FairPrice",
1419            "indicativeTaxRate": 0.00075,
1420            "timestamp": "2024-11-25T12:00:00.000Z"
1421        }"#;
1422
1423        let ws_msg: BitmexInstrumentMsg =
1424            serde_json::from_str(json_data).expect("Failed to deserialize instrument message");
1425
1426        let result = crate::http::models::BitmexInstrument::try_from(ws_msg);
1427        assert!(
1428            result.is_err(),
1429            "TryFrom should fail with partial instrument data (update action)"
1430        );
1431
1432        let err = result.unwrap_err();
1433        assert!(
1434            err.to_string().contains("Missing"),
1435            "Error should indicate missing required fields"
1436        );
1437    }
1438
1439    #[rstest]
1440    fn test_order_sparse_update_deserializes_exactly() {
1441        let message: BitmexTableMessage = serde_json::from_str(include_str!(
1442            "../../test_data/ws_order_update_canceled.json"
1443        ))
1444        .unwrap();
1445        let BitmexTableMessage::Order { action, data } = message else {
1446            panic!("expected order table message");
1447        };
1448        let OrderData::Update(update) = &data[0] else {
1449            panic!("expected sparse order update");
1450        };
1451
1452        assert_eq!(action, BitmexAction::Update);
1453        assert_eq!(
1454            update.order_id,
1455            Uuid::parse_str("550e8400-e29b-41d4-a716-446655440001").unwrap()
1456        );
1457        assert_eq!(update.ord_status, Some(BitmexOrderStatus::Canceled));
1458        assert_eq!(
1459            update.avg_px,
1460            FieldUpdate::Value("30000.500000000004".parse::<Decimal>().unwrap())
1461        );
1462    }
1463
1464    #[rstest]
1465    fn test_order_avg_px_deserializes_exactly() {
1466        let message: BitmexTableMessage =
1467            serde_json::from_str(include_str!("../../test_data/ws_order_avg_px.json")).unwrap();
1468        let BitmexTableMessage::Order { data, .. } = message else {
1469            panic!("expected order table message");
1470        };
1471        let OrderData::Full(order) = &data[0] else {
1472            panic!("expected full order message");
1473        };
1474
1475        assert_eq!(
1476            order.avg_px,
1477            Some("30000.500000000004".parse::<Decimal>().unwrap())
1478        );
1479    }
1480
1481    #[rstest]
1482    fn test_order_row_cache_merges_sparse_update_and_evicts_terminal_order() {
1483        let order: BitmexOrderMsg =
1484            serde_json::from_str(include_str!("../../test_data/ws_order.json")).unwrap();
1485        let original = order.clone();
1486        let message: BitmexTableMessage = serde_json::from_str(include_str!(
1487            "../../test_data/ws_order_update_canceled.json"
1488        ))
1489        .unwrap();
1490        let BitmexTableMessage::Order { action, data } = message else {
1491            panic!("expected order table message");
1492        };
1493        let mut cache = OrderRowCache::default();
1494
1495        cache.apply(BitmexAction::Partial, vec![OrderData::Full(order)]);
1496        let resolved = cache.apply(action, data);
1497        let ResolvedOrderData::Terminal(canceled) = &resolved[0] else {
1498            panic!("expected resolved terminal order");
1499        };
1500
1501        assert_eq!(canceled.cl_ord_id, original.cl_ord_id);
1502        assert_eq!(canceled.account, original.account);
1503        assert_eq!(canceled.symbol, original.symbol);
1504        assert_eq!(canceled.price, original.price);
1505        assert_eq!(canceled.text, original.text);
1506        assert_eq!(
1507            canceled.avg_px,
1508            Some("30000.500000000004".parse::<Decimal>().unwrap())
1509        );
1510        assert_eq!(canceled.ord_status, BitmexOrderStatus::Canceled);
1511        assert!(cache.rows.is_empty());
1512    }
1513
1514    #[rstest]
1515    fn test_order_row_cache_applies_explicit_nulls() {
1516        let order: BitmexOrderMsg =
1517            serde_json::from_str(include_str!("../../test_data/ws_order.json")).unwrap();
1518        let message: BitmexTableMessage =
1519            serde_json::from_str(include_str!("../../test_data/ws_order_update_nulls.json"))
1520                .unwrap();
1521        let BitmexTableMessage::Order { action, data } = message else {
1522            panic!("expected order table message");
1523        };
1524        let order_id = order.order_id;
1525        let mut cache = OrderRowCache::default();
1526
1527        cache.apply(BitmexAction::Partial, vec![OrderData::Full(order)]);
1528        cache.apply(action, data);
1529        let cached = cache.rows.get(&order_id).unwrap();
1530
1531        assert_eq!(cached.price, None);
1532        assert_eq!(cached.text, None);
1533    }
1534
1535    #[rstest]
1536    fn test_order_row_cache_applies_values() {
1537        let order: BitmexOrderMsg =
1538            serde_json::from_str(include_str!("../../test_data/ws_order.json")).unwrap();
1539        let message: BitmexTableMessage =
1540            serde_json::from_str(include_str!("../../test_data/ws_order_update_values.json"))
1541                .unwrap();
1542        let BitmexTableMessage::Order { action, data } = message else {
1543            panic!("expected order table message");
1544        };
1545        let order_id = order.order_id;
1546        let mut cache = OrderRowCache::default();
1547
1548        cache.apply(BitmexAction::Partial, vec![OrderData::Full(order)]);
1549        cache.apply(action, data);
1550        let cached = cache.rows.get(&order_id).unwrap();
1551
1552        assert_eq!(cached.price, Some(99_000.0));
1553        assert_eq!(cached.text, Some(Ustr::from("Amended")));
1554    }
1555
1556    #[rstest]
1557    fn test_order_row_cache_resets_partial_and_evicts_full_terminal_order() {
1558        let order: BitmexOrderMsg =
1559            serde_json::from_str(include_str!("../../test_data/ws_order.json")).unwrap();
1560        let order_id = order.order_id;
1561        let mut cache = OrderRowCache::default();
1562
1563        cache.apply(BitmexAction::Partial, vec![OrderData::Full(order.clone())]);
1564        assert!(cache.rows.contains_key(&order_id));
1565
1566        cache.apply(BitmexAction::Partial, Vec::new());
1567        assert!(cache.rows.is_empty());
1568
1569        let mut terminal = order;
1570        terminal.ord_status = BitmexOrderStatus::Canceled;
1571        cache.apply(BitmexAction::Insert, vec![OrderData::Full(terminal)]);
1572        assert!(cache.rows.is_empty());
1573    }
1574}
nautilus_bitmex/websocket/messages.rs

nautilus_bitmex/websocket/
messages.rs
