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

nautilus_bitmex/websocket/
messages.rs
