Skip to main content

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//! Parsers 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, greeks::OptionGreekValues,
26        option_chain::OptionGreeks,
27    },
28    enums::{
29        AccountType, AggressorSide, BookAction, GreeksConvention, LiquiditySide, OrderSide,
30        OrderStatus, PositionSide, RecordFlag, TimeInForce, TriggerType,
31    },
32    events::account::state::AccountState,
33    identifiers::{AccountId, ClientOrderId, InstrumentId, PositionId, TradeId, VenueOrderId},
34    instruments::{Instrument, any::InstrumentAny},
35    reports::{FillReport, OrderStatusReport, PositionStatusReport},
36    types::{AccountBalance, MarginBalance, Money},
37};
38use rust_decimal::Decimal;
39
40use super::{
41    enums::{BybitWsOperation, BybitWsPrivateChannel, BybitWsPublicChannel},
42    messages::{
43        BybitWsAccountExecution, BybitWsAccountExecutionFast, BybitWsAccountOrder,
44        BybitWsAccountPosition, BybitWsAccountWallet, BybitWsAuthResponse, BybitWsFrame,
45        BybitWsKline, BybitWsOrderResponse, BybitWsOrderbookDepthMsg, BybitWsResponse,
46        BybitWsSubscriptionMsg, BybitWsTickerLinear, BybitWsTickerOptionMsg, BybitWsTrade,
47    },
48};
49use crate::common::{
50    consts::BYBIT_QUOTE_DEPTH,
51    enums::{BybitOrderStatus, BybitPositionSide, BybitTimeInForce},
52    parse::{
53        bybit_rejection_due_post_only, get_currency, make_hedge_venue_position_id,
54        parse_book_level, parse_bybit_order_type, parse_millis_timestamp,
55        parse_price_with_precision, parse_quantity_with_precision,
56    },
57};
58
59/// Classifies a parsed JSON value into a typed Bybit WebSocket frame.
60///
61/// Returns `Unknown(value)` if no specific type matches.
62pub fn parse_bybit_ws_frame(value: serde_json::Value) -> BybitWsFrame {
63    if let Some(op_val) = value.get("op") {
64        if let Ok(op) = serde_json::from_value::<BybitWsOperation>(op_val.clone())
65            && op == BybitWsOperation::Auth
66            && let Ok(auth) = serde_json::from_value::<BybitWsAuthResponse>(value.clone())
67        {
68            let is_success = auth.success.unwrap_or(false) || auth.ret_code.unwrap_or(-1) == 0;
69            if is_success {
70                return BybitWsFrame::Auth(auth);
71            }
72            let resp = BybitWsResponse {
73                op: Some(auth.op.clone()),
74                topic: None,
75                success: auth.success,
76                conn_id: auth.conn_id.clone(),
77                req_id: None,
78                ret_code: auth.ret_code,
79                ret_msg: auth.ret_msg,
80            };
81            return BybitWsFrame::ErrorResponse(resp);
82        }
83
84        if let Some(op_str) = op_val.as_str()
85            && op_str.starts_with("order.")
86        {
87            return serde_json::from_value::<BybitWsOrderResponse>(value.clone()).map_or_else(
88                |_| BybitWsFrame::Unknown(value),
89                BybitWsFrame::OrderResponse,
90            );
91        }
92    }
93
94    if let Some(success) = value.get("success").and_then(serde_json::Value::as_bool) {
95        if success {
96            return serde_json::from_value::<BybitWsSubscriptionMsg>(value.clone())
97                .map_or_else(|_| BybitWsFrame::Unknown(value), BybitWsFrame::Subscription);
98        }
99        return serde_json::from_value::<BybitWsResponse>(value.clone()).map_or_else(
100            |_| BybitWsFrame::Unknown(value),
101            BybitWsFrame::ErrorResponse,
102        );
103    }
104
105    if let Some(topic) = value.get("topic").and_then(serde_json::Value::as_str) {
106        if topic.starts_with(BybitWsPublicChannel::OrderBook.as_ref()) {
107            return serde_json::from_value(value.clone())
108                .map_or_else(|_| BybitWsFrame::Unknown(value), BybitWsFrame::Orderbook);
109        }
110
111        if topic.contains(BybitWsPublicChannel::PublicTrade.as_ref())
112            || topic.starts_with(BybitWsPublicChannel::Trade.as_ref())
113        {
114            return serde_json::from_value(value.clone())
115                .map_or_else(|_| BybitWsFrame::Unknown(value), BybitWsFrame::Trade);
116        }
117
118        if topic.starts_with(BybitWsPublicChannel::Kline.as_ref()) {
119            return serde_json::from_value(value.clone())
120                .map_or_else(|_| BybitWsFrame::Unknown(value), BybitWsFrame::Kline);
121        }
122
123        if topic.starts_with(BybitWsPublicChannel::Tickers.as_ref()) {
124            // Option symbols have 3+ hyphens: BTC-6JAN23-17500-C
125            let is_option = value
126                .get("data")
127                .and_then(|d| d.get("symbol"))
128                .and_then(|s| s.as_str())
129                .is_some_and(|symbol| symbol.contains('-') && symbol.matches('-').count() >= 3);
130
131            if is_option {
132                return serde_json::from_value(value.clone())
133                    .map_or_else(|_| BybitWsFrame::Unknown(value), BybitWsFrame::TickerOption);
134            }
135            return serde_json::from_value(value.clone())
136                .map_or_else(|_| BybitWsFrame::Unknown(value), BybitWsFrame::TickerLinear);
137        }
138
139        if topic.starts_with(BybitWsPrivateChannel::Order.as_ref()) {
140            return serde_json::from_value(value.clone())
141                .map_or_else(|_| BybitWsFrame::Unknown(value), BybitWsFrame::AccountOrder);
142        }
143
144        if topic.starts_with(BybitWsPrivateChannel::ExecutionFast.as_ref()) {
145            return serde_json::from_value(value.clone()).map_or_else(
146                |_| BybitWsFrame::Unknown(value),
147                BybitWsFrame::AccountExecutionFast,
148            );
149        }
150
151        if topic.starts_with(BybitWsPrivateChannel::Execution.as_ref()) {
152            return serde_json::from_value(value.clone()).map_or_else(
153                |_| BybitWsFrame::Unknown(value),
154                BybitWsFrame::AccountExecution,
155            );
156        }
157
158        if topic.starts_with(BybitWsPrivateChannel::Wallet.as_ref()) {
159            return serde_json::from_value(value.clone()).map_or_else(
160                |_| BybitWsFrame::Unknown(value),
161                BybitWsFrame::AccountWallet,
162            );
163        }
164
165        if topic.starts_with(BybitWsPrivateChannel::Position.as_ref()) {
166            return serde_json::from_value(value.clone()).map_or_else(
167                |_| BybitWsFrame::Unknown(value),
168                BybitWsFrame::AccountPosition,
169            );
170        }
171    }
172
173    BybitWsFrame::Unknown(value)
174}
175
176/// Parses a Bybit WebSocket topic string into its components.
177///
178/// # Errors
179///
180/// Returns an error if the topic format is invalid.
181pub fn parse_topic(topic: &str) -> anyhow::Result<Vec<&str>> {
182    let parts: Vec<&str> = topic.split('.').collect();
183    if parts.is_empty() {
184        anyhow::bail!("Invalid topic format: empty topic");
185    }
186    Ok(parts)
187}
188
189/// Parses a Bybit kline topic into (interval, symbol).
190///
191/// Topic format: "kline.{interval}.{symbol}" (e.g., "kline.5.BTCUSDT")
192///
193/// # Errors
194///
195/// Returns an error if the topic format is invalid.
196pub fn parse_kline_topic(topic: &str) -> anyhow::Result<(&str, &str)> {
197    let kline = BybitWsPublicChannel::Kline.as_ref();
198    let parts = parse_topic(topic)?;
199    if parts.len() != 3 || parts[0] != kline {
200        anyhow::bail!(
201            "Invalid kline topic format: expected '{kline}.{{interval}}.{{symbol}}', was '{topic}'"
202        );
203    }
204    Ok((parts[1], parts[2]))
205}
206
207/// Parses a WebSocket trade frame into a [`TradeTick`].
208pub fn parse_ws_trade_tick(
209    trade: &BybitWsTrade,
210    instrument: &InstrumentAny,
211    ts_init: UnixNanos,
212) -> anyhow::Result<TradeTick> {
213    let price = parse_price_with_precision(&trade.p, instrument.price_precision(), "trade.p")?;
214    let size = parse_quantity_with_precision(&trade.v, instrument.size_precision(), "trade.v")?;
215    let aggressor: AggressorSide = trade.taker_side.into();
216    let trade_id = TradeId::new_checked(trade.i.as_str())
217        .context("invalid trade identifier in Bybit trade message")?;
218    let ts_event = parse_millis_i64(trade.t, "trade.T")?;
219
220    TradeTick::new_checked(
221        instrument.id(),
222        price,
223        size,
224        aggressor,
225        trade_id,
226        ts_event,
227        ts_init,
228    )
229    .context("failed to construct TradeTick from Bybit trade message")
230}
231
232/// Parses an order book depth message into [`OrderBookDeltas`].
233pub fn parse_orderbook_deltas(
234    msg: &BybitWsOrderbookDepthMsg,
235    instrument: &InstrumentAny,
236    ts_init: UnixNanos,
237) -> anyhow::Result<OrderBookDeltas> {
238    let is_snapshot = msg.msg_type.eq_ignore_ascii_case("snapshot");
239    let ts_event = parse_millis_i64(msg.ts, "orderbook.ts")?;
240    let ts_init = if ts_init.is_zero() { ts_event } else { ts_init };
241
242    let depth = &msg.data;
243    let instrument_id = instrument.id();
244    let price_precision = instrument.price_precision();
245    let size_precision = instrument.size_precision();
246    let update_id = u64::try_from(depth.u)
247        .context("received negative update id in Bybit order book message")?;
248    let sequence = u64::try_from(depth.seq)
249        .context("received negative sequence in Bybit order book message")?;
250
251    let total_levels = depth.b.len() + depth.a.len();
252    let capacity = if is_snapshot {
253        total_levels + 1
254    } else {
255        total_levels
256    };
257    let mut deltas = Vec::with_capacity(capacity);
258
259    if is_snapshot {
260        deltas.push(OrderBookDelta::clear(
261            instrument_id,
262            sequence,
263            ts_event,
264            ts_init,
265        ));
266    }
267    let mut processed = 0_usize;
268
269    let mut push_level = |values: &[String], side: OrderSide| -> anyhow::Result<()> {
270        let (price, size) = parse_book_level(values, price_precision, size_precision, "orderbook")?;
271        let action = if size.is_zero() {
272            BookAction::Delete
273        } else if is_snapshot {
274            BookAction::Add
275        } else {
276            BookAction::Update
277        };
278
279        processed += 1;
280        let mut flags = RecordFlag::F_MBP as u8;
281
282        if processed == total_levels {
283            flags |= RecordFlag::F_LAST as u8;
284        }
285
286        let order = BookOrder::new(side, price, size, update_id);
287        let delta = OrderBookDelta::new_checked(
288            instrument_id,
289            action,
290            order,
291            flags,
292            sequence,
293            ts_event,
294            ts_init,
295        )
296        .context("failed to construct OrderBookDelta from Bybit book level")?;
297        deltas.push(delta);
298        Ok(())
299    };
300
301    for level in &depth.b {
302        push_level(level, OrderSide::Buy)?;
303    }
304
305    for level in &depth.a {
306        push_level(level, OrderSide::Sell)?;
307    }
308
309    if total_levels == 0
310        && let Some(last) = deltas.last_mut()
311    {
312        last.flags |= RecordFlag::F_LAST as u8;
313    }
314
315    OrderBookDeltas::new_checked(instrument_id, deltas)
316        .context("failed to assemble OrderBookDeltas from Bybit message")
317}
318
319/// Parses a depth-1 order book snapshot into a [`QuoteTick`].
320///
321/// # Errors
322///
323/// Returns an error for a different depth or message type, missing sides, or invalid price or size data.
324pub fn parse_orderbook_quote(
325    msg: &BybitWsOrderbookDepthMsg,
326    instrument: &InstrumentAny,
327    ts_init: UnixNanos,
328) -> anyhow::Result<QuoteTick> {
329    let (depth, _) = parse_orderbook_topic(msg.topic.as_str())?;
330    anyhow::ensure!(
331        depth == BYBIT_QUOTE_DEPTH && msg.msg_type == "snapshot",
332        "Expected depth-1 orderbook snapshot"
333    );
334    let ts_event = parse_millis_i64(msg.ts, "orderbook.ts")?;
335    let ts_init = if ts_init.is_zero() { ts_event } else { ts_init };
336    let price_precision = instrument.price_precision();
337    let size_precision = instrument.size_precision();
338    let bid = msg
339        .data
340        .b
341        .first()
342        .context("orderbook snapshot missing bid")?;
343    let ask = msg
344        .data
345        .a
346        .first()
347        .context("orderbook snapshot missing ask")?;
348    let (bid_price, bid_size) = parse_book_level(bid, price_precision, size_precision, "bid")?;
349    let (ask_price, ask_size) = parse_book_level(ask, price_precision, size_precision, "ask")?;
350
351    QuoteTick::new_checked(
352        instrument.id(),
353        bid_price,
354        ask_price,
355        bid_size,
356        ask_size,
357        ts_event,
358        ts_init,
359    )
360    .context("failed to construct QuoteTick from Bybit order book message")
361}
362
363pub(crate) fn parse_orderbook_topic(topic: &str) -> anyhow::Result<(u32, &str)> {
364    let mut parts = topic.splitn(3, '.');
365    anyhow::ensure!(
366        parts.next() == Some("orderbook"),
367        "Invalid orderbook topic: {topic}"
368    );
369    let depth = parts.next().context("missing orderbook depth")?.parse()?;
370    let symbol = parts
371        .next()
372        .filter(|s| !s.is_empty())
373        .context("missing orderbook symbol")?;
374    Ok((depth, symbol))
375}
376
377/// Parses an option ticker payload into a [`QuoteTick`].
378pub fn parse_ticker_option_quote(
379    msg: &BybitWsTickerOptionMsg,
380    instrument: &InstrumentAny,
381    ts_init: UnixNanos,
382) -> anyhow::Result<QuoteTick> {
383    let ts_event = parse_millis_i64(msg.ts, "ticker.ts")?;
384    let ts_init = if ts_init.is_zero() { ts_event } else { ts_init };
385    let price_precision = instrument.price_precision();
386    let size_precision = instrument.size_precision();
387
388    let data = &msg.data;
389    let bid_price =
390        parse_price_with_precision(&data.bid_price, price_precision, "ticker.bidPrice")?;
391    let ask_price =
392        parse_price_with_precision(&data.ask_price, price_precision, "ticker.askPrice")?;
393    let bid_size = parse_quantity_with_precision(&data.bid_size, size_precision, "ticker.bidSize")?;
394    let ask_size = parse_quantity_with_precision(&data.ask_size, size_precision, "ticker.askSize")?;
395
396    QuoteTick::new_checked(
397        instrument.id(),
398        bid_price,
399        ask_price,
400        bid_size,
401        ask_size,
402        ts_event,
403        ts_init,
404    )
405    .context("failed to construct QuoteTick from Bybit option ticker message")
406}
407
408/// Parses a linear ticker payload into a [`FundingRateUpdate`].
409///
410/// # Errors
411///
412/// Returns an error if funding rate, funding interval or next funding time fields are missing or cannot be parsed.
413pub fn parse_ticker_linear_funding(
414    data: &BybitWsTickerLinear,
415    instrument_id: InstrumentId,
416    ts_event: UnixNanos,
417    ts_init: UnixNanos,
418) -> anyhow::Result<FundingRateUpdate> {
419    let funding_rate_str = data
420        .funding_rate
421        .as_ref()
422        .context("Bybit ticker missing funding_rate")?;
423
424    if funding_rate_str.is_empty() {
425        anyhow::bail!(
426            "empty funding_rate for {instrument_id} (dated futures do not have funding rates)"
427        );
428    }
429
430    let funding_rate = funding_rate_str
431        .as_str()
432        .parse::<Decimal>()
433        .with_context(|| {
434            format!("invalid funding_rate value '{funding_rate_str}' for {instrument_id}")
435        })?;
436
437    let funding_interval = if let Some(funding_interval_hour) = &data.funding_interval_hour {
438        let funding_interval_hour = funding_interval_hour
439            .as_str()
440            .parse::<u16>()
441            .context("invalid funding_interval_hour value")?;
442        Some(
443            funding_interval_hour
444                .checked_mul(60)
445                .ok_or_else(|| anyhow::anyhow!("funding_interval_hour out of bounds"))?,
446        )
447    } else {
448        None
449    };
450
451    let next_funding_ns = if let Some(next_funding_time) = &data.next_funding_time {
452        let next_funding_millis = next_funding_time
453            .as_str()
454            .parse::<i64>()
455            .context("invalid next_funding_time value")?;
456        Some(parse_millis_i64(next_funding_millis, "next_funding_time")?)
457    } else {
458        None
459    };
460
461    Ok(FundingRateUpdate::new(
462        instrument_id,
463        funding_rate,
464        funding_interval,
465        next_funding_ns,
466        ts_event,
467        ts_init,
468    ))
469}
470
471/// Parses a linear/inverse ticker payload into a [`MarkPriceUpdate`].
472///
473/// # Errors
474///
475/// Returns an error if the mark_price field is missing or cannot be parsed.
476pub fn parse_ticker_linear_mark_price(
477    data: &BybitWsTickerLinear,
478    instrument: &InstrumentAny,
479    ts_event: UnixNanos,
480    ts_init: UnixNanos,
481) -> anyhow::Result<MarkPriceUpdate> {
482    let mark_price_str = data
483        .mark_price
484        .as_ref()
485        .context("Bybit ticker missing mark_price")?;
486
487    let price =
488        parse_price_with_precision(mark_price_str, instrument.price_precision(), "mark_price")?;
489
490    Ok(MarkPriceUpdate::new(
491        instrument.id(),
492        price,
493        ts_event,
494        ts_init,
495    ))
496}
497
498/// Parses a linear/inverse ticker payload into an [`IndexPriceUpdate`].
499///
500/// # Errors
501///
502/// Returns an error if the index_price field is missing or cannot be parsed.
503pub fn parse_ticker_linear_index_price(
504    data: &BybitWsTickerLinear,
505    instrument: &InstrumentAny,
506    ts_event: UnixNanos,
507    ts_init: UnixNanos,
508) -> anyhow::Result<IndexPriceUpdate> {
509    let index_price_str = data
510        .index_price
511        .as_ref()
512        .context("Bybit ticker missing index_price")?;
513
514    let price =
515        parse_price_with_precision(index_price_str, instrument.price_precision(), "index_price")?;
516
517    Ok(IndexPriceUpdate::new(
518        instrument.id(),
519        price,
520        ts_event,
521        ts_init,
522    ))
523}
524
525/// Parses an option ticker payload into a [`MarkPriceUpdate`].
526///
527/// # Errors
528///
529/// Returns an error if the mark_price field cannot be parsed.
530pub fn parse_ticker_option_mark_price(
531    msg: &BybitWsTickerOptionMsg,
532    instrument: &InstrumentAny,
533    ts_init: UnixNanos,
534) -> anyhow::Result<MarkPriceUpdate> {
535    let ts_event = parse_millis_i64(msg.ts, "ticker.ts")?;
536
537    let price = parse_price_with_precision(
538        &msg.data.mark_price,
539        instrument.price_precision(),
540        "mark_price",
541    )?;
542
543    Ok(MarkPriceUpdate::new(
544        instrument.id(),
545        price,
546        ts_event,
547        ts_init,
548    ))
549}
550
551/// Parses an option ticker payload into an [`IndexPriceUpdate`].
552///
553/// # Errors
554///
555/// Returns an error if the index_price field cannot be parsed.
556pub fn parse_ticker_option_index_price(
557    msg: &BybitWsTickerOptionMsg,
558    instrument: &InstrumentAny,
559    ts_init: UnixNanos,
560) -> anyhow::Result<IndexPriceUpdate> {
561    let ts_event = parse_millis_i64(msg.ts, "ticker.ts")?;
562
563    let price = parse_price_with_precision(
564        &msg.data.index_price,
565        instrument.price_precision(),
566        "index_price",
567    )?;
568
569    Ok(IndexPriceUpdate::new(
570        instrument.id(),
571        price,
572        ts_event,
573        ts_init,
574    ))
575}
576
577/// Parses an option ticker payload into [`OptionGreeks`].
578///
579/// # Errors
580///
581/// Returns an error if any of the greek fields cannot be parsed as f64.
582pub fn parse_ticker_option_greeks(
583    msg: &BybitWsTickerOptionMsg,
584    instrument: &InstrumentAny,
585    ts_init: UnixNanos,
586) -> anyhow::Result<OptionGreeks> {
587    let ts_event = parse_millis_i64(msg.ts, "ticker.ts")?;
588
589    let delta: f64 = msg.data.delta.parse().context("invalid delta")?;
590    let gamma: f64 = msg.data.gamma.parse().context("invalid gamma")?;
591    let vega: f64 = msg.data.vega.parse().context("invalid vega")?;
592    let theta: f64 = msg.data.theta.parse().context("invalid theta")?;
593
594    let bid_iv: f64 = msg.data.bid_iv.parse().context("invalid bid_iv")?;
595    let ask_iv: f64 = msg.data.ask_iv.parse().context("invalid ask_iv")?;
596    let mark_iv: f64 = msg
597        .data
598        .mark_price_iv
599        .parse()
600        .context("invalid mark_price_iv")?;
601    let underlying_price: f64 = msg
602        .data
603        .underlying_price
604        .parse()
605        .context("invalid underlying_price")?;
606    let open_interest: f64 = msg
607        .data
608        .open_interest
609        .parse()
610        .context("invalid open_interest")?;
611
612    Ok(OptionGreeks {
613        instrument_id: instrument.id(),
614        convention: GreeksConvention::BlackScholes,
615        greeks: OptionGreekValues {
616            delta,
617            gamma,
618            vega,
619            theta,
620            rho: 0.0, // Bybit doesn't provide rho
621        },
622        mark_iv: Some(mark_iv),
623        bid_iv: Some(bid_iv),
624        ask_iv: Some(ask_iv),
625        underlying_price: Some(underlying_price),
626        open_interest: Some(open_interest),
627        ts_event,
628        ts_init,
629    })
630}
631
632pub(crate) fn parse_millis_i64(value: i64, field: &str) -> anyhow::Result<UnixNanos> {
633    if value < 0 {
634        Err(anyhow::anyhow!("{field} must be non-negative, was {value}"))
635    } else {
636        let nanos = (value as u64)
637            .checked_mul(NANOSECONDS_IN_MILLISECOND)
638            .ok_or_else(|| anyhow::anyhow!("millisecond timestamp overflowed"))?;
639        Ok(UnixNanos::from(nanos))
640    }
641}
642
643/// Parses a WebSocket kline payload into a [`Bar`].
644///
645/// # Errors
646///
647/// Returns an error if price or volume fields cannot be parsed or if the bar cannot be constructed.
648pub fn parse_ws_kline_bar(
649    kline: &BybitWsKline,
650    instrument: &InstrumentAny,
651    bar_type: BarType,
652    timestamp_on_close: bool,
653    ts_init: UnixNanos,
654) -> anyhow::Result<Bar> {
655    let price_precision = instrument.price_precision();
656    let size_precision = instrument.size_precision();
657
658    let open = parse_price_with_precision(&kline.open, price_precision, "kline.open")?;
659    let high = parse_price_with_precision(&kline.high, price_precision, "kline.high")?;
660    let low = parse_price_with_precision(&kline.low, price_precision, "kline.low")?;
661    let close = parse_price_with_precision(&kline.close, price_precision, "kline.close")?;
662    let volume = parse_quantity_with_precision(&kline.volume, size_precision, "kline.volume")?;
663
664    let mut ts_event = parse_millis_i64(kline.start, "kline.start")?;
665
666    if timestamp_on_close {
667        let interval_ns = bar_type.spec().timedelta().as_nanos();
668        let interval_ns = u64::try_from(interval_ns)
669            .context("bar interval overflowed the u64 range for nanoseconds")?;
670        let updated = ts_event
671            .as_u64()
672            .checked_add(interval_ns)
673            .context("bar timestamp overflowed when adjusting to close time")?;
674        ts_event = UnixNanos::from(updated);
675    }
676    let ts_init = if ts_init.is_zero() { ts_event } else { ts_init };
677
678    Bar::new_checked(bar_type, open, high, low, close, volume, ts_event, ts_init)
679        .context("failed to construct Bar from Bybit WebSocket kline")
680}
681
682/// Parses a WebSocket account order payload into an [`OrderStatusReport`].
683///
684/// # Errors
685///
686/// Returns an error if price or quantity fields cannot be parsed or timestamps are invalid.
687pub fn parse_ws_order_status_report(
688    order: &BybitWsAccountOrder,
689    instrument: &InstrumentAny,
690    account_id: AccountId,
691    ts_init: UnixNanos,
692) -> anyhow::Result<OrderStatusReport> {
693    let instrument_id = instrument.id();
694    let venue_order_id = VenueOrderId::new(order.order_id);
695    let order_side: Option<OrderSide> = order.side.into();
696
697    let order_type = parse_bybit_order_type(
698        order.order_type,
699        order.stop_order_type,
700        order.trigger_direction,
701        order.side,
702    );
703
704    let time_in_force: TimeInForce = match order.time_in_force {
705        BybitTimeInForce::Gtc => TimeInForce::Gtc,
706        BybitTimeInForce::Ioc => TimeInForce::Ioc,
707        BybitTimeInForce::Fok => TimeInForce::Fok,
708        BybitTimeInForce::PostOnly | BybitTimeInForce::Rpi => TimeInForce::Gtc,
709    };
710
711    let quantity =
712        parse_quantity_with_precision(&order.qty, instrument.size_precision(), "order.qty")?;
713
714    let filled_qty = parse_quantity_with_precision(
715        &order.cum_exec_qty,
716        instrument.size_precision(),
717        "order.cumExecQty",
718    )?;
719
720    // Map Bybit order status to Nautilus order status
721    // Special case: if Bybit reports "Rejected" but the order has fills, treat it as Canceled.
722    // This handles the case where the exchange partially fills an order then rejects the
723    // remaining quantity (e.g., due to margin, risk limits, or liquidity constraints).
724    // The state machine does not allow PARTIALLY_FILLED -> REJECTED transitions.
725    let order_status: OrderStatus = match order.order_status {
726        BybitOrderStatus::Created | BybitOrderStatus::New | BybitOrderStatus::Untriggered => {
727            OrderStatus::Accepted
728        }
729        BybitOrderStatus::Rejected => {
730            if filled_qty.is_positive() {
731                OrderStatus::Canceled
732            } else {
733                OrderStatus::Rejected
734            }
735        }
736        BybitOrderStatus::PartiallyFilled => OrderStatus::PartiallyFilled,
737        BybitOrderStatus::Filled => OrderStatus::Filled,
738        // A post-only order that would take liquidity is reported as Cancelled with
739        // rejectReason=EC_PostOnlyWillTakeLiquidity (not Rejected). Surface it as Rejected
740        // for consistency with the tracked-order event path.
741        BybitOrderStatus::Canceled
742            if filled_qty.is_zero()
743                && bybit_rejection_due_post_only(order.reject_reason.as_str()) =>
744        {
745            OrderStatus::Rejected
746        }
747        BybitOrderStatus::Canceled | BybitOrderStatus::PartiallyFilledCanceled => {
748            OrderStatus::Canceled
749        }
750        BybitOrderStatus::Triggered => OrderStatus::Triggered,
751        BybitOrderStatus::Deactivated => OrderStatus::Canceled,
752    };
753
754    let ts_accepted = parse_millis_timestamp(&order.created_time, "order.createdTime")?;
755    let ts_last = parse_millis_timestamp(&order.updated_time, "order.updatedTime")?;
756
757    let mut report = OrderStatusReport::new(
758        account_id,
759        instrument_id,
760        None,
761        venue_order_id,
762        order_side,
763        order_type,
764        time_in_force,
765        order_status,
766        quantity,
767        filled_qty,
768        ts_accepted,
769        ts_last,
770        ts_init,
771        Some(UUID4::new()),
772    );
773
774    if !order.order_link_id.is_empty() {
775        report = report.with_client_order_id(ClientOrderId::new(order.order_link_id));
776    }
777
778    if !order.price.is_empty() && order.price != "0" {
779        let price =
780            parse_price_with_precision(&order.price, instrument.price_precision(), "order.price")?;
781        report = report.with_price(price);
782    }
783
784    if !order.avg_price.is_empty() && order.avg_price != "0" {
785        let avg_px = order.avg_price.parse::<Decimal>().with_context(|| {
786            format!("Failed to parse avg_price='{}' as Decimal", order.avg_price)
787        })?;
788        report = report.with_avg_px(avg_px);
789    }
790
791    if !order.trigger_price.is_empty() && order.trigger_price != "0" {
792        let trigger_price = parse_price_with_precision(
793            &order.trigger_price,
794            instrument.price_precision(),
795            "order.triggerPrice",
796        )?;
797        report = report.with_trigger_price(trigger_price);
798
799        // Set trigger_type for conditional orders
800        let trigger_type: TriggerType = order.trigger_by.into();
801        report = report.with_trigger_type(trigger_type);
802    }
803
804    if let Some(venue_position_id) = make_hedge_venue_position_id(instrument_id, order.position_idx)
805    {
806        report = report.with_venue_position_id(venue_position_id);
807    }
808
809    if order.reduce_only {
810        report = report.with_reduce_only(true);
811    }
812
813    if matches!(
814        order.time_in_force,
815        BybitTimeInForce::PostOnly | BybitTimeInForce::Rpi
816    ) {
817        report = report.with_post_only(true);
818    }
819
820    if !order.reject_reason.is_empty() {
821        report = report.with_cancel_reason(order.reject_reason.to_string());
822    }
823
824    Ok(report)
825}
826
827/// Parses a WebSocket account execution payload into a [`FillReport`].
828///
829/// # Errors
830///
831/// Returns an error if price or quantity fields cannot be parsed or timestamps are invalid.
832pub fn parse_ws_fill_report(
833    execution: &BybitWsAccountExecution,
834    account_id: AccountId,
835    instrument: &InstrumentAny,
836    ts_init: UnixNanos,
837) -> anyhow::Result<FillReport> {
838    let instrument_id = instrument.id();
839    let venue_order_id = VenueOrderId::new(execution.order_id);
840    let trade_id = TradeId::new_checked(execution.exec_id.as_str())
841        .context("invalid execId in Bybit WebSocket execution payload")?;
842
843    let order_side = OrderSide::try_from(execution.side)?;
844    let last_qty = parse_quantity_with_precision(
845        &execution.exec_qty,
846        instrument.size_precision(),
847        "execution.execQty",
848    )?;
849    let last_px = parse_price_with_precision(
850        &execution.exec_price,
851        instrument.price_precision(),
852        "execution.execPrice",
853    )?;
854
855    let liquidity_side = if execution.is_maker {
856        LiquiditySide::Maker
857    } else {
858        LiquiditySide::Taker
859    };
860
861    let fee_decimal: Decimal = execution
862        .exec_fee
863        .parse()
864        .with_context(|| format!("Failed to parse execFee='{}'", execution.exec_fee))?;
865
866    let commission_currency = get_currency(&execution.fee_currency);
867    let commission = Money::from_decimal(fee_decimal, commission_currency).with_context(|| {
868        format!(
869            "Failed to create commission from execFee='{}'",
870            execution.exec_fee
871        )
872    })?;
873    let ts_event = parse_millis_timestamp(&execution.exec_time, "execution.execTime")?;
874
875    let client_order_id = if execution.order_link_id.is_empty() {
876        None
877    } else {
878        Some(ClientOrderId::new(execution.order_link_id))
879    };
880
881    Ok(FillReport::new(
882        account_id,
883        instrument_id,
884        venue_order_id,
885        trade_id,
886        order_side,
887        last_qty,
888        last_px,
889        commission,
890        liquidity_side,
891        client_order_id,
892        None, // venue_position_id: execution data lacks position_idx
893        ts_event,
894        ts_init,
895        None, // report_id
896    ))
897}
898
899/// Parses a fast-stream WebSocket execution payload into a [`FillReport`].
900///
901/// The `execution.fast` channel omits fee and exec-type fields, so the resulting
902/// report carries zero commission. Liquidity side is derived from the payload's
903/// `isMaker` flag. Pair with the standard `execution` channel if exact fee data
904/// is required.
905///
906/// `venue_position_id` should be supplied for tracked hedge-mode orders so the
907/// emitted report carries the long/short position identity that the standard
908/// channel preserves via `OrderFilled`.
909///
910/// # Errors
911///
912/// Returns an error if price or quantity fields cannot be parsed or timestamps are invalid.
913pub fn parse_ws_fill_report_fast(
914    execution: &BybitWsAccountExecutionFast,
915    account_id: AccountId,
916    instrument: &InstrumentAny,
917    venue_position_id: Option<PositionId>,
918    ts_init: UnixNanos,
919) -> anyhow::Result<FillReport> {
920    let instrument_id = instrument.id();
921    let venue_order_id = VenueOrderId::new(execution.order_id);
922    let trade_id = TradeId::new_checked(execution.exec_id.as_str())
923        .context("invalid execId in Bybit WebSocket fast-execution payload")?;
924
925    let order_side = OrderSide::try_from(execution.side)?;
926    let last_qty = parse_quantity_with_precision(
927        &execution.exec_qty,
928        instrument.size_precision(),
929        "execution.execQty",
930    )?;
931    let last_px = parse_price_with_precision(
932        &execution.exec_price,
933        instrument.price_precision(),
934        "execution.execPrice",
935    )?;
936
937    let liquidity_side = if execution.is_maker {
938        LiquiditySide::Maker
939    } else {
940        LiquiditySide::Taker
941    };
942
943    // execution.fast carries no fee data (no rate or currency)
944    let commission_currency = instrument.quote_currency();
945    let commission = Money::from_decimal(Decimal::ZERO, commission_currency)
946        .with_context(|| format!("Failed to create zero commission for {commission_currency}"))?;
947    let ts_event = parse_millis_timestamp(&execution.exec_time, "execution.execTime")?;
948
949    let client_order_id = if execution.order_link_id.is_empty() {
950        None
951    } else {
952        Some(ClientOrderId::new(execution.order_link_id))
953    };
954
955    Ok(FillReport::new(
956        account_id,
957        instrument_id,
958        venue_order_id,
959        trade_id,
960        order_side,
961        last_qty,
962        last_px,
963        commission,
964        liquidity_side,
965        client_order_id,
966        venue_position_id,
967        ts_event,
968        ts_init,
969        None,
970    ))
971}
972
973/// Parses a WebSocket account position payload into a [`PositionStatusReport`].
974///
975/// # Errors
976///
977/// Returns an error if position size or prices cannot be parsed.
978pub fn parse_ws_position_status_report(
979    position: &BybitWsAccountPosition,
980    account_id: AccountId,
981    instrument: &InstrumentAny,
982    ts_init: UnixNanos,
983) -> anyhow::Result<PositionStatusReport> {
984    let instrument_id = instrument.id();
985
986    // Parse absolute size as unsigned Quantity
987    let quantity = parse_quantity_with_precision(
988        &position.size,
989        instrument.size_precision(),
990        "position.size",
991    )?;
992
993    let position_side = match position.side {
994        BybitPositionSide::Buy => PositionSide::Long,
995        BybitPositionSide::Sell => PositionSide::Short,
996        BybitPositionSide::Flat => PositionSide::Flat,
997    };
998
999    // Bybit ranks open positions 1-5 by ADL priority (5 = next to be deleveraged);
1000    // 0 means the account has no open position or is flat. Warn when approaching the
1001    // top tier so operators can react before the venue force-closes.
1002    if position.adl_rank_indicator >= 4 {
1003        log::warn!(
1004            "Elevated ADL risk: {} position size={} adl_rank={}",
1005            instrument_id,
1006            position.size,
1007            position.adl_rank_indicator,
1008        );
1009    }
1010
1011    let ts_last = parse_millis_timestamp(&position.updated_time, "position.updatedTime")?;
1012
1013    let venue_position_id = make_hedge_venue_position_id(instrument_id, position.position_idx);
1014
1015    Ok(PositionStatusReport::new(
1016        account_id,
1017        instrument_id,
1018        position_side,
1019        quantity,
1020        ts_last,
1021        ts_init,
1022        None, // report_id
1023        venue_position_id,
1024        position.entry_price, // avg_px_open
1025    ))
1026}
1027
1028/// Parses a WebSocket account wallet payload into an [`AccountState`].
1029///
1030/// # Errors
1031///
1032/// Returns an error if balance fields cannot be parsed.
1033pub fn parse_ws_account_state(
1034    wallet: &BybitWsAccountWallet,
1035    account_id: AccountId,
1036    ts_event: UnixNanos,
1037    ts_init: UnixNanos,
1038) -> anyhow::Result<AccountState> {
1039    let mut balances = Vec::new();
1040    let mut margins = Vec::new();
1041
1042    for coin_data in &wallet.coin {
1043        let currency = get_currency(coin_data.coin.as_str());
1044        let total_dec = coin_data.wallet_balance - coin_data.spot_borrow;
1045        let locked_dec = coin_data.total_order_im + coin_data.total_position_im;
1046
1047        balances.push(AccountBalance::from_total_and_locked(
1048            total_dec, locked_dec, currency,
1049        )?);
1050
1051        // Sum position IM (reserved by open positions) and order IM (reserved by
1052        // pending orders) so the reported initial margin reflects either source.
1053        let initial_margin_dec = coin_data.total_position_im + coin_data.total_order_im;
1054        let maintenance_margin_dec = match &coin_data.total_position_mm {
1055            Some(mm) if !mm.is_empty() => mm.parse::<Decimal>()?,
1056            _ => Decimal::ZERO,
1057        };
1058
1059        if !initial_margin_dec.is_zero() || !maintenance_margin_dec.is_zero() {
1060            margins.push(MarginBalance::new(
1061                Money::from_decimal(initial_margin_dec, currency)?,
1062                Money::from_decimal(maintenance_margin_dec, currency)?,
1063                None,
1064            ));
1065        }
1066    }
1067
1068    Ok(AccountState::new(
1069        account_id,
1070        AccountType::Margin, // Bybit unified account
1071        balances,
1072        margins,
1073        true, // is_reported
1074        UUID4::new(),
1075        ts_event,
1076        ts_init,
1077        None, // base_currency
1078    ))
1079}
1080
1081#[cfg(test)]
1082mod tests {
1083    use std::str::FromStr;
1084
1085    use nautilus_model::{
1086        data::BarSpecification,
1087        enums::{
1088            AggregationSource, BarAggregation, OrderType, PositionSide, PriceType, TriggerType,
1089        },
1090        identifiers::PositionId,
1091    };
1092    use rstest::rstest;
1093    use rust_decimal_macros::dec;
1094
1095    use super::*;
1096    use crate::{
1097        common::{
1098            enums::{BybitExecType, BybitOrderSide, BybitProductType},
1099            parse::{parse_linear_instrument, parse_option_instrument},
1100            testing::load_test_json,
1101        },
1102        http::models::{BybitInstrumentLinearResponse, BybitInstrumentOptionResponse},
1103        websocket::messages::{
1104            BybitWsAccountExecutionMsg, BybitWsOrderbookDepthMsg, BybitWsTickerLinearMsg,
1105            BybitWsTickerOptionMsg, BybitWsTradeMsg,
1106        },
1107    };
1108
1109    const TS: UnixNanos = UnixNanos::new(1_700_000_000_000_000_000);
1110
1111    use ustr::Ustr;
1112
1113    use crate::http::models::BybitFeeRate;
1114
1115    fn sample_fee_rate(
1116        symbol: &str,
1117        taker: &str,
1118        maker: &str,
1119        base_coin: Option<&str>,
1120    ) -> BybitFeeRate {
1121        BybitFeeRate {
1122            symbol: Ustr::from(symbol),
1123            taker_fee_rate: taker.to_string(),
1124            maker_fee_rate: maker.to_string(),
1125            base_coin: base_coin.map(Ustr::from),
1126        }
1127    }
1128
1129    fn linear_instrument() -> InstrumentAny {
1130        let json = load_test_json("http_get_instruments_linear.json");
1131        let response: BybitInstrumentLinearResponse = serde_json::from_str(&json).unwrap();
1132        let instrument = &response.result.list[0];
1133        let fee_rate = sample_fee_rate("BTCUSDT", "0.00055", "0.0001", Some("BTC"));
1134        parse_linear_instrument(instrument, &fee_rate, TS, TS).unwrap()
1135    }
1136
1137    fn option_instrument() -> InstrumentAny {
1138        let json = load_test_json("http_get_instruments_option.json");
1139        let response: BybitInstrumentOptionResponse = serde_json::from_str(&json).unwrap();
1140        let instrument = &response.result.list[0];
1141        parse_option_instrument(instrument, None, TS, TS).unwrap()
1142    }
1143
1144    #[rstest]
1145    fn parse_ws_trade_into_trade_tick() {
1146        let instrument = linear_instrument();
1147        let json = load_test_json("ws_public_trade.json");
1148        let msg: BybitWsTradeMsg = serde_json::from_str(&json).unwrap();
1149        let trade = &msg.data[0];
1150
1151        let tick = parse_ws_trade_tick(trade, &instrument, TS).unwrap();
1152
1153        assert_eq!(tick.instrument_id, instrument.id());
1154        assert_eq!(tick.price, instrument.make_price(27451.00));
1155        assert_eq!(tick.size, instrument.make_qty(0.010, None));
1156        assert_eq!(tick.aggressor_side, AggressorSide::Buy);
1157        assert_eq!(
1158            tick.trade_id.to_string(),
1159            "9dc75fca-4bdd-4773-9f78-6f5d7ab2a110"
1160        );
1161        assert_eq!(tick.ts_event, UnixNanos::new(1_709_891_679_000_000_000));
1162    }
1163
1164    #[rstest]
1165    fn parse_orderbook_snapshot_into_deltas() {
1166        let instrument = linear_instrument();
1167        let json = load_test_json("ws_orderbook_snapshot.json");
1168        let msg: BybitWsOrderbookDepthMsg = serde_json::from_str(&json).unwrap();
1169
1170        let deltas = parse_orderbook_deltas(&msg, &instrument, TS).unwrap();
1171
1172        assert_eq!(deltas.instrument_id, instrument.id());
1173        assert_eq!(deltas.deltas.len(), 5);
1174        assert_eq!(deltas.deltas[0].action, BookAction::Clear);
1175        assert_eq!(
1176            deltas.deltas[1].order.price,
1177            instrument.make_price(27450.00)
1178        );
1179        assert_eq!(
1180            deltas.deltas[1].order.size,
1181            instrument.make_qty(0.500, None)
1182        );
1183        let last = deltas.deltas.last().unwrap();
1184        assert_eq!(last.order.side, OrderSide::Sell.into());
1185        assert_eq!(last.order.price, instrument.make_price(27451.50));
1186        assert_eq!(
1187            last.flags & RecordFlag::F_LAST as u8,
1188            RecordFlag::F_LAST as u8
1189        );
1190    }
1191
1192    #[rstest]
1193    fn parse_orderbook_delta_marks_actions() {
1194        let instrument = linear_instrument();
1195        let json = load_test_json("ws_orderbook_delta.json");
1196        let msg: BybitWsOrderbookDepthMsg = serde_json::from_str(&json).unwrap();
1197
1198        let deltas = parse_orderbook_deltas(&msg, &instrument, TS).unwrap();
1199
1200        assert_eq!(deltas.deltas.len(), 2);
1201        let bid = &deltas.deltas[0];
1202        assert_eq!(bid.action, BookAction::Update);
1203        assert_eq!(bid.order.side, OrderSide::Buy.into());
1204        assert_eq!(bid.order.size, instrument.make_qty(0.400, None));
1205
1206        let ask = &deltas.deltas[1];
1207        assert_eq!(ask.action, BookAction::Delete);
1208        assert_eq!(ask.order.side, OrderSide::Sell.into());
1209        assert_eq!(ask.order.size, instrument.make_qty(0.0, None));
1210        assert_eq!(
1211            ask.flags & RecordFlag::F_LAST as u8,
1212            RecordFlag::F_LAST as u8
1213        );
1214    }
1215
1216    #[rstest]
1217    fn parse_orderbook_quote_produces_top_of_book() {
1218        let instrument = linear_instrument();
1219        let json = load_test_json("ws_orderbook_snapshot.json");
1220        let msg: BybitWsOrderbookDepthMsg = serde_json::from_str(&json).unwrap();
1221
1222        let quote = parse_orderbook_quote(&msg, &instrument, TS).unwrap();
1223
1224        assert_eq!(quote.instrument_id, instrument.id());
1225        assert_eq!(quote.bid_price, instrument.make_price(27450.00));
1226        assert_eq!(quote.bid_size, instrument.make_qty(0.500, None));
1227        assert_eq!(quote.ask_price, instrument.make_price(27451.00));
1228        assert_eq!(quote.ask_size, instrument.make_qty(0.750, None));
1229    }
1230
1231    #[rstest]
1232    #[case::delta(
1233        "orderbook.1.BTCUSDT",
1234        "delta",
1235        false,
1236        false,
1237        "Expected depth-1 orderbook snapshot"
1238    )]
1239    #[case::depth_50(
1240        "orderbook.50.BTCUSDT",
1241        "snapshot",
1242        false,
1243        false,
1244        "Expected depth-1 orderbook snapshot"
1245    )]
1246    #[case::missing_bid(
1247        "orderbook.1.BTCUSDT",
1248        "snapshot",
1249        true,
1250        false,
1251        "orderbook snapshot missing bid"
1252    )]
1253    #[case::missing_ask(
1254        "orderbook.1.BTCUSDT",
1255        "snapshot",
1256        false,
1257        true,
1258        "orderbook snapshot missing ask"
1259    )]
1260    fn parse_orderbook_quote_rejects_non_top_of_book(
1261        #[case] topic: &str,
1262        #[case] msg_type: &str,
1263        #[case] missing_bid: bool,
1264        #[case] missing_ask: bool,
1265        #[case] expected_error: &str,
1266    ) {
1267        let instrument = linear_instrument();
1268        let mut msg: BybitWsOrderbookDepthMsg =
1269            serde_json::from_str(&load_test_json("ws_orderbook_snapshot.json")).unwrap();
1270        msg.topic = topic.into();
1271        msg.msg_type = msg_type.into();
1272        if missing_bid {
1273            msg.data.b.clear();
1274        }
1275
1276        if missing_ask {
1277            msg.data.a.clear();
1278        }
1279
1280        assert_eq!(
1281            parse_orderbook_quote(&msg, &instrument, TS)
1282                .unwrap_err()
1283                .to_string(),
1284            expected_error
1285        );
1286    }
1287
1288    #[rstest]
1289    fn parse_option_ticker_quote_to_quote_tick() {
1290        let instrument = option_instrument();
1291        let json = load_test_json("ws_ticker_option.json");
1292        let msg: BybitWsTickerOptionMsg = serde_json::from_str(&json).unwrap();
1293
1294        let quote = parse_ticker_option_quote(&msg, &instrument, TS).unwrap();
1295
1296        assert_eq!(quote.instrument_id, instrument.id());
1297        assert_eq!(quote.bid_price, instrument.make_price(0.0));
1298        assert_eq!(quote.ask_price, instrument.make_price(10.0));
1299        assert_eq!(quote.bid_size, instrument.make_qty(0.0, None));
1300        assert_eq!(quote.ask_size, instrument.make_qty(5.1, None));
1301        assert_eq!(quote.ts_event, UnixNanos::new(1_672_917_511_074_000_000));
1302        assert_eq!(quote.ts_init, TS);
1303    }
1304
1305    #[rstest]
1306    #[case::timestamp_on_open(false, 1_672_324_800_000_000_000)]
1307    #[case::timestamp_on_close(true, 1_672_325_100_000_000_000)]
1308    fn parse_ws_kline_into_bar(#[case] timestamp_on_close: bool, #[case] expected_ts_event: u64) {
1309        use std::num::NonZero;
1310
1311        let instrument = linear_instrument();
1312        let json = load_test_json("ws_kline.json");
1313        let msg: crate::websocket::messages::BybitWsKlineMsg = serde_json::from_str(&json).unwrap();
1314        let kline = &msg.data[0];
1315
1316        let bar_spec = BarSpecification {
1317            step: NonZero::new(5).unwrap(),
1318            aggregation: BarAggregation::Minute,
1319            price_type: PriceType::Last,
1320        };
1321        let bar_type = BarType::new(instrument.id(), bar_spec, AggregationSource::External);
1322
1323        let bar = parse_ws_kline_bar(kline, &instrument, bar_type, timestamp_on_close, TS).unwrap();
1324
1325        assert_eq!(bar.bar_type, bar_type);
1326        assert_eq!(bar.open, instrument.make_price(16649.5));
1327        assert_eq!(bar.high, instrument.make_price(16677.0));
1328        assert_eq!(bar.low, instrument.make_price(16608.0));
1329        assert_eq!(bar.close, instrument.make_price(16677.0));
1330        assert_eq!(bar.volume, instrument.make_qty(2.081, None));
1331        assert_eq!(bar.ts_event, UnixNanos::new(expected_ts_event));
1332        assert_eq!(bar.ts_init, TS);
1333    }
1334
1335    #[rstest]
1336    fn parse_ws_order_into_order_status_report() {
1337        let instrument = linear_instrument();
1338        let json = load_test_json("ws_account_order_filled.json");
1339        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1340            serde_json::from_str(&json).unwrap();
1341        let order = &msg.data[0];
1342        let account_id = AccountId::new("BYBIT-001");
1343
1344        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
1345
1346        assert_eq!(report.account_id, account_id);
1347        assert_eq!(report.instrument_id, instrument.id());
1348        assert_eq!(report.order_side, OrderSide::Buy.into());
1349        assert_eq!(report.order_type, OrderType::Limit);
1350        assert_eq!(report.time_in_force, TimeInForce::Gtc);
1351        assert_eq!(report.order_status, OrderStatus::Filled);
1352        assert_eq!(report.quantity, instrument.make_qty(0.100, None));
1353        assert_eq!(report.filled_qty, instrument.make_qty(0.100, None));
1354        assert_eq!(report.price, Some(instrument.make_price(30000.50)));
1355        assert_eq!(report.avg_px, Some(dec!(30000.50)));
1356        assert_eq!(
1357            report.client_order_id.as_ref().unwrap().to_string(),
1358            "test-client-order-001"
1359        );
1360        assert_eq!(
1361            report.ts_accepted,
1362            UnixNanos::new(1_672_364_262_444_000_000)
1363        );
1364        assert_eq!(report.ts_last, UnixNanos::new(1_672_364_262_457_000_000));
1365    }
1366
1367    #[rstest]
1368    fn parse_ws_order_avg_price_keeps_every_digit_the_venue_sent() {
1369        // 28 significant digits is exactly what `Decimal` holds, and more than `f64` can:
1370        // routing the same string through `f64` first collapses it to 30000.500000000004.
1371        let raw = "30000.50000000000372529029846";
1372        let instrument = linear_instrument();
1373        let json = load_test_json("ws_account_order_filled.json");
1374        let mut msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1375            serde_json::from_str(&json).unwrap();
1376        msg.data[0].avg_price = raw.to_string();
1377
1378        let report = parse_ws_order_status_report(
1379            &msg.data[0],
1380            &instrument,
1381            AccountId::new("BYBIT-001"),
1382            TS,
1383        )
1384        .unwrap();
1385
1386        let via_f64: Decimal = raw.parse::<f64>().unwrap().to_string().parse().unwrap();
1387        assert_eq!(report.avg_px, Some(Decimal::from_str(raw).unwrap()));
1388        assert_ne!(report.avg_px, Some(via_f64));
1389    }
1390
1391    #[rstest]
1392    fn parse_ws_order_partially_filled_rejected_maps_to_canceled() {
1393        let instrument = linear_instrument();
1394        let json = load_test_json("ws_account_order_partially_filled_rejected.json");
1395        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1396            serde_json::from_str(&json).unwrap();
1397        let order = &msg.data[0];
1398        let account_id = AccountId::new("BYBIT-001");
1399
1400        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
1401
1402        // Verify that Bybit "Rejected" status with fills is mapped to Canceled, not Rejected
1403        assert_eq!(report.order_status, OrderStatus::Canceled);
1404        assert_eq!(report.filled_qty, instrument.make_qty(50.0, None));
1405        assert_eq!(
1406            report.client_order_id.as_ref().unwrap().to_string(),
1407            "O-20251001-164609-APEX-000-49"
1408        );
1409        assert_eq!(report.cancel_reason, Some("UNKNOWN".to_string()));
1410    }
1411
1412    #[rstest]
1413    fn parse_ws_order_post_only_cancel_maps_to_rejected() {
1414        let instrument = linear_instrument();
1415        let json = load_test_json("ws_account_order.json");
1416        let mut msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1417            serde_json::from_str(&json).unwrap();
1418
1419        let order = msg.data.first_mut().unwrap();
1420        order.reject_reason = Ustr::from("EC_PostOnlyWillTakeLiquidity");
1421        order.cum_exec_qty = "0".to_string();
1422        let account_id = AccountId::new("BYBIT-001");
1423
1424        let report =
1425            parse_ws_order_status_report(&msg.data[0], &instrument, account_id, TS).unwrap();
1426
1427        assert_eq!(report.order_status, OrderStatus::Rejected);
1428        assert_eq!(
1429            report.cancel_reason,
1430            Some("EC_PostOnlyWillTakeLiquidity".to_string())
1431        );
1432    }
1433
1434    #[rstest]
1435    fn parse_ws_execution_into_fill_report() {
1436        let instrument = linear_instrument();
1437        let json = load_test_json("ws_account_execution.json");
1438        let msg: crate::websocket::messages::BybitWsAccountExecutionMsg =
1439            serde_json::from_str(&json).unwrap();
1440        let execution = &msg.data[0];
1441        let account_id = AccountId::new("BYBIT-001");
1442
1443        let report = parse_ws_fill_report(execution, account_id, &instrument, TS).unwrap();
1444
1445        assert_eq!(report.account_id, account_id);
1446        assert_eq!(report.instrument_id, instrument.id());
1447        assert_eq!(
1448            report.venue_order_id.to_string(),
1449            "9aac161b-8ed6-450d-9cab-c5cc67c21784"
1450        );
1451        assert_eq!(
1452            report.trade_id.to_string(),
1453            "0ab1bdf7-4219-438b-b30a-32ec863018f7"
1454        );
1455        assert_eq!(report.order_side, OrderSide::Sell);
1456        assert_eq!(report.last_qty, instrument.make_qty(0.5, None));
1457        assert_eq!(report.last_px, instrument.make_price(95900.1));
1458        assert_eq!(report.commission.as_f64(), 26.3725275);
1459        assert_eq!(report.commission.currency.code, "USDT");
1460        assert_eq!(report.liquidity_side, LiquiditySide::Taker);
1461        assert_eq!(
1462            report.client_order_id.as_ref().unwrap().to_string(),
1463            "test-order-link-001"
1464        );
1465        assert_eq!(report.ts_event, UnixNanos::new(1_746_270_400_353_000_000));
1466    }
1467
1468    #[rstest]
1469    fn parse_ws_adl_execution_into_fill_report() {
1470        let instrument = linear_instrument();
1471        let json = load_test_json("ws_account_execution_adl.json");
1472        let msg: crate::websocket::messages::BybitWsAccountExecutionMsg =
1473            serde_json::from_str(&json).unwrap();
1474        let execution = &msg.data[0];
1475        let account_id = AccountId::new("BYBIT-001");
1476
1477        assert_eq!(execution.exec_type, BybitExecType::AdlTrade);
1478        assert!(execution.exec_type.is_exchange_generated());
1479        assert!(execution.order_link_id.is_empty());
1480
1481        let report = parse_ws_fill_report(execution, account_id, &instrument, TS).unwrap();
1482
1483        // ADL fills carry an empty orderLinkId; client_order_id is None so the engine
1484        // creates the order as external from the accompanying order status report.
1485        assert_eq!(report.client_order_id, None);
1486        assert_eq!(
1487            report.venue_order_id.to_string(),
1488            "9aac161b-8ed6-450d-9cab-c5cc67c21785"
1489        );
1490        assert_eq!(report.order_side, OrderSide::Sell);
1491        assert_eq!(report.last_qty, instrument.make_qty(0.5, None));
1492        assert_eq!(report.last_px, instrument.make_price(95850.0));
1493        assert_eq!(report.commission.as_f64(), 0.0);
1494        assert_eq!(report.commission.currency.code, "USDT");
1495    }
1496
1497    #[rstest]
1498    #[case::forward_split_settle("ForwardSplitSettle", BybitExecType::ForwardSplitSettle)]
1499    #[case::reverse_split_settle("ReverseSplitSettle", BybitExecType::ReverseSplitSettle)]
1500    #[case::dividend("Dividend", BybitExecType::Dividend)]
1501    fn parse_ws_exchange_generated_execution_into_fill_report(
1502        #[case] exec_type: &str,
1503        #[case] expected: BybitExecType,
1504    ) {
1505        let instrument = linear_instrument();
1506        let json = load_test_json("ws_account_execution_adl.json");
1507        let mut value: serde_json::Value = serde_json::from_str(&json).unwrap();
1508        value["data"][0]["execType"] = serde_json::json!(exec_type);
1509        let msg: crate::websocket::messages::BybitWsAccountExecutionMsg =
1510            serde_json::from_value(value).unwrap();
1511        let execution = &msg.data[0];
1512        let account_id = AccountId::new("BYBIT-001");
1513
1514        assert_eq!(execution.exec_type, expected);
1515        assert!(execution.exec_type.is_exchange_generated());
1516        assert!(execution.order_link_id.is_empty());
1517
1518        let report = parse_ws_fill_report(execution, account_id, &instrument, TS).unwrap();
1519
1520        assert_eq!(report.client_order_id, None);
1521        assert_eq!(
1522            report.venue_order_id.to_string(),
1523            "9aac161b-8ed6-450d-9cab-c5cc67c21785"
1524        );
1525    }
1526
1527    #[rstest]
1528    fn parse_ws_fill_report_venue_position_id_is_none() {
1529        let instrument = linear_instrument();
1530        let json = load_test_json("ws_account_execution.json");
1531        let msg: crate::websocket::messages::BybitWsAccountExecutionMsg =
1532            serde_json::from_str(&json).unwrap();
1533        let execution = &msg.data[0];
1534        let account_id = AccountId::new("BYBIT-001");
1535
1536        let report = parse_ws_fill_report(execution, account_id, &instrument, TS).unwrap();
1537
1538        assert_eq!(report.venue_position_id, None);
1539    }
1540
1541    #[rstest]
1542    fn parse_ws_fill_report_uses_payload_fee_currency() {
1543        let instrument = linear_instrument();
1544        let json = load_test_json("ws_account_execution.json");
1545        let msg: crate::websocket::messages::BybitWsAccountExecutionMsg =
1546            serde_json::from_str(&json).unwrap();
1547
1548        let mut execution = msg.data[0].clone();
1549        execution.fee_currency = Ustr::from("BTC");
1550        let account_id = AccountId::new("BYBIT-001");
1551
1552        let report = parse_ws_fill_report(&execution, account_id, &instrument, TS).unwrap();
1553
1554        assert_eq!(report.commission.currency.code, "BTC");
1555    }
1556
1557    fn fast_execution(is_maker: bool, order_link_id: &str) -> BybitWsAccountExecutionFast {
1558        BybitWsAccountExecutionFast {
1559            category: BybitProductType::Linear,
1560            symbol: Ustr::from("BTCUSDT"),
1561            exec_id: "abc-123".to_string(),
1562            exec_price: "50000.0".to_string(),
1563            exec_qty: "0.5".to_string(),
1564            order_id: Ustr::from("ord-1"),
1565            order_link_id: Ustr::from(order_link_id),
1566            side: BybitOrderSide::Buy,
1567            exec_time: "1716800399334".to_string(),
1568            is_maker,
1569            seq: 42,
1570        }
1571    }
1572
1573    #[rstest]
1574    // Maker fast fill: docs say orderLinkId is always empty -> client_order_id is None.
1575    #[case(true, "", LiquiditySide::Maker, None)]
1576    // Taker fast fill: orderLinkId is populated -> client_order_id is set.
1577    #[case(false, "link-1", LiquiditySide::Taker, Some("link-1"))]
1578    fn parse_ws_fill_report_fast_maps_is_maker_and_link_id(
1579        #[case] is_maker: bool,
1580        #[case] order_link_id: &str,
1581        #[case] expected_liquidity: LiquiditySide,
1582        #[case] expected_cid: Option<&str>,
1583    ) {
1584        let instrument = linear_instrument();
1585        let exec = fast_execution(is_maker, order_link_id);
1586        let account_id = AccountId::new("BYBIT-001");
1587
1588        let report = parse_ws_fill_report_fast(&exec, account_id, &instrument, None, TS).unwrap();
1589
1590        assert_eq!(report.account_id, account_id);
1591        assert_eq!(report.instrument_id, instrument.id());
1592        assert_eq!(report.venue_order_id.to_string(), "ord-1");
1593        assert_eq!(report.trade_id.to_string(), "abc-123");
1594        assert_eq!(report.order_side, OrderSide::Buy);
1595        assert_eq!(report.last_qty, instrument.make_qty(0.5, None));
1596        assert_eq!(report.last_px, instrument.make_price(50000.0));
1597        assert_eq!(report.commission.as_f64(), 0.0);
1598        assert_eq!(report.liquidity_side, expected_liquidity);
1599        assert_eq!(
1600            report.client_order_id.map(|c| c.to_string()),
1601            expected_cid.map(str::to_string),
1602        );
1603        assert_eq!(report.venue_position_id, None);
1604        assert_eq!(report.ts_event, UnixNanos::new(1_716_800_399_334_000_000));
1605    }
1606
1607    #[rstest]
1608    fn parse_ws_fill_report_fast_preserves_venue_position_id() {
1609        let instrument = linear_instrument();
1610        let exec = fast_execution(false, "link-hedge");
1611        let account_id = AccountId::new("BYBIT-001");
1612        let venue_pid = PositionId::from("BTCUSDT-LINEAR.BYBIT-LONG");
1613
1614        let report =
1615            parse_ws_fill_report_fast(&exec, account_id, &instrument, Some(venue_pid), TS).unwrap();
1616
1617        assert_eq!(report.venue_position_id, Some(venue_pid));
1618    }
1619
1620    #[rstest]
1621    fn parse_bybit_ws_frame_routes_execution_fast_topic() {
1622        // The fixture topic is `execution.fast` (matches the venue-doc sample),
1623        // which the prefix check at parse_bybit_ws_frame routes to the fast frame.
1624        let value: serde_json::Value =
1625            serde_json::from_str(&load_test_json("ws_account_execution_fast.json")).unwrap();
1626        let frame = parse_bybit_ws_frame(value);
1627        assert!(
1628            matches!(frame, BybitWsFrame::AccountExecutionFast(_)),
1629            "expected AccountExecutionFast, found {frame:?}",
1630        );
1631    }
1632
1633    #[rstest]
1634    fn parse_bybit_ws_frame_routes_standard_execution_topic() {
1635        // Sanity: with the fast prefix checked first, a plain `execution.<cat>` topic
1636        // must still route to the standard variant (not the fast one).
1637        let envelope = BybitWsAccountExecutionMsg {
1638            topic: Ustr::from("execution.linear"),
1639            id: "std-1".to_string(),
1640            creation_time: 1_716_800_399_338,
1641            data: vec![],
1642        };
1643        let value = serde_json::to_value(envelope).unwrap();
1644        let frame = parse_bybit_ws_frame(value);
1645        assert!(
1646            matches!(frame, BybitWsFrame::AccountExecution(_)),
1647            "expected AccountExecution, found {frame:?}",
1648        );
1649    }
1650
1651    #[rstest]
1652    fn parse_bybit_ws_frame_unrecognized_exec_type_still_routes_execution() {
1653        let json = load_test_json("ws_account_execution_adl.json");
1654        let mut value: serde_json::Value = serde_json::from_str(&json).unwrap();
1655        value["data"][0]["execType"] = serde_json::json!("StockMerger");
1656
1657        let frame = parse_bybit_ws_frame(value);
1658
1659        match frame {
1660            BybitWsFrame::AccountExecution(msg) => {
1661                assert_eq!(msg.data[0].exec_type, BybitExecType::Unknown);
1662            }
1663            other => panic!("Expected AccountExecution, found {other:?}"),
1664        }
1665    }
1666
1667    #[rstest]
1668    fn parse_ws_order_status_report_venue_position_id_is_none_for_tp() {
1669        let instrument = linear_instrument();
1670        let json = load_test_json("ws_account_order_take_profit.json");
1671        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1672            serde_json::from_str(&json).unwrap();
1673        let order = &msg.data[0]; // positionIdx=0
1674        let account_id = AccountId::new("BYBIT-001");
1675
1676        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
1677
1678        assert_eq!(report.venue_position_id, None);
1679    }
1680
1681    #[rstest]
1682    fn parse_ws_order_status_report_venue_position_id_for_hedge() {
1683        let instrument = linear_instrument();
1684        let json = load_test_json("ws_account_order_take_profit.json");
1685        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1686            serde_json::from_str(&json).unwrap();
1687        let mut order = msg.data[0].clone();
1688        order.position_idx = 1;
1689        let account_id = AccountId::new("BYBIT-001");
1690
1691        let report = parse_ws_order_status_report(&order, &instrument, account_id, TS).unwrap();
1692
1693        assert_eq!(
1694            report.venue_position_id,
1695            Some(PositionId::from("BTCUSDT-LINEAR.BYBIT-LONG"))
1696        );
1697    }
1698
1699    #[rstest]
1700    fn parse_ws_position_into_position_status_report() {
1701        let instrument = linear_instrument();
1702        let json = load_test_json("ws_account_position.json");
1703        let msg: crate::websocket::messages::BybitWsAccountPositionMsg =
1704            serde_json::from_str(&json).unwrap();
1705        let position = &msg.data[0];
1706        let account_id = AccountId::new("BYBIT-001");
1707
1708        let report =
1709            parse_ws_position_status_report(position, account_id, &instrument, TS).unwrap();
1710
1711        assert_eq!(report.account_id, account_id);
1712        assert_eq!(report.instrument_id, instrument.id());
1713        assert_eq!(report.position_side, PositionSide::Short);
1714        assert_eq!(report.quantity, instrument.make_qty(0.01, None));
1715        assert_eq!(
1716            report.avg_px_open,
1717            Some(Decimal::try_from(3641.075).unwrap())
1718        );
1719        assert_eq!(report.ts_last, UnixNanos::new(1_762_199_125_472_000_000));
1720        assert_eq!(report.ts_init, TS);
1721    }
1722
1723    #[rstest]
1724    fn parse_ws_position_status_report_venue_position_id_for_hedge() {
1725        let instrument = linear_instrument();
1726        let json = load_test_json("ws_account_position.json");
1727        let msg: crate::websocket::messages::BybitWsAccountPositionMsg =
1728            serde_json::from_str(&json).unwrap();
1729        let mut position = msg.data[0].clone();
1730        position.position_idx = 2;
1731        let account_id = AccountId::new("BYBIT-001");
1732
1733        let report =
1734            parse_ws_position_status_report(&position, account_id, &instrument, TS).unwrap();
1735
1736        assert_eq!(
1737            report.venue_position_id,
1738            Some(PositionId::from("BTCUSDT-LINEAR.BYBIT-SHORT"))
1739        );
1740    }
1741
1742    #[rstest]
1743    fn parse_ws_position_short_into_position_status_report() {
1744        // Create ETHUSDT instrument
1745        let instruments_json = load_test_json("http_get_instruments_linear.json");
1746        let instruments_response: crate::http::models::BybitInstrumentLinearResponse =
1747            serde_json::from_str(&instruments_json).unwrap();
1748        let eth_def = &instruments_response.result.list[1]; // ETHUSDT is second in the list
1749        let fee_rate = crate::http::models::BybitFeeRate {
1750            symbol: Ustr::from("ETHUSDT"),
1751            taker_fee_rate: "0.00055".to_string(),
1752            maker_fee_rate: "0.0001".to_string(),
1753            base_coin: Some(Ustr::from("ETH")),
1754        };
1755        let instrument =
1756            crate::common::parse::parse_linear_instrument(eth_def, &fee_rate, TS, TS).unwrap();
1757
1758        let json = load_test_json("ws_account_position_short.json");
1759        let msg: crate::websocket::messages::BybitWsAccountPositionMsg =
1760            serde_json::from_str(&json).unwrap();
1761        let position = &msg.data[0];
1762        let account_id = AccountId::new("BYBIT-001");
1763
1764        let report =
1765            parse_ws_position_status_report(position, account_id, &instrument, TS).unwrap();
1766
1767        assert_eq!(report.account_id, account_id);
1768        assert_eq!(report.instrument_id.symbol.as_str(), "ETHUSDT-LINEAR");
1769        assert_eq!(report.position_side, PositionSide::Short);
1770        assert_eq!(report.quantity, instrument.make_qty(0.01, None));
1771        assert_eq!(
1772            report.avg_px_open,
1773            Some(Decimal::try_from(3641.075).unwrap())
1774        );
1775        assert_eq!(report.ts_last, UnixNanos::new(1_762_199_125_472_000_000));
1776        assert_eq!(report.ts_init, TS);
1777    }
1778
1779    #[rstest]
1780    fn parse_ws_wallet_into_account_state() {
1781        let json = load_test_json("ws_account_wallet.json");
1782        let msg: crate::websocket::messages::BybitWsAccountWalletMsg =
1783            serde_json::from_str(&json).unwrap();
1784        let wallet = &msg.data[0];
1785        let account_id = AccountId::new("BYBIT-001");
1786        let ts_event = UnixNanos::new(1_700_034_722_104_000_000);
1787
1788        let state = parse_ws_account_state(wallet, account_id, ts_event, TS).unwrap();
1789
1790        assert_eq!(state.account_id, account_id);
1791        assert_eq!(state.account_type, AccountType::Margin);
1792        assert_eq!(state.balances.len(), 2);
1793        assert!(state.is_reported);
1794
1795        // Check BTC balance
1796        let btc_balance = &state.balances[0];
1797        assert_eq!(btc_balance.currency.code, "BTC");
1798        assert!((btc_balance.total.as_f64() - 0.00102964).abs() < 1e-8);
1799        assert!((btc_balance.free.as_f64() - 0.00092964).abs() < 1e-8);
1800        assert!((btc_balance.locked.as_f64() - 0.0001).abs() < 1e-8);
1801
1802        // Check USDT balance
1803        let usdt_balance = &state.balances[1];
1804        assert_eq!(usdt_balance.currency.code, "USDT");
1805        assert!((usdt_balance.total.as_f64() - 9647.75537647).abs() < 1e-6);
1806        assert!((usdt_balance.free.as_f64() - 9519.89806037).abs() < 1e-6);
1807        assert!((usdt_balance.locked.as_f64() - 127.8573161).abs() < 1e-6);
1808
1809        // BTC has order IM only (no position), USDT has position IM+MM (no orders).
1810        assert_eq!(state.margins.len(), 2);
1811        assert!(state.margins.iter().all(|m| m.instrument_id.is_none()));
1812
1813        let btc_margin = state
1814            .margins
1815            .iter()
1816            .find(|m| m.currency.code == "BTC")
1817            .expect("BTC margin missing");
1818        assert!((btc_margin.initial.as_f64() - 0.0001).abs() < 1e-8);
1819        assert!(btc_margin.maintenance.as_f64().abs() < 1e-9);
1820
1821        let usdt_margin = state
1822            .margins
1823            .iter()
1824            .find(|m| m.currency.code == "USDT")
1825            .expect("USDT margin missing");
1826        assert!((usdt_margin.initial.as_f64() - 127.8573161).abs() < 1e-6);
1827        assert!((usdt_margin.maintenance.as_f64() - 12.78573161).abs() < 1e-6);
1828
1829        assert_eq!(state.ts_event, ts_event);
1830        assert_eq!(state.ts_init, TS);
1831    }
1832
1833    #[rstest]
1834    fn parse_ws_wallet_with_small_order_calculates_free_correctly() {
1835        // Regression test for issue where availableToWithdraw=0 caused all funds to appear locked
1836        // When a small order is placed, Bybit may report availableToWithdraw=0 due to margin calculations,
1837        // but totalOrderIM correctly shows only the margin locked for the order
1838        let json = load_test_json("ws_account_wallet_small_order.json");
1839        let msg: crate::websocket::messages::BybitWsAccountWalletMsg =
1840            serde_json::from_str(&json).unwrap();
1841        let wallet = &msg.data[0];
1842        let account_id = AccountId::new("BYBIT-UNIFIED");
1843        let ts_event = UnixNanos::new(1_762_960_669_000_000_000);
1844
1845        let state = parse_ws_account_state(wallet, account_id, ts_event, TS).unwrap();
1846
1847        assert_eq!(state.account_id, account_id);
1848        assert_eq!(state.balances.len(), 1);
1849
1850        // Check USDT balance
1851        let usdt_balance = &state.balances[0];
1852        assert_eq!(usdt_balance.currency.code, "USDT");
1853
1854        // Wallet has 51,333.82 USDT total
1855        assert!((usdt_balance.total.as_f64() - 51333.82543837).abs() < 1e-6);
1856
1857        // Only 50.028 USDT should be locked (for the order), not all funds
1858        assert!((usdt_balance.locked.as_f64() - 50.028).abs() < 1e-6);
1859
1860        // Free should be total - locked = 51,333.82 - 50.028 = 51,283.79
1861        assert!((usdt_balance.free.as_f64() - 51283.79743837).abs() < 1e-6);
1862
1863        // The bug would have calculated: locked = total - availableToWithdraw = 51,333.82 - 0 = 51,333.82 (all locked!)
1864        // This test verifies that we now correctly use totalOrderIM instead of deriving from availableToWithdraw
1865
1866        // The small order reserves 50.028 USDT of initial margin via `totalOrderIM`,
1867        // so the account-wide USDT margin must be populated even with no open position.
1868        assert_eq!(state.margins.len(), 1);
1869        let usdt_margin = &state.margins[0];
1870        assert!(usdt_margin.instrument_id.is_none());
1871        assert_eq!(usdt_margin.currency.code, "USDT");
1872        assert!((usdt_margin.initial.as_f64() - 50.028).abs() < 1e-6);
1873        assert!(usdt_margin.maintenance.as_f64().abs() < 1e-9);
1874    }
1875
1876    #[rstest]
1877    fn parse_ticker_linear_into_funding_rate() {
1878        let instrument = linear_instrument();
1879        let json = load_test_json("ws_ticker_linear.json");
1880        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
1881
1882        let ts_event = UnixNanos::new(1_673_272_861_686_000_000);
1883
1884        let funding =
1885            parse_ticker_linear_funding(&msg.data, instrument.id(), ts_event, TS).unwrap();
1886
1887        assert_eq!(funding.instrument_id, instrument.id());
1888        assert_eq!(funding.rate, dec!(-0.000212)); // -0.000212
1889        assert_eq!(funding.interval, Some(8 * 60));
1890        assert_eq!(
1891            funding.next_funding_ns,
1892            Some(UnixNanos::new(1_673_280_000_000_000_000))
1893        );
1894        assert_eq!(funding.ts_event, ts_event);
1895        assert_eq!(funding.ts_init, TS);
1896    }
1897
1898    #[rstest]
1899    fn parse_ticker_linear_into_mark_price() {
1900        let instrument = linear_instrument();
1901        let json = load_test_json("ws_ticker_linear.json");
1902        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
1903
1904        let ts_event = UnixNanos::new(1_673_272_861_686_000_000);
1905
1906        let mark_price =
1907            parse_ticker_linear_mark_price(&msg.data, &instrument, ts_event, TS).unwrap();
1908
1909        assert_eq!(mark_price.instrument_id, instrument.id());
1910        assert_eq!(mark_price.value, instrument.make_price(17217.33));
1911        assert_eq!(mark_price.ts_event, ts_event);
1912        assert_eq!(mark_price.ts_init, TS);
1913    }
1914
1915    #[rstest]
1916    fn parse_ticker_linear_into_index_price() {
1917        let instrument = linear_instrument();
1918        let json = load_test_json("ws_ticker_linear.json");
1919        let msg: BybitWsTickerLinearMsg = serde_json::from_str(&json).unwrap();
1920
1921        let ts_event = UnixNanos::new(1_673_272_861_686_000_000);
1922
1923        let index_price =
1924            parse_ticker_linear_index_price(&msg.data, &instrument, ts_event, TS).unwrap();
1925
1926        assert_eq!(index_price.instrument_id, instrument.id());
1927        assert_eq!(index_price.value, instrument.make_price(17227.36));
1928        assert_eq!(index_price.ts_event, ts_event);
1929        assert_eq!(index_price.ts_init, TS);
1930    }
1931
1932    #[rstest]
1933    fn parse_ticker_option_into_mark_price() {
1934        let instrument = option_instrument();
1935        let json = load_test_json("ws_ticker_option.json");
1936        let msg: BybitWsTickerOptionMsg = serde_json::from_str(&json).unwrap();
1937
1938        let mark_price = parse_ticker_option_mark_price(&msg, &instrument, TS).unwrap();
1939
1940        assert_eq!(mark_price.instrument_id, instrument.id());
1941        assert_eq!(mark_price.value, instrument.make_price(7.86976724));
1942        assert_eq!(mark_price.ts_init, TS);
1943    }
1944
1945    #[rstest]
1946    fn parse_ticker_option_into_index_price() {
1947        let instrument = option_instrument();
1948        let json = load_test_json("ws_ticker_option.json");
1949        let msg: BybitWsTickerOptionMsg = serde_json::from_str(&json).unwrap();
1950
1951        let index_price = parse_ticker_option_index_price(&msg, &instrument, TS).unwrap();
1952
1953        assert_eq!(index_price.instrument_id, instrument.id());
1954        assert_eq!(index_price.value, instrument.make_price(16823.73));
1955        assert_eq!(index_price.ts_init, TS);
1956    }
1957
1958    #[rstest]
1959    fn parse_ws_order_stop_market_sell_preserves_type() {
1960        let instrument = linear_instrument();
1961        let json = load_test_json("ws_account_order_stop_market.json");
1962        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1963            serde_json::from_str(&json).unwrap();
1964        let order = &msg.data[0];
1965        let account_id = AccountId::new("BYBIT-001");
1966
1967        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
1968
1969        // Verify sell StopMarket: orderType=Market + stopOrderType=Stop + triggerDirection=2 (falls to)
1970        assert_eq!(report.order_type, OrderType::StopMarket);
1971        assert_eq!(report.order_side, OrderSide::Sell.into());
1972        assert_eq!(report.order_status, OrderStatus::Accepted); // Untriggered maps to Accepted
1973        assert_eq!(report.trigger_price, Some(instrument.make_price(45000.00)));
1974        assert_eq!(report.trigger_type, Some(TriggerType::LastPrice));
1975        assert_eq!(
1976            report.client_order_id.as_ref().unwrap().to_string(),
1977            "test-client-stop-market-001"
1978        );
1979    }
1980
1981    #[rstest]
1982    fn parse_ws_order_stop_market_buy_preserves_type() {
1983        let instrument = linear_instrument();
1984        let json = load_test_json("ws_account_order_buy_stop_market.json");
1985        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
1986            serde_json::from_str(&json).unwrap();
1987        let order = &msg.data[0];
1988        let account_id = AccountId::new("BYBIT-001");
1989
1990        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
1991
1992        // Verify buy StopMarket: orderType=Market + stopOrderType=Stop + triggerDirection=1 (rises to)
1993        assert_eq!(report.order_type, OrderType::StopMarket);
1994        assert_eq!(report.order_side, OrderSide::Buy.into());
1995        assert_eq!(report.order_status, OrderStatus::Accepted);
1996        assert_eq!(report.trigger_price, Some(instrument.make_price(55000.00)));
1997        assert_eq!(report.trigger_type, Some(TriggerType::LastPrice));
1998        assert_eq!(
1999            report.client_order_id.as_ref().unwrap().to_string(),
2000            "test-client-buy-stop-market-001"
2001        );
2002    }
2003
2004    #[rstest]
2005    fn parse_ws_order_market_if_touched_buy_preserves_type() {
2006        let instrument = linear_instrument();
2007        let json = load_test_json("ws_account_order_market_if_touched.json");
2008        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
2009            serde_json::from_str(&json).unwrap();
2010        let order = &msg.data[0];
2011        let account_id = AccountId::new("BYBIT-001");
2012
2013        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
2014
2015        // Verify buy MIT: orderType=Market + stopOrderType=Stop + triggerDirection=2 (falls to)
2016        assert_eq!(report.order_type, OrderType::MarketIfTouched);
2017        assert_eq!(report.order_side, OrderSide::Buy.into());
2018        assert_eq!(report.order_status, OrderStatus::Accepted); // Untriggered maps to Accepted
2019        assert_eq!(report.trigger_price, Some(instrument.make_price(55000.00)));
2020        assert_eq!(report.trigger_type, Some(TriggerType::LastPrice));
2021        assert_eq!(
2022            report.client_order_id.as_ref().unwrap().to_string(),
2023            "test-client-mit-001"
2024        );
2025    }
2026
2027    #[rstest]
2028    fn parse_ws_order_market_if_touched_sell_preserves_type() {
2029        let instrument = linear_instrument();
2030        let json = load_test_json("ws_account_order_sell_market_if_touched.json");
2031        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
2032            serde_json::from_str(&json).unwrap();
2033        let order = &msg.data[0];
2034        let account_id = AccountId::new("BYBIT-001");
2035
2036        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
2037
2038        // Verify sell MIT: orderType=Market + stopOrderType=Stop + triggerDirection=1 (rises to)
2039        assert_eq!(report.order_type, OrderType::MarketIfTouched);
2040        assert_eq!(report.order_side, OrderSide::Sell.into());
2041        assert_eq!(report.order_status, OrderStatus::Accepted);
2042        assert_eq!(report.trigger_price, Some(instrument.make_price(55000.00)));
2043        assert_eq!(
2044            report.client_order_id.as_ref().unwrap().to_string(),
2045            "test-client-sell-mit-001"
2046        );
2047    }
2048
2049    #[rstest]
2050    fn parse_ws_order_stop_limit_preserves_type() {
2051        let instrument = linear_instrument();
2052        let json = load_test_json("ws_account_order_stop_limit.json");
2053        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
2054            serde_json::from_str(&json).unwrap();
2055        let order = &msg.data[0];
2056        let account_id = AccountId::new("BYBIT-001");
2057
2058        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
2059
2060        // Verify StopLimit order type is correctly parsed
2061        // orderType=Limit + stopOrderType=Stop + triggerDirection=2 (falls to)
2062        assert_eq!(report.order_type, OrderType::StopLimit);
2063        assert_eq!(report.order_side, OrderSide::Sell.into());
2064        assert_eq!(report.order_status, OrderStatus::Accepted); // Untriggered maps to Accepted
2065        assert_eq!(report.price, Some(instrument.make_price(44500.00)));
2066        assert_eq!(report.trigger_price, Some(instrument.make_price(45000.00)));
2067        assert_eq!(
2068            report.client_order_id.as_ref().unwrap().to_string(),
2069            "test-client-stop-limit-001"
2070        );
2071    }
2072
2073    #[rstest]
2074    fn parse_ws_order_limit_if_touched_preserves_type() {
2075        let instrument = linear_instrument();
2076        let json = load_test_json("ws_account_order_limit_if_touched.json");
2077        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
2078            serde_json::from_str(&json).unwrap();
2079        let order = &msg.data[0];
2080        let account_id = AccountId::new("BYBIT-001");
2081
2082        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
2083
2084        // Verify LimitIfTouched order type is correctly parsed
2085        // orderType=Limit + stopOrderType=Stop + triggerDirection=1 (rises to)
2086        assert_eq!(report.order_type, OrderType::LimitIfTouched);
2087        assert_eq!(report.order_side, OrderSide::Buy.into());
2088        assert_eq!(report.order_status, OrderStatus::Accepted); // Untriggered maps to Accepted
2089        assert_eq!(report.price, Some(instrument.make_price(55500.00)));
2090        assert_eq!(report.trigger_price, Some(instrument.make_price(55000.00)));
2091        assert_eq!(
2092            report.client_order_id.as_ref().unwrap().to_string(),
2093            "test-client-lit-001"
2094        );
2095    }
2096
2097    #[rstest]
2098    fn parse_ws_wallet_clamps_free_to_zero_when_locked_exceeds_total() {
2099        // totalOrderIM (80) + totalPositionIM (40) = 120, which exceeds
2100        // walletBalance (100). Free balance should clamp to zero, not underflow.
2101        let json = load_test_json("ws_account_wallet_locked_exceeds_total.json");
2102        let msg: crate::websocket::messages::BybitWsAccountWalletMsg =
2103            serde_json::from_str(&json).unwrap();
2104        let wallet = &msg.data[0];
2105        let account_id = AccountId::new("BYBIT-UNIFIED");
2106        let ts_event = UnixNanos::new(1_762_960_669_000_000_000);
2107
2108        let state = parse_ws_account_state(wallet, account_id, ts_event, TS).unwrap();
2109
2110        let usdt_balance = &state.balances[0];
2111        assert_eq!(usdt_balance.currency.code, "USDT");
2112        assert!((usdt_balance.total.as_f64() - 100.0).abs() < 1e-6);
2113        // Locked is capped at total to prevent negative free balance
2114        assert!((usdt_balance.locked.as_f64() - 100.0).abs() < 1e-6);
2115        assert_eq!(usdt_balance.free.as_f64(), 0.0);
2116    }
2117
2118    #[rstest]
2119    fn parse_ws_order_take_profit_maps_to_market_if_touched() {
2120        let instrument = linear_instrument();
2121        let json = load_test_json("ws_account_order_take_profit.json");
2122        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
2123            serde_json::from_str(&json).unwrap();
2124        let order = &msg.data[0];
2125        let account_id = AccountId::new("BYBIT-001");
2126
2127        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
2128
2129        assert_eq!(report.order_type, OrderType::MarketIfTouched);
2130        assert_eq!(report.order_side, OrderSide::Sell.into());
2131        assert_eq!(report.trigger_price, Some(instrument.make_price(55000.00)));
2132        assert_eq!(report.trigger_type, Some(TriggerType::LastPrice));
2133        assert!(report.reduce_only);
2134    }
2135
2136    #[rstest]
2137    fn parse_ws_order_stop_loss_maps_to_stop_market() {
2138        let instrument = linear_instrument();
2139        let json = load_test_json("ws_account_order_stop_loss.json");
2140        let msg: crate::websocket::messages::BybitWsAccountOrderMsg =
2141            serde_json::from_str(&json).unwrap();
2142        let order = &msg.data[0];
2143        let account_id = AccountId::new("BYBIT-001");
2144
2145        let report = parse_ws_order_status_report(order, &instrument, account_id, TS).unwrap();
2146
2147        assert_eq!(report.order_type, OrderType::StopMarket);
2148        assert_eq!(report.order_side, OrderSide::Sell.into());
2149        assert_eq!(report.trigger_price, Some(instrument.make_price(48000.00)));
2150        assert_eq!(report.trigger_type, Some(TriggerType::LastPrice));
2151        assert!(report.reduce_only);
2152    }
2153}