nautilus_bybit/websocket/
parse.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//! Parsing helpers for Bybit WebSocket payloads.
17
18use std::convert::TryFrom;
19
20use anyhow::Context;
21use nautilus_core::{datetime::NANOSECONDS_IN_MILLISECOND, nanos::UnixNanos, uuid::UUID4};
22use nautilus_model::{
23    data::{
24        Bar, BarType, BookOrder, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate,
25        OrderBookDelta, OrderBookDeltas, QuoteTick, TradeTick,
26    },
27    enums::{
28        AccountType, AggressorSide, BookAction, LiquiditySide, OrderSide, OrderStatus, OrderType,
29        PositionSideSpecified, RecordFlag, TimeInForce, TriggerType,
30    },
31    events::account::state::AccountState,
32    identifiers::{AccountId, ClientOrderId, InstrumentId, TradeId, VenueOrderId},
33    instruments::{Instrument, any::InstrumentAny},
34    reports::{FillReport, OrderStatusReport, PositionStatusReport},
35    types::{AccountBalance, Money, Price, Quantity},
36};
37use rust_decimal::Decimal;
38
39use super::messages::{
40    BybitWsAccountExecution, BybitWsAccountOrder, BybitWsAccountPosition, BybitWsAccountWallet,
41    BybitWsKline, BybitWsOrderbookDepthMsg, BybitWsTickerLinear, BybitWsTickerLinearMsg,
42    BybitWsTickerOptionMsg, BybitWsTrade,
43};
44use crate::common::{
45    consts::BYBIT_TOPIC_KLINE,
46    enums::{
47        BybitOrderStatus, BybitOrderType, BybitStopOrderType, BybitTimeInForce,
48        BybitTriggerDirection,
49    },
50    parse::{
51        get_currency, parse_book_level, parse_millis_timestamp, parse_price_with_precision,
52        parse_quantity_with_precision,
53    },
54};
55
56/// Parses a Bybit WebSocket topic string into its components.
57///
58/// # Errors
59///
60/// Returns an error if the topic format is invalid.
61pub fn parse_topic(topic: &str) -> anyhow::Result<Vec<&str>> {
62    let parts: Vec<&str> = topic.split('.').collect();
63    if parts.is_empty() {
64        anyhow::bail!("Invalid topic format: empty topic");
65    }
66    Ok(parts)
67}
68
69/// Parses a Bybit kline topic into (interval, symbol).
70///
71/// Topic format: "kline.{interval}.{symbol}" (e.g., "kline.5.BTCUSDT")
72///
73/// # Errors
74///
75/// Returns an error if the topic format is invalid.
76pub fn parse_kline_topic(topic: &str) -> anyhow::Result<(&str, &str)> {
77    let parts = parse_topic(topic)?;
78    if parts.len() != 3 || parts[0] != BYBIT_TOPIC_KLINE {
79        anyhow::bail!(
80            "Invalid kline topic format: expected '{BYBIT_TOPIC_KLINE}.{{interval}}.{{symbol}}', was '{topic}'"
81        );
82    }
83    Ok((parts[1], parts[2]))
84}
85
86/// Parses a WebSocket trade frame into a [`TradeTick`].
87pub fn parse_ws_trade_tick(
88    trade: &BybitWsTrade,
89    instrument: &InstrumentAny,
90    ts_init: UnixNanos,
91) -> anyhow::Result<TradeTick> {
92    let price = parse_price_with_precision(&trade.p, instrument.price_precision(), "trade.p")?;
93    let size = parse_quantity_with_precision(&trade.v, instrument.size_precision(), "trade.v")?;
94    let aggressor: AggressorSide = trade.taker_side.into();
95    let trade_id = TradeId::new_checked(trade.i.as_str())
96        .context("invalid trade identifier in Bybit trade message")?;
97    let ts_event = parse_millis_i64(trade.t, "trade.T")?;
98
99    TradeTick::new_checked(
100        instrument.id(),
101        price,
102        size,
103        aggressor,
104        trade_id,
105        ts_event,
106        ts_init,
107    )
108    .context("failed to construct TradeTick from Bybit trade message")
109}
110
111/// Parses an order book depth message into [`OrderBookDeltas`].
112pub fn parse_orderbook_deltas(
113    msg: &BybitWsOrderbookDepthMsg,
114    instrument: &InstrumentAny,
115    ts_init: UnixNanos,
116) -> anyhow::Result<OrderBookDeltas> {
117    let is_snapshot = msg.msg_type.eq_ignore_ascii_case("snapshot");
118    let ts_event = parse_millis_i64(msg.ts, "orderbook.ts")?;
119    let ts_init = if ts_init.is_zero() { ts_event } else { ts_init };
120
121    let depth = &msg.data;
122    let instrument_id = instrument.id();
123    let price_precision = instrument.price_precision();
124    let size_precision = instrument.size_precision();
125    let update_id = u64::try_from(depth.u)
126        .context("received negative update id in Bybit order book message")?;
127    let sequence = u64::try_from(depth.seq)
128        .context("received negative sequence in Bybit order book message")?;
129
130    let total_levels = depth.b.len() + depth.a.len();
131    let capacity = if is_snapshot {
132        total_levels + 1
133    } else {
134        total_levels
135    };
136    let mut deltas = Vec::with_capacity(capacity);
137
138    if is_snapshot {
139        deltas.push(OrderBookDelta::clear(
140            instrument_id,
141            sequence,
142            ts_event,
143            ts_init,
144        ));
145    }
146    let mut processed = 0_usize;
147
148    let mut push_level = |values: &[String], side: OrderSide| -> anyhow::Result<()> {
149        let (price, size) = parse_book_level(values, price_precision, size_precision, "orderbook")?;
150        let action = if size.is_zero() {
151            BookAction::Delete
152        } else if is_snapshot {
153            BookAction::Add
154        } else {
155            BookAction::Update
156        };
157
158        processed += 1;
159        let mut flags = RecordFlag::F_MBP as u8;
160        if processed == total_levels {
161            flags |= RecordFlag::F_LAST as u8;
162        }
163
164        let order = BookOrder::new(side, price, size, update_id);
165        let delta = OrderBookDelta::new_checked(
166            instrument_id,
167            action,
168            order,
169            flags,
170            sequence,
171            ts_event,
172            ts_init,
173        )
174        .context("failed to construct OrderBookDelta from Bybit book level")?;
175        deltas.push(delta);
176        Ok(())
177    };
178
179    for level in &depth.b {
180        push_level(level, OrderSide::Buy)?;
181    }
182    for level in &depth.a {
183        push_level(level, OrderSide::Sell)?;
184    }
185
186    if total_levels == 0
187        && let Some(last) = deltas.last_mut()
188    {
189        last.flags |= RecordFlag::F_LAST as u8;
190    }
191
192    OrderBookDeltas::new_checked(instrument_id, deltas)
193        .context("failed to assemble OrderBookDeltas from Bybit message")
194}
195
196/// Parses an order book snapshot or delta into a [`QuoteTick`].
197pub fn parse_orderbook_quote(
198    msg: &BybitWsOrderbookDepthMsg,
199    instrument: &InstrumentAny,
200    last_quote: Option<&QuoteTick>,
201    ts_init: UnixNanos,
202) -> anyhow::Result<QuoteTick> {
203    let ts_event = parse_millis_i64(msg.ts, "orderbook.ts")?;
204    let ts_init = if ts_init.is_zero() { ts_event } else { ts_init };
205    let price_precision = instrument.price_precision();
206    let size_precision = instrument.size_precision();
207
208    let get_best =
209        |levels: &[Vec<String>], label: &str| -> anyhow::Result<Option<(Price, Quantity)>> {
210            if let Some(values) = levels.first() {
211                parse_book_level(values, price_precision, size_precision, label).map(Some)
212            } else {
213                Ok(None)
214            }
215        };
216
217    let bids = get_best(&msg.data.b, "bid")?;
218    let asks = get_best(&msg.data.a, "ask")?;
219
220    let (bid_price, bid_size) = match (bids, last_quote) {
221        (Some(level), _) => level,
222        (None, Some(prev)) => (prev.bid_price, prev.bid_size),
223        (None, None) => {
224            anyhow::bail!(
225                "Bybit order book update missing bid levels and no previous quote provided"
226            );
227        }
228    };
229
230    let (ask_price, ask_size) = match (asks, last_quote) {
231        (Some(level), _) => level,
232        (None, Some(prev)) => (prev.ask_price, prev.ask_size),
233        (None, None) => {
234            anyhow::bail!(
235                "Bybit order book update missing ask levels and no previous quote provided"
236            );
237        }
238    };
239
240    QuoteTick::new_checked(
241        instrument.id(),
242        bid_price,
243        ask_price,
244        bid_size,
245        ask_size,
246        ts_event,
247        ts_init,
248    )
249    .context("failed to construct QuoteTick from Bybit order book message")
250}
251
252/// Parses a linear or inverse ticker payload into a [`QuoteTick`].
253pub fn parse_ticker_linear_quote(
254    msg: &BybitWsTickerLinearMsg,
255    instrument: &InstrumentAny,
256    ts_init: UnixNanos,
257) -> anyhow::Result<QuoteTick> {
258    let ts_event = parse_millis_i64(msg.ts, "ticker.ts")?;
259    let ts_init = if ts_init.is_zero() { ts_event } else { ts_init };
260    let price_precision = instrument.price_precision();
261    let size_precision = instrument.size_precision();
262
263    let data = &msg.data;
264    let bid_price = data
265        .bid1_price
266        .as_ref()
267        .context("Bybit ticker message missing bid1Price")?
268        .as_str();
269    let ask_price = data
270        .ask1_price
271        .as_ref()
272        .context("Bybit ticker message missing ask1Price")?
273        .as_str();
274
275    let bid_price = parse_price_with_precision(bid_price, price_precision, "ticker.bid1Price")?;
276    let ask_price = parse_price_with_precision(ask_price, price_precision, "ticker.ask1Price")?;
277
278    let bid_size_str = data.bid1_size.as_deref().unwrap_or("0");
279    let ask_size_str = data.ask1_size.as_deref().unwrap_or("0");
280
281    let bid_size = parse_quantity_with_precision(bid_size_str, size_precision, "ticker.bid1Size")?;
282    let ask_size = parse_quantity_with_precision(ask_size_str, size_precision, "ticker.ask1Size")?;
283
284    QuoteTick::new_checked(
285        instrument.id(),
286        bid_price,
287        ask_price,
288        bid_size,
289        ask_size,
290        ts_event,
291        ts_init,
292    )
293    .context("failed to construct QuoteTick from Bybit linear ticker message")
294}
295
296/// Parses an option ticker payload into a [`QuoteTick`].
297pub fn parse_ticker_option_quote(
298    msg: &BybitWsTickerOptionMsg,
299    instrument: &InstrumentAny,
300    ts_init: UnixNanos,
301) -> anyhow::Result<QuoteTick> {
302    let ts_event = parse_millis_i64(msg.ts, "ticker.ts")?;
303    let ts_init = if ts_init.is_zero() { ts_event } else { ts_init };
304    let price_precision = instrument.price_precision();
305    let size_precision = instrument.size_precision();
306
307    let data = &msg.data;
308    let bid_price =
309        parse_price_with_precision(&data.bid_price, price_precision, "ticker.bidPrice")?;
310    let ask_price =
311        parse_price_with_precision(&data.ask_price, price_precision, "ticker.askPrice")?;
312    let bid_size = parse_quantity_with_precision(&data.bid_size, size_precision, "ticker.bidSize")?;
313    let ask_size = parse_quantity_with_precision(&data.ask_size, size_precision, "ticker.askSize")?;
314
315    QuoteTick::new_checked(
316        instrument.id(),
317        bid_price,
318        ask_price,
319        bid_size,
320        ask_size,
321        ts_event,
322        ts_init,
323    )
324    .context("failed to construct QuoteTick from Bybit option ticker message")
325}
326
327/// Parses a linear ticker payload into a [`FundingRateUpdate`].
328///
329/// # Errors
330///
331/// Returns an error if funding rate or next funding time fields are missing or cannot be parsed.
332pub fn parse_ticker_linear_funding(
333    data: &BybitWsTickerLinear,
334    instrument_id: InstrumentId,
335    ts_event: UnixNanos,
336    ts_init: UnixNanos,
337) -> anyhow::Result<FundingRateUpdate> {
338    let funding_rate_str = data
339        .funding_rate
340        .as_ref()
341        .context("Bybit ticker missing funding_rate")?;
342
343    let funding_rate = funding_rate_str
344        .as_str()
345        .parse::<Decimal>()
346        .context("invalid funding_rate value")?;
347
348    let next_funding_ns = if let Some(next_funding_time) = &data.next_funding_time {
349        let next_funding_millis = next_funding_time
350            .as_str()
351            .parse::<i64>()
352            .context("invalid next_funding_time value")?;
353        Some(parse_millis_i64(next_funding_millis, "next_funding_time")?)
354    } else {
355        None
356    };
357
358    Ok(FundingRateUpdate::new(
359        instrument_id,
360        funding_rate,
361        next_funding_ns,
362        ts_event,
363        ts_init,
364    ))
365}
366
367/// Parses a linear/inverse ticker payload into a [`MarkPriceUpdate`].
368///
369/// # Errors
370///
371/// Returns an error if the mark_price field is missing or cannot be parsed.
372pub fn parse_ticker_linear_mark_price(
373    data: &BybitWsTickerLinear,
374    instrument: &InstrumentAny,
375    ts_event: UnixNanos,
376    ts_init: UnixNanos,
377) -> anyhow::Result<MarkPriceUpdate> {
378    let mark_price_str = data
379        .mark_price
380        .as_ref()
381        .context("Bybit ticker missing mark_price")?;
382
383    let price =
384        parse_price_with_precision(mark_price_str, instrument.price_precision(), "mark_price")?;
385
386    Ok(MarkPriceUpdate::new(
387        instrument.id(),
388        price,
389        ts_event,
390        ts_init,
391    ))
392}
393
394/// Parses a linear/inverse ticker payload into an [`IndexPriceUpdate`].
395///
396/// # Errors
397///
398/// Returns an error if the index_price field is missing or cannot be parsed.
399pub fn parse_ticker_linear_index_price(
400    data: &BybitWsTickerLinear,
401    instrument: &InstrumentAny,
402    ts_event: UnixNanos,
403    ts_init: UnixNanos,
404) -> anyhow::Result<IndexPriceUpdate> {
405    let index_price_str = data
406        .index_price
407        .as_ref()
408        .context("Bybit ticker missing index_price")?;
409
410    let price =
411        parse_price_with_precision(index_price_str, instrument.price_precision(), "index_price")?;
412
413    Ok(IndexPriceUpdate::new(
414        instrument.id(),
415        price,
416        ts_event,
417        ts_init,
418    ))
419}
420
421/// Parses an option ticker payload into a [`MarkPriceUpdate`].
422///
423/// # Errors
424///
425/// Returns an error if the mark_price field cannot be parsed.
426pub fn parse_ticker_option_mark_price(
427    msg: &BybitWsTickerOptionMsg,
428    instrument: &InstrumentAny,
429    ts_init: UnixNanos,
430) -> anyhow::Result<MarkPriceUpdate> {
431    let ts_event = parse_millis_i64(msg.ts, "ticker.ts")?;
432
433    let price = parse_price_with_precision(
434        &msg.data.mark_price,
435        instrument.price_precision(),
436        "mark_price",
437    )?;
438
439    Ok(MarkPriceUpdate::new(
440        instrument.id(),
441        price,
442        ts_event,
443        ts_init,
444    ))
445}
446
447/// Parses an option ticker payload into an [`IndexPriceUpdate`].
448///
449/// # Errors
450///
451/// Returns an error if the index_price field cannot be parsed.
452pub fn parse_ticker_option_index_price(
453    msg: &BybitWsTickerOptionMsg,
454    instrument: &InstrumentAny,
455    ts_init: UnixNanos,
456) -> anyhow::Result<IndexPriceUpdate> {
457    let ts_event = parse_millis_i64(msg.ts, "ticker.ts")?;
458
459    let price = parse_price_with_precision(
460        &msg.data.index_price,
461        instrument.price_precision(),
462        "index_price",
463    )?;
464
465    Ok(IndexPriceUpdate::new(
466        instrument.id(),
467        price,
468        ts_event,
469        ts_init,
470    ))
471}
472
473pub(crate) fn parse_millis_i64(value: i64, field: &str) -> anyhow::Result<UnixNanos> {
474    if value < 0 {
475        Err(anyhow::anyhow!("{field} must be non-negative, was {value}"))
476    } else {
477        let nanos = (value as u64)
478            .checked_mul(NANOSECONDS_IN_MILLISECOND)
479            .ok_or_else(|| anyhow::anyhow!("millisecond timestamp overflowed"))?;
480        Ok(UnixNanos::from(nanos))
481    }
482}
483
484/// Parses a WebSocket kline payload into a [`Bar`].
485///
486/// # Errors
487///
488/// Returns an error if price or volume fields cannot be parsed or if the bar cannot be constructed.
489pub fn parse_ws_kline_bar(
490    kline: &BybitWsKline,
491    instrument: &InstrumentAny,
492    bar_type: BarType,
493    timestamp_on_close: bool,
494    ts_init: UnixNanos,
495) -> anyhow::Result<Bar> {
496    let price_precision = instrument.price_precision();
497    let size_precision = instrument.size_precision();
498
499    let open = parse_price_with_precision(&kline.open, price_precision, "kline.open")?;
500    let high = parse_price_with_precision(&kline.high, price_precision, "kline.high")?;
501    let low = parse_price_with_precision(&kline.low, price_precision, "kline.low")?;
502    let close = parse_price_with_precision(&kline.close, price_precision, "kline.close")?;
503    let volume = parse_quantity_with_precision(&kline.volume, size_precision, "kline.volume")?;
504
505    let mut ts_event = parse_millis_i64(kline.start, "kline.start")?;
506    if timestamp_on_close {
507        let interval_ns = bar_type
508            .spec()
509            .timedelta()
510            .num_nanoseconds()
511            .context("bar specification produced non-integer interval")?;
512        let interval_ns = u64::try_from(interval_ns)
513            .context("bar interval overflowed the u64 range for nanoseconds")?;
514        let updated = ts_event
515            .as_u64()
516            .checked_add(interval_ns)
517            .context("bar timestamp overflowed when adjusting to close time")?;
518        ts_event = UnixNanos::from(updated);
519    }
520    let ts_init = if ts_init.is_zero() { ts_event } else { ts_init };
521
522    Bar::new_checked(bar_type, open, high, low, close, volume, ts_event, ts_init)
523        .context("failed to construct Bar from Bybit WebSocket kline")
524}
525
526/// Parses a WebSocket account order payload into an [`OrderStatusReport`].
527///
528/// # Errors
529///
530/// Returns an error if price or quantity fields cannot be parsed or timestamps are invalid.
531pub fn parse_ws_order_status_report(
532    order: &BybitWsAccountOrder,
533    instrument: &InstrumentAny,
534    account_id: AccountId,
535    ts_init: UnixNanos,
536) -> anyhow::Result<OrderStatusReport> {
537    use crate::common::enums::BybitOrderSide;
538
539    let instrument_id = instrument.id();
540    let venue_order_id = VenueOrderId::new(order.order_id.as_str());
541    let order_side: OrderSide = order.side.into();
542
543    // Bybit represents conditional orders using orderType + stopOrderType + triggerDirection + side
544    let order_type: OrderType = match (
545        order.order_type,
546        order.stop_order_type,
547        order.trigger_direction,
548        order.side,
549    ) {
550        (BybitOrderType::Market, BybitStopOrderType::None | BybitStopOrderType::Unknown, _, _) => {
551            OrderType::Market
552        }
553        (BybitOrderType::Limit, BybitStopOrderType::None | BybitStopOrderType::Unknown, _, _) => {
554            OrderType::Limit
555        }
556
557        (
558            BybitOrderType::Market,
559            BybitStopOrderType::Stop,
560            BybitTriggerDirection::RisesTo,
561            BybitOrderSide::Buy,
562        ) => OrderType::StopMarket,
563        (
564            BybitOrderType::Market,
565            BybitStopOrderType::Stop,
566            BybitTriggerDirection::FallsTo,
567            BybitOrderSide::Buy,
568        ) => OrderType::MarketIfTouched,
569
570        (
571            BybitOrderType::Market,
572            BybitStopOrderType::Stop,
573            BybitTriggerDirection::FallsTo,
574            BybitOrderSide::Sell,
575        ) => OrderType::StopMarket,
576        (
577            BybitOrderType::Market,
578            BybitStopOrderType::Stop,
579            BybitTriggerDirection::RisesTo,
580            BybitOrderSide::Sell,
581        ) => OrderType::MarketIfTouched,
582
583        (
584            BybitOrderType::Limit,
585            BybitStopOrderType::Stop,
586            BybitTriggerDirection::RisesTo,
587            BybitOrderSide::Buy,
588        ) => OrderType::StopLimit,
589        (
590            BybitOrderType::Limit,
591            BybitStopOrderType::Stop,
592            BybitTriggerDirection::FallsTo,
593            BybitOrderSide::Buy,
594        ) => OrderType::LimitIfTouched,
595
596        (
597            BybitOrderType::Limit,
598            BybitStopOrderType::Stop,
599            BybitTriggerDirection::FallsTo,
600            BybitOrderSide::Sell,
601        ) => OrderType::StopLimit,
602        (
603            BybitOrderType::Limit,
604            BybitStopOrderType::Stop,
605            BybitTriggerDirection::RisesTo,
606            BybitOrderSide::Sell,
607        ) => OrderType::LimitIfTouched,
608
609        // triggerDirection=None means regular order with TP/SL attached, not a standalone conditional order
610        (BybitOrderType::Market, BybitStopOrderType::Stop, BybitTriggerDirection::None, _) => {
611            OrderType::Market
612        }
613        (BybitOrderType::Limit, BybitStopOrderType::Stop, BybitTriggerDirection::None, _) => {
614            OrderType::Limit
615        }
616
617        // TP/SL stopOrderTypes are attached to positions, not standalone conditional orders
618        (BybitOrderType::Market, _, _, _) => OrderType::Market,
619        (BybitOrderType::Limit, _, _, _) => OrderType::Limit,
620
621        (BybitOrderType::Unknown, _, _, _) => OrderType::Limit,
622    };
623
624    let time_in_force: TimeInForce = match order.time_in_force {
625        BybitTimeInForce::Gtc => TimeInForce::Gtc,
626        BybitTimeInForce::Ioc => TimeInForce::Ioc,
627        BybitTimeInForce::Fok => TimeInForce::Fok,
628        BybitTimeInForce::PostOnly => TimeInForce::Gtc,
629    };
630
631    let quantity =
632        parse_quantity_with_precision(&order.qty, instrument.size_precision(), "order.qty")?;
633
634    let filled_qty = parse_quantity_with_precision(
635        &order.cum_exec_qty,
636        instrument.size_precision(),
637        "order.cumExecQty",
638    )?;
639
640    // Map Bybit order status to Nautilus order status
641    // Special case: if Bybit reports "Rejected" but the order has fills, treat it as Canceled.
642    // This handles the case where the exchange partially fills an order then rejects the
643    // remaining quantity (e.g., due to margin, risk limits, or liquidity constraints).
644    // The state machine does not allow PARTIALLY_FILLED -> REJECTED transitions.
645    let order_status: OrderStatus = match order.order_status {
646        BybitOrderStatus::Created | BybitOrderStatus::New | BybitOrderStatus::Untriggered => {
647            OrderStatus::Accepted
648        }
649        BybitOrderStatus::Rejected => {
650            if filled_qty.is_positive() {
651                OrderStatus::Canceled
652            } else {
653                OrderStatus::Rejected
654            }
655        }
656        BybitOrderStatus::PartiallyFilled => OrderStatus::PartiallyFilled,
657        BybitOrderStatus::Filled => OrderStatus::Filled,
658        BybitOrderStatus::Canceled | BybitOrderStatus::PartiallyFilledCanceled => {
659            OrderStatus::Canceled
660        }
661        BybitOrderStatus::Triggered => OrderStatus::Triggered,
662        BybitOrderStatus::Deactivated => OrderStatus::Canceled,
663    };
664
665    let ts_accepted = parse_millis_timestamp(&order.created_time, "order.createdTime")?;
666    let ts_last = parse_millis_timestamp(&order.updated_time, "order.updatedTime")?;
667
668    let mut report = OrderStatusReport::new(
669        account_id,
670        instrument_id,
671        None,
672        venue_order_id,
673        order_side,
674        order_type,
675        time_in_force,
676        order_status,
677        quantity,
678        filled_qty,
679        ts_accepted,
680        ts_last,
681        ts_init,
682        Some(UUID4::new()),
683    );
684
685    if !order.order_link_id.is_empty() {
686        report = report.with_client_order_id(ClientOrderId::new(order.order_link_id.as_str()));
687    }
688
689    if !order.price.is_empty() && order.price != "0" {
690        let price =
691            parse_price_with_precision(&order.price, instrument.price_precision(), "order.price")?;
692        report = report.with_price(price);
693    }
694
695    if !order.avg_price.is_empty() && order.avg_price != "0" {
696        let avg_px = order
697            .avg_price
698            .parse::<f64>()
699            .with_context(|| format!("Failed to parse avg_price='{}' as f64", order.avg_price))?;
700        report = report.with_avg_px(avg_px)?;
701    }
702
703    if !order.trigger_price.is_empty() && order.trigger_price != "0" {
704        let trigger_price = parse_price_with_precision(
705            &order.trigger_price,
706            instrument.price_precision(),
707            "order.triggerPrice",
708        )?;
709        report = report.with_trigger_price(trigger_price);
710
711        // Set trigger_type for conditional orders
712        let trigger_type: TriggerType = order.trigger_by.into();
713        report = report.with_trigger_type(trigger_type);
714    }
715
716    if order.reduce_only {
717        report = report.with_reduce_only(true);
718    }
719
720    if order.time_in_force == BybitTimeInForce::PostOnly {
721        report = report.with_post_only(true);
722    }
723
724    if !order.reject_reason.is_empty() {
725        report = report.with_cancel_reason(order.reject_reason.to_string());
726    }
727
728    Ok(report)
729}
730
731/// Parses a WebSocket account execution payload into a [`FillReport`].
732///
733/// # Errors
734///
735/// Returns an error if price or quantity fields cannot be parsed or timestamps are invalid.
736pub fn parse_ws_fill_report(
737    execution: &BybitWsAccountExecution,
738    account_id: AccountId,
739    instrument: &InstrumentAny,
740    ts_init: UnixNanos,
741) -> anyhow::Result<FillReport> {
742    let instrument_id = instrument.id();
743    let venue_order_id = VenueOrderId::new(execution.order_id.as_str());
744    let trade_id = TradeId::new_checked(execution.exec_id.as_str())
745        .context("invalid execId in Bybit WebSocket execution payload")?;
746
747    let order_side: OrderSide = execution.side.into();
748    let last_qty = parse_quantity_with_precision(
749        &execution.exec_qty,
750        instrument.size_precision(),
751        "execution.execQty",
752    )?;
753    let last_px = parse_price_with_precision(
754        &execution.exec_price,
755        instrument.price_precision(),
756        "execution.execPrice",
757    )?;
758
759    let liquidity_side = if execution.is_maker {
760        LiquiditySide::Maker
761    } else {
762        LiquiditySide::Taker
763    };
764
765    let fee_decimal: Decimal = execution
766        .exec_fee
767        .parse()
768        .with_context(|| format!("Failed to parse execFee='{}'", execution.exec_fee))?;
769
770    let commission_currency = instrument.quote_currency();
771    let commission = Money::from_decimal(fee_decimal, commission_currency).with_context(|| {
772        format!(
773            "Failed to create commission from execFee='{}'",
774            execution.exec_fee
775        )
776    })?;
777    let ts_event = parse_millis_timestamp(&execution.exec_time, "execution.execTime")?;
778
779    let client_order_id = if execution.order_link_id.is_empty() {
780        None
781    } else {
782        Some(ClientOrderId::new(execution.order_link_id.as_str()))
783    };
784
785    Ok(FillReport::new(
786        account_id,
787        instrument_id,
788        venue_order_id,
789        trade_id,
790        order_side,
791        last_qty,
792        last_px,
793        commission,
794        liquidity_side,
795        client_order_id,
796        None, // venue_position_id
797        ts_event,
798        ts_init,
799        None, // report_id
800    ))
801}
802
803/// Parses a WebSocket account position payload into a [`PositionStatusReport`].
804///
805/// # Errors
806///
807/// Returns an error if position size or prices cannot be parsed.
808pub fn parse_ws_position_status_report(
809    position: &BybitWsAccountPosition,
810    account_id: AccountId,
811    instrument: &InstrumentAny,
812    ts_init: UnixNanos,
813) -> anyhow::Result<PositionStatusReport> {
814    let instrument_id = instrument.id();
815
816    // Parse absolute size as unsigned Quantity
817    let quantity = parse_quantity_with_precision(
818        &position.size,
819        instrument.size_precision(),
820        "position.size",
821    )?;
822
823    // Derive position side from the side field
824    let position_side = if position.side.eq_ignore_ascii_case("buy") {
825        PositionSideSpecified::Long
826    } else if position.side.eq_ignore_ascii_case("sell") {
827        PositionSideSpecified::Short
828    } else {
829        PositionSideSpecified::Flat
830    };
831
832    let ts_last = parse_millis_timestamp(&position.updated_time, "position.updatedTime")?;
833
834    Ok(PositionStatusReport::new(
835        account_id,
836        instrument_id,
837        position_side,
838        quantity,
839        ts_last,
840        ts_init,
841        None,                 // report_id
842        None,                 // venue_position_id
843        position.entry_price, // avg_px_open
844    ))
845}
846
847/// Parses a WebSocket account wallet payload into an [`AccountState`].
848///
849/// # Errors
850///
851/// Returns an error if balance fields cannot be parsed.
852pub fn parse_ws_account_state(
853    wallet: &BybitWsAccountWallet,
854    account_id: AccountId,
855    ts_event: UnixNanos,
856    ts_init: UnixNanos,
857) -> anyhow::Result<AccountState> {
858    let mut balances = Vec::new();
859
860    for coin_data in &wallet.coin {
861        let currency = get_currency(coin_data.coin.as_str());
862        let total_dec = coin_data.wallet_balance - coin_data.spot_borrow;
863        let locked_dec = coin_data.total_order_im + coin_data.total_position_im;
864
865        let total = Money::from_decimal(total_dec, currency)?;
866        let locked = Money::from_decimal(locked_dec, currency)?;
867        let free = Money::from_raw(total.raw - locked.raw, currency);
868
869        let balance = AccountBalance::new(total, locked, free);
870        balances.push(balance);
871    }
872
873    Ok(AccountState::new(
874        account_id,
875        AccountType::Margin, // Bybit unified account
876        balances,
877        vec![], // margins - Bybit doesn't provide per-instrument margin in wallet updates
878        true,   // is_reported
879        UUID4::new(),
880        ts_event,
881        ts_init,
882        None, // base_currency
883    ))
884}
885
886#[cfg(test)]
887mod tests {
888    use nautilus_model::{
889        data::BarSpecification,
890        enums::{AggregationSource, BarAggregation, PositionSide, PriceType},
891    };
892    use rstest::rstest;
893    use rust_decimal_macros::dec;
894
895    use super::*;
896    use crate::{
897        common::{
898            parse::{parse_linear_instrument, parse_option_instrument},
899            testing::load_test_json,
900        },
901        http::models::{BybitInstrumentLinearResponse, BybitInstrumentOptionResponse},
902        websocket::messages::{
903            BybitWsOrderbookDepthMsg, BybitWsTickerLinearMsg, BybitWsTickerOptionMsg,
904            BybitWsTradeMsg,
905        },
906    };
907
908    const TS: UnixNanos = UnixNanos::new(1_700_000_000_000_000_000);
909
910    use ustr::Ustr;
911
912    use crate::http::models::BybitFeeRate;
913
914    fn sample_fee_rate(
915        symbol: &str,
916        taker: &str,
917        maker: &str,
918        base_coin: Option<&str>,
919    ) -> BybitFeeRate {
920        BybitFeeRate {
921            symbol: Ustr::from(symbol),
922            taker_fee_rate: taker.to_string(),
923            maker_fee_rate: maker.to_string(),
924            base_coin: base_coin.map(Ustr::from),
925        }
926    }
927
928    fn linear_instrument() -> InstrumentAny {
929        let json = load_test_json("http_get_instruments_linear.json");
930        let response: BybitInstrumentLinearResponse = serde_json::from_str(&json).unwrap();
931        let instrument = &response.result.list[0];
932        let fee_rate = sample_fee_rate("BTCUSDT", "0.00055", "0.0001", Some("BTC"));
933        parse_linear_instrument(instrument, &fee_rate, TS, TS).unwrap()
934    }
935
936    fn option_instrument() -> InstrumentAny {
937        let json = load_test_json("http_get_instruments_option.json");
938        let response: BybitInstrumentOptionResponse = serde_json::from_str(&json).unwrap();
939        let instrument = &response.result.list[0];
940        parse_option_instrument(instrument, TS, TS).unwrap()
941    }
942
943    #[rstest]
944    fn parse_ws_trade_into_trade_tick() {
945        let instrument = linear_instrument();
946        let json = load_test_json("ws_public_trade.json");
947        let msg: BybitWsTradeMsg = serde_json::from_str(&json).unwrap();
948        let trade = &msg.data[0];
949
950        let tick = parse_ws_trade_tick(trade, &instrument, TS).unwrap();
951
952        assert_eq!(tick.instrument_id, instrument.id());
953        assert_eq!(tick.price, instrument.make_price(27451.00));
954        assert_eq!(tick.size, instrument.make_qty(0.010, None));
955        assert_eq!(tick.aggressor_side, AggressorSide::Buyer);
956        assert_eq!(
957            tick.trade_id.to_string(),
958            "9dc75fca-4bdd-4773-9f78-6f5d7ab2a110"
959        );
960        assert_eq!(tick.ts_event, UnixNanos::new(1_709_891_679_000_000_000));
961    }
962
963    #[rstest]
964    fn parse_orderbook_snapshot_into_deltas() {
965        let instrument = linear_instrument();
966        let json = load_test_json("ws_orderbook_snapshot.json");
967        let msg: BybitWsOrderbookDepthMsg = serde_json::from_str(&json).unwrap();
968
969        let deltas = parse_orderbook_deltas(&msg, &instrument, TS).unwrap();
970
971        assert_eq!(deltas.instrument_id, instrument.id());
972        assert_eq!(deltas.deltas.len(), 5);
973        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
974        assert_eq!(
975            deltas.deltas[1].order.price,
976            instrument.make_price(27450.00)
977        );
978        assert_eq!(
979            deltas.deltas[1].order.size,
980            instrument.make_qty(0.500, None)
981        );
982        let last = deltas.deltas.last().unwrap();
983        assert_eq!(last.order.side, OrderSide::Sell);
984        assert_eq!(last.order.price, instrument.make_price(27451.50));
985        assert_eq!(
986            last.flags & RecordFlag::F_LAST as u8,
987            RecordFlag::F_LAST as u8
988        );
989    }
990
991    #[rstest]
992    fn parse_orderbook_delta_marks_actions() {
993        let instrument = linear_instrument();
994        let json = load_test_json("ws_orderbook_delta.json");
995        let msg: BybitWsOrderbookDepthMsg = serde_json::from_str(&json).unwrap();
996
997        let deltas = parse_orderbook_deltas(&msg, &instrument, TS).unwrap();
998
999        assert_eq!(deltas.deltas.len(), 2);
1000        let bid = &deltas.deltas[0];
1001        assert_eq!(bid.action, BookAction::Update);
1002        assert_eq!(bid.order.side, OrderSide::Buy);
1003        assert_eq!(bid.order.size, instrument.make_qty(0.400, None));
1004
1005        let ask = &deltas.deltas[1];
1006        assert_eq!(ask.action, BookAction::Delete);
1007        assert_eq!(ask.order.side, OrderSide::Sell);
1008        assert_eq!(ask.order.size, instrument.make_qty(0.0, None));
1009        assert_eq!(
1010            ask.flags & RecordFlag::F_LAST as u8,
1011            RecordFlag::F_LAST as u8
1012        );
1013    }
1014
1015    #[rstest]
1016    fn parse_orderbook_quote_produces_top_of_book() {
1017        let instrument = linear_instrument();
1018        let json = load_test_json("ws_orderbook_snapshot.json");
1019        let msg: BybitWsOrderbookDepthMsg = serde_json::from_str(&json).unwrap();
1020
1021        let quote = parse_orderbook_quote(&msg, &instrument, None, TS).unwrap();
1022
1023        assert_eq!(quote.instrument_id, instrument.id());
1024        assert_eq!(quote.bid_price, instrument.make_price(27450.00));
1025        assert_eq!(quote.bid_size, instrument.make_qty(0.500, None));
1026        assert_eq!(quote.ask_price, instrument.make_price(27451.00));
1027        assert_eq!(quote.ask_size, instrument.make_qty(0.750, None));
1028    }
1029
1030    #[rstest]
1031    fn parse_orderbook_quote_with_delta_updates_sizes() {
1032        let instrument = linear_instrument();
1033        let snapshot: BybitWsOrderbookDepthMsg =
1034            serde_json::from_str(&load_test_json("ws_orderbook_snapshot.json")).unwrap();
1035        let base_quote = parse_orderbook_quote(&snapshot, &instrument, None, TS).unwrap();
1036
1037        let delta: BybitWsOrderbookDepthMsg =
1038            serde_json::from_str(&load_test_json("ws_orderbook_delta.json")).unwrap();
1039        let updated = parse_orderbook_quote(&delta, &instrument, Some(&base_quote), TS).unwrap();
1040
1041        assert_eq!(updated.bid_price, instrument.make_price(27450.00));
1042        assert_eq!(updated.bid_size, instrument.make_qty(0.400, None));
1043        assert_eq!(updated.ask_price, instrument.make_price(27451.00));
1044        assert_eq!(updated.ask_size, instrument.make_qty(0.0, None));
1045    }
1046
1047    #[rstest]
1048    fn parse_linear_ticker_quote_to_quote_tick() {
1049        let instrument = linear_instrument();
1050        let json = load_test_json("ws_ticker_linear.json");
1051        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
1052
1053        let quote = parse_ticker_linear_quote(&msg, &instrument, TS).unwrap();
1054
1055        assert_eq!(quote.instrument_id, instrument.id());
1056        assert_eq!(quote.bid_price, instrument.make_price(17215.50));
1057        assert_eq!(quote.ask_price, instrument.make_price(17216.00));
1058        assert_eq!(quote.bid_size, instrument.make_qty(84.489, None));
1059        assert_eq!(quote.ask_size, instrument.make_qty(83.020, None));
1060        assert_eq!(quote.ts_event, UnixNanos::new(1_673_272_861_686_000_000));
1061        assert_eq!(quote.ts_init, TS);
1062    }
1063
1064    #[rstest]
1065    fn parse_option_ticker_quote_to_quote_tick() {
1066        let instrument = option_instrument();
1067        let json = load_test_json("ws_ticker_option.json");
1068        let msg: BybitWsTickerOptionMsg = serde_json::from_str(&json).unwrap();
1069
1070        let quote = parse_ticker_option_quote(&msg, &instrument, TS).unwrap();
1071
1072        assert_eq!(quote.instrument_id, instrument.id());
1073        assert_eq!(quote.bid_price, instrument.make_price(0.0));
1074        assert_eq!(quote.ask_price, instrument.make_price(10.0));
1075        assert_eq!(quote.bid_size, instrument.make_qty(0.0, None));
1076        assert_eq!(quote.ask_size, instrument.make_qty(5.1, None));
1077        assert_eq!(quote.ts_event, UnixNanos::new(1_672_917_511_074_000_000));
1078        assert_eq!(quote.ts_init, TS);
1079    }
1080
1081    #[rstest]
1082    #[case::timestamp_on_open(false, 1_672_324_800_000_000_000)]
1083    #[case::timestamp_on_close(true, 1_672_325_100_000_000_000)]
1084    fn parse_ws_kline_into_bar(#[case] timestamp_on_close: bool, #[case] expected_ts_event: u64) {
1085        use std::num::NonZero;
1086
1087        let instrument = linear_instrument();
1088        let json = load_test_json("ws_kline.json");
1089        let msg: crate::websocket::messages::BybitWsKlineMsg = serde_json::from_str(&json).unwrap();
1090        let kline = &msg.data[0];
1091
1092        let bar_spec = BarSpecification {
1093            step: NonZero::new(5).unwrap(),
1094            aggregation: BarAggregation::Minute,
1095            price_type: PriceType::Last,
1096        };
1097        let bar_type = BarType::new(instrument.id(), bar_spec, AggregationSource::External);
1098
1099        let bar = parse_ws_kline_bar(kline, &instrument, bar_type, timestamp_on_close, TS).unwrap();
1100
1101        assert_eq!(bar.bar_type, bar_type);
1102        assert_eq!(bar.open, instrument.make_price(16649.5));
1103        assert_eq!(bar.high, instrument.make_price(16677.0));
1104        assert_eq!(bar.low, instrument.make_price(16608.0));
1105        assert_eq!(bar.close, instrument.make_price(16677.0));
1106        assert_eq!(bar.volume, instrument.make_qty(2.081, None));
1107        assert_eq!(bar.ts_event, UnixNanos::new(expected_ts_event));
1108        assert_eq!(bar.ts_init, TS);
1109    }
1110
1111    #[rstest]
1112    fn parse_ws_order_into_order_status_report() {
1113        let instrument = linear_instrument();
1114        let json = load_test_json("ws_account_order_filled.json");
1115        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1116            serde_json::from_str(&json).unwrap();
1117        let order = &msg.data[0];
1118        let account_id = AccountId::new("BYBIT-001");
1119
1120        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
1121
1122        assert_eq!(report.account_id, account_id);
1123        assert_eq!(report.instrument_id, instrument.id());
1124        assert_eq!(report.order_side, OrderSide::Buy);
1125        assert_eq!(report.order_type, OrderType::Limit);
1126        assert_eq!(report.time_in_force, TimeInForce::Gtc);
1127        assert_eq!(report.order_status, OrderStatus::Filled);
1128        assert_eq!(report.quantity, instrument.make_qty(0.100, None));
1129        assert_eq!(report.filled_qty, instrument.make_qty(0.100, None));
1130        assert_eq!(report.price, Some(instrument.make_price(30000.50)));
1131        assert_eq!(report.avg_px, Some(dec!(30000.50)));
1132        assert_eq!(
1133            report.client_order_id.as_ref().unwrap().to_string(),
1134            "test-client-order-001"
1135        );
1136        assert_eq!(
1137            report.ts_accepted,
1138            UnixNanos::new(1_672_364_262_444_000_000)
1139        );
1140        assert_eq!(report.ts_last, UnixNanos::new(1_672_364_262_457_000_000));
1141    }
1142
1143    #[rstest]
1144    fn parse_ws_order_partially_filled_rejected_maps_to_canceled() {
1145        let instrument = linear_instrument();
1146        let json = load_test_json("ws_account_order_partially_filled_rejected.json");
1147        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1148            serde_json::from_str(&json).unwrap();
1149        let order = &msg.data[0];
1150        let account_id = AccountId::new("BYBIT-001");
1151
1152        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
1153
1154        // Verify that Bybit "Rejected" status with fills is mapped to Canceled, not Rejected
1155        assert_eq!(report.order_status, OrderStatus::Canceled);
1156        assert_eq!(report.filled_qty, instrument.make_qty(50.0, None));
1157        assert_eq!(
1158            report.client_order_id.as_ref().unwrap().to_string(),
1159            "O-20251001-164609-APEX-000-49"
1160        );
1161        assert_eq!(report.cancel_reason, Some("UNKNOWN".to_string()));
1162    }
1163
1164    #[rstest]
1165    fn parse_ws_execution_into_fill_report() {
1166        let instrument = linear_instrument();
1167        let json = load_test_json("ws_account_execution.json");
1168        let msg: crate::websocket::messages::BybitWsAccountExecutionMsg =
1169            serde_json::from_str(&json).unwrap();
1170        let execution = &msg.data[0];
1171        let account_id = AccountId::new("BYBIT-001");
1172
1173        let report = parse_ws_fill_report(execution, account_id, &instrument, TS).unwrap();
1174
1175        assert_eq!(report.account_id, account_id);
1176        assert_eq!(report.instrument_id, instrument.id());
1177        assert_eq!(
1178            report.venue_order_id.to_string(),
1179            "9aac161b-8ed6-450d-9cab-c5cc67c21784"
1180        );
1181        assert_eq!(
1182            report.trade_id.to_string(),
1183            "0ab1bdf7-4219-438b-b30a-32ec863018f7"
1184        );
1185        assert_eq!(report.order_side, OrderSide::Sell);
1186        assert_eq!(report.last_qty, instrument.make_qty(0.5, None));
1187        assert_eq!(report.last_px, instrument.make_price(95900.1));
1188        assert_eq!(report.commission.as_f64(), 26.3725275);
1189        assert_eq!(report.liquidity_side, LiquiditySide::Taker);
1190        assert_eq!(
1191            report.client_order_id.as_ref().unwrap().to_string(),
1192            "test-order-link-001"
1193        );
1194        assert_eq!(report.ts_event, UnixNanos::new(1_746_270_400_353_000_000));
1195    }
1196
1197    #[rstest]
1198    fn parse_ws_position_into_position_status_report() {
1199        let instrument = linear_instrument();
1200        let json = load_test_json("ws_account_position.json");
1201        let msg: crate::websocket::messages::BybitWsAccountPositionMsg =
1202            serde_json::from_str(&json).unwrap();
1203        let position = &msg.data[0];
1204        let account_id = AccountId::new("BYBIT-001");
1205
1206        let report =
1207            parse_ws_position_status_report(position, account_id, &instrument, TS).unwrap();
1208
1209        assert_eq!(report.account_id, account_id);
1210        assert_eq!(report.instrument_id, instrument.id());
1211        assert_eq!(report.position_side.as_position_side(), PositionSide::Short);
1212        assert_eq!(report.quantity, instrument.make_qty(0.01, None));
1213        assert_eq!(
1214            report.avg_px_open,
1215            Some(Decimal::try_from(3641.075).unwrap())
1216        );
1217        assert_eq!(report.ts_last, UnixNanos::new(1_762_199_125_472_000_000));
1218        assert_eq!(report.ts_init, TS);
1219    }
1220
1221    #[rstest]
1222    fn parse_ws_position_short_into_position_status_report() {
1223        // Create ETHUSDT instrument
1224        let instruments_json = load_test_json("http_get_instruments_linear.json");
1225        let instruments_response: crate::http::models::BybitInstrumentLinearResponse =
1226            serde_json::from_str(&instruments_json).unwrap();
1227        let eth_def = &instruments_response.result.list[1]; // ETHUSDT is second in the list
1228        let fee_rate = crate::http::models::BybitFeeRate {
1229            symbol: Ustr::from("ETHUSDT"),
1230            taker_fee_rate: "0.00055".to_string(),
1231            maker_fee_rate: "0.0001".to_string(),
1232            base_coin: Some(Ustr::from("ETH")),
1233        };
1234        let instrument =
1235            crate::common::parse::parse_linear_instrument(eth_def, &fee_rate, TS, TS).unwrap();
1236
1237        let json = load_test_json("ws_account_position_short.json");
1238        let msg: crate::websocket::messages::BybitWsAccountPositionMsg =
1239            serde_json::from_str(&json).unwrap();
1240        let position = &msg.data[0];
1241        let account_id = AccountId::new("BYBIT-001");
1242
1243        let report =
1244            parse_ws_position_status_report(position, account_id, &instrument, TS).unwrap();
1245
1246        assert_eq!(report.account_id, account_id);
1247        assert_eq!(report.instrument_id.symbol.as_str(), "ETHUSDT-LINEAR");
1248        assert_eq!(report.position_side.as_position_side(), PositionSide::Short);
1249        assert_eq!(report.quantity, instrument.make_qty(0.01, None));
1250        assert_eq!(
1251            report.avg_px_open,
1252            Some(Decimal::try_from(3641.075).unwrap())
1253        );
1254        assert_eq!(report.ts_last, UnixNanos::new(1_762_199_125_472_000_000));
1255        assert_eq!(report.ts_init, TS);
1256    }
1257
1258    #[rstest]
1259    fn parse_ws_wallet_into_account_state() {
1260        let json = load_test_json("ws_account_wallet.json");
1261        let msg: crate::websocket::messages::BybitWsAccountWalletMsg =
1262            serde_json::from_str(&json).unwrap();
1263        let wallet = &msg.data[0];
1264        let account_id = AccountId::new("BYBIT-001");
1265        let ts_event = UnixNanos::new(1_700_034_722_104_000_000);
1266
1267        let state = parse_ws_account_state(wallet, account_id, ts_event, TS).unwrap();
1268
1269        assert_eq!(state.account_id, account_id);
1270        assert_eq!(state.account_type, AccountType::Margin);
1271        assert_eq!(state.balances.len(), 2);
1272        assert!(state.is_reported);
1273
1274        // Check BTC balance
1275        let btc_balance = &state.balances[0];
1276        assert_eq!(btc_balance.currency.code.as_str(), "BTC");
1277        assert!((btc_balance.total.as_f64() - 0.00102964).abs() < 1e-8);
1278        assert!((btc_balance.free.as_f64() - 0.00092964).abs() < 1e-8);
1279        assert!((btc_balance.locked.as_f64() - 0.0001).abs() < 1e-8);
1280
1281        // Check USDT balance
1282        let usdt_balance = &state.balances[1];
1283        assert_eq!(usdt_balance.currency.code.as_str(), "USDT");
1284        assert!((usdt_balance.total.as_f64() - 9647.75537647).abs() < 1e-6);
1285        assert!((usdt_balance.free.as_f64() - 9519.89806037).abs() < 1e-6);
1286        assert!((usdt_balance.locked.as_f64() - 127.8573161).abs() < 1e-6);
1287
1288        assert_eq!(state.ts_event, ts_event);
1289        assert_eq!(state.ts_init, TS);
1290    }
1291
1292    #[rstest]
1293    fn parse_ws_wallet_with_small_order_calculates_free_correctly() {
1294        // Regression test for issue where availableToWithdraw=0 caused all funds to appear locked
1295        // When a small order is placed, Bybit may report availableToWithdraw=0 due to margin calculations,
1296        // but totalOrderIM correctly shows only the margin locked for the order
1297        let json = load_test_json("ws_account_wallet_small_order.json");
1298        let msg: crate::websocket::messages::BybitWsAccountWalletMsg =
1299            serde_json::from_str(&json).unwrap();
1300        let wallet = &msg.data[0];
1301        let account_id = AccountId::new("BYBIT-UNIFIED");
1302        let ts_event = UnixNanos::new(1_762_960_669_000_000_000);
1303
1304        let state = parse_ws_account_state(wallet, account_id, ts_event, TS).unwrap();
1305
1306        assert_eq!(state.account_id, account_id);
1307        assert_eq!(state.balances.len(), 1);
1308
1309        // Check USDT balance
1310        let usdt_balance = &state.balances[0];
1311        assert_eq!(usdt_balance.currency.code.as_str(), "USDT");
1312
1313        // Wallet has 51,333.82 USDT total
1314        assert!((usdt_balance.total.as_f64() - 51333.82543837).abs() < 1e-6);
1315
1316        // Only 50.028 USDT should be locked (for the order), not all funds
1317        assert!((usdt_balance.locked.as_f64() - 50.028).abs() < 1e-6);
1318
1319        // Free should be total - locked = 51,333.82 - 50.028 = 51,283.79
1320        assert!((usdt_balance.free.as_f64() - 51283.79743837).abs() < 1e-6);
1321
1322        // The bug would have calculated: locked = total - availableToWithdraw = 51,333.82 - 0 = 51,333.82 (all locked!)
1323        // This test verifies that we now correctly use totalOrderIM instead of deriving from availableToWithdraw
1324    }
1325
1326    #[rstest]
1327    fn parse_ticker_linear_into_funding_rate() {
1328        let instrument = linear_instrument();
1329        let json = load_test_json("ws_ticker_linear.json");
1330        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
1331
1332        let ts_event = UnixNanos::new(1_673_272_861_686_000_000);
1333
1334        let funding =
1335            parse_ticker_linear_funding(&msg.data, instrument.id(), ts_event, TS).unwrap();
1336
1337        assert_eq!(funding.instrument_id, instrument.id());
1338        assert_eq!(funding.rate, dec!(-0.000212)); // -0.000212
1339        assert_eq!(
1340            funding.next_funding_ns,
1341            Some(UnixNanos::new(1_673_280_000_000_000_000))
1342        );
1343        assert_eq!(funding.ts_event, ts_event);
1344        assert_eq!(funding.ts_init, TS);
1345    }
1346
1347    #[rstest]
1348    fn parse_ticker_linear_into_mark_price() {
1349        let instrument = linear_instrument();
1350        let json = load_test_json("ws_ticker_linear.json");
1351        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
1352
1353        let ts_event = UnixNanos::new(1_673_272_861_686_000_000);
1354
1355        let mark_price =
1356            parse_ticker_linear_mark_price(&msg.data, &instrument, ts_event, TS).unwrap();
1357
1358        assert_eq!(mark_price.instrument_id, instrument.id());
1359        assert_eq!(mark_price.value, instrument.make_price(17217.33));
1360        assert_eq!(mark_price.ts_event, ts_event);
1361        assert_eq!(mark_price.ts_init, TS);
1362    }
1363
1364    #[rstest]
1365    fn parse_ticker_linear_into_index_price() {
1366        let instrument = linear_instrument();
1367        let json = load_test_json("ws_ticker_linear.json");
1368        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
1369
1370        let ts_event = UnixNanos::new(1_673_272_861_686_000_000);
1371
1372        let index_price =
1373            parse_ticker_linear_index_price(&msg.data, &instrument, ts_event, TS).unwrap();
1374
1375        assert_eq!(index_price.instrument_id, instrument.id());
1376        assert_eq!(index_price.value, instrument.make_price(17227.36));
1377        assert_eq!(index_price.ts_event, ts_event);
1378        assert_eq!(index_price.ts_init, TS);
1379    }
1380
1381    #[rstest]
1382    fn parse_ticker_option_into_mark_price() {
1383        let instrument = option_instrument();
1384        let json = load_test_json("ws_ticker_option.json");
1385        let msg: BybitWsTickerOptionMsg = serde_json::from_str(&json).unwrap();
1386
1387        let mark_price = parse_ticker_option_mark_price(&msg, &instrument, TS).unwrap();
1388
1389        assert_eq!(mark_price.instrument_id, instrument.id());
1390        assert_eq!(mark_price.value, instrument.make_price(7.86976724));
1391        assert_eq!(mark_price.ts_init, TS);
1392    }
1393
1394    #[rstest]
1395    fn parse_ticker_option_into_index_price() {
1396        let instrument = option_instrument();
1397        let json = load_test_json("ws_ticker_option.json");
1398        let msg: BybitWsTickerOptionMsg = serde_json::from_str(&json).unwrap();
1399
1400        let index_price = parse_ticker_option_index_price(&msg, &instrument, TS).unwrap();
1401
1402        assert_eq!(index_price.instrument_id, instrument.id());
1403        assert_eq!(index_price.value, instrument.make_price(16823.73));
1404        assert_eq!(index_price.ts_init, TS);
1405    }
1406
1407    #[rstest]
1408    fn parse_ws_order_stop_market_sell_preserves_type() {
1409        let instrument = linear_instrument();
1410        let json = load_test_json("ws_account_order_stop_market.json");
1411        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1412            serde_json::from_str(&json).unwrap();
1413        let order = &msg.data[0];
1414        let account_id = AccountId::new("BYBIT-001");
1415
1416        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
1417
1418        // Verify sell StopMarket: orderType=Market + stopOrderType=Stop + triggerDirection=2 (falls to)
1419        assert_eq!(report.order_type, OrderType::StopMarket);
1420        assert_eq!(report.order_side, OrderSide::Sell);
1421        assert_eq!(report.order_status, OrderStatus::Accepted); // Untriggered maps to Accepted
1422        assert_eq!(report.trigger_price, Some(instrument.make_price(45000.00)));
1423        assert_eq!(report.trigger_type, Some(TriggerType::LastPrice));
1424        assert_eq!(
1425            report.client_order_id.as_ref().unwrap().to_string(),
1426            "test-client-stop-market-001"
1427        );
1428    }
1429
1430    #[rstest]
1431    fn parse_ws_order_stop_market_buy_preserves_type() {
1432        let instrument = linear_instrument();
1433        let json = load_test_json("ws_account_order_buy_stop_market.json");
1434        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1435            serde_json::from_str(&json).unwrap();
1436        let order = &msg.data[0];
1437        let account_id = AccountId::new("BYBIT-001");
1438
1439        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
1440
1441        // Verify buy StopMarket: orderType=Market + stopOrderType=Stop + triggerDirection=1 (rises to)
1442        assert_eq!(report.order_type, OrderType::StopMarket);
1443        assert_eq!(report.order_side, OrderSide::Buy);
1444        assert_eq!(report.order_status, OrderStatus::Accepted);
1445        assert_eq!(report.trigger_price, Some(instrument.make_price(55000.00)));
1446        assert_eq!(report.trigger_type, Some(TriggerType::LastPrice));
1447        assert_eq!(
1448            report.client_order_id.as_ref().unwrap().to_string(),
1449            "test-client-buy-stop-market-001"
1450        );
1451    }
1452
1453    #[rstest]
1454    fn parse_ws_order_market_if_touched_buy_preserves_type() {
1455        let instrument = linear_instrument();
1456        let json = load_test_json("ws_account_order_market_if_touched.json");
1457        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1458            serde_json::from_str(&json).unwrap();
1459        let order = &msg.data[0];
1460        let account_id = AccountId::new("BYBIT-001");
1461
1462        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
1463
1464        // Verify buy MIT: orderType=Market + stopOrderType=Stop + triggerDirection=2 (falls to)
1465        assert_eq!(report.order_type, OrderType::MarketIfTouched);
1466        assert_eq!(report.order_side, OrderSide::Buy);
1467        assert_eq!(report.order_status, OrderStatus::Accepted); // Untriggered maps to Accepted
1468        assert_eq!(report.trigger_price, Some(instrument.make_price(55000.00)));
1469        assert_eq!(report.trigger_type, Some(TriggerType::LastPrice));
1470        assert_eq!(
1471            report.client_order_id.as_ref().unwrap().to_string(),
1472            "test-client-mit-001"
1473        );
1474    }
1475
1476    #[rstest]
1477    fn parse_ws_order_market_if_touched_sell_preserves_type() {
1478        let instrument = linear_instrument();
1479        let json = load_test_json("ws_account_order_sell_market_if_touched.json");
1480        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1481            serde_json::from_str(&json).unwrap();
1482        let order = &msg.data[0];
1483        let account_id = AccountId::new("BYBIT-001");
1484
1485        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
1486
1487        // Verify sell MIT: orderType=Market + stopOrderType=Stop + triggerDirection=1 (rises to)
1488        assert_eq!(report.order_type, OrderType::MarketIfTouched);
1489        assert_eq!(report.order_side, OrderSide::Sell);
1490        assert_eq!(report.order_status, OrderStatus::Accepted);
1491        assert_eq!(report.trigger_price, Some(instrument.make_price(55000.00)));
1492        assert_eq!(
1493            report.client_order_id.as_ref().unwrap().to_string(),
1494            "test-client-sell-mit-001"
1495        );
1496    }
1497
1498    #[rstest]
1499    fn parse_ws_order_stop_limit_preserves_type() {
1500        let instrument = linear_instrument();
1501        let json = load_test_json("ws_account_order_stop_limit.json");
1502        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1503            serde_json::from_str(&json).unwrap();
1504        let order = &msg.data[0];
1505        let account_id = AccountId::new("BYBIT-001");
1506
1507        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
1508
1509        // Verify StopLimit order type is correctly parsed
1510        // orderType=Limit + stopOrderType=Stop + triggerDirection=2 (falls to)
1511        assert_eq!(report.order_type, OrderType::StopLimit);
1512        assert_eq!(report.order_side, OrderSide::Sell);
1513        assert_eq!(report.order_status, OrderStatus::Accepted); // Untriggered maps to Accepted
1514        assert_eq!(report.price, Some(instrument.make_price(44500.00)));
1515        assert_eq!(report.trigger_price, Some(instrument.make_price(45000.00)));
1516        assert_eq!(
1517            report.client_order_id.as_ref().unwrap().to_string(),
1518            "test-client-stop-limit-001"
1519        );
1520    }
1521
1522    #[rstest]
1523    fn parse_ws_order_limit_if_touched_preserves_type() {
1524        let instrument = linear_instrument();
1525        let json = load_test_json("ws_account_order_limit_if_touched.json");
1526        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1527            serde_json::from_str(&json).unwrap();
1528        let order = &msg.data[0];
1529        let account_id = AccountId::new("BYBIT-001");
1530
1531        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
1532
1533        // Verify LimitIfTouched order type is correctly parsed
1534        // orderType=Limit + stopOrderType=Stop + triggerDirection=1 (rises to)
1535        assert_eq!(report.order_type, OrderType::LimitIfTouched);
1536        assert_eq!(report.order_side, OrderSide::Buy);
1537        assert_eq!(report.order_status, OrderStatus::Accepted); // Untriggered maps to Accepted
1538        assert_eq!(report.price, Some(instrument.make_price(55500.00)));
1539        assert_eq!(report.trigger_price, Some(instrument.make_price(55000.00)));
1540        assert_eq!(
1541            report.client_order_id.as_ref().unwrap().to_string(),
1542            "test-client-lit-001"
1543        );
1544    }
1545}
nautilus_bybit/websocket/parse.rs

nautilus_bybit/websocket/
parse.rs
