bybit/
account_state.rs

1//! Local copy of a user's account: open orders, positions and wallet balances.
2//!
3//! Every SDK user who trades needs the same bookkeeping: load REST snapshots,
4//! then keep them current with the private `order`, `position` and `wallet`
5//! WebSocket topics, without letting a late or out-of-order message overwrite
6//! newer data. [`AccountState`] does exactly that.
7//!
8//! # Typical flow
9//! 1. Connect the private stream, authenticate and subscribe to `order`,
10//!    `position` and `wallet`. Feed every [`TopicMessage`] into
11//!    [`AccountState::apply`] from now on.
12//! 2. Load the REST snapshots (`get_open_closed_orders_all`,
13//!    `get_position_info_all`, `get_wallet_balance`) and pass them to the
14//!    `set_*` methods together with the response `time`.
15//! 3. After every reconnect repeat step 2: messages missed while
16//!    disconnected are only recovered by a fresh snapshot.
17//!
18//! Steps 1 and 2 may overlap: stream messages and snapshots are merged by
19//! `updated_time`, so the newer version of an order or position wins
20//! regardless of arrival order.
21//!
22//! Executions (`execution`, `execution.fast`) are a feed of events, not state,
23//! and are ignored by [`AccountState::apply`].
24
25use std::collections::{HashMap, HashSet, VecDeque};
26
27use crate::{
28    AccountType, Category, Error, PositionIdx, Timestamp,
29    http::{
30        Client, GetOpenClosedOrdersParams, GetPositionInfoParams, GetWalletBalanceParams, Order,
31        Position, WalletBalance,
32    },
33    ws::{OrderMsg, PositionMsg, TopicMessage, WalletMsg},
34};
35
36/// What [`AccountState::load`] and [`AccountState::reload`] fetch.
37///
38/// ```
39/// use bybit::{AccountScope, AccountType};
40///
41/// // USDT perpetuals, spot orders and the unified wallet.
42/// let scope = AccountScope::new()
43///     .linear("USDT")
44///     .spot()
45///     .wallet(AccountType::UNIFIED);
46/// ```
47#[derive(Debug, Clone, Default)]
48pub struct AccountScope {
49    categories: Vec<(Category, Option<String>)>,
50    wallet: Option<AccountType>,
51}
52
53impl AccountScope {
54    /// An empty scope; add categories and the wallet with the builder methods.
55    pub fn new() -> Self {
56        Self::default()
57    }
58
59    /// Linear contracts settled in `settle_coin` (Bybit requires a settle
60    /// coin to list all linear orders and positions), e.g. "USDT" or "USDC".
61    pub fn linear(self, settle_coin: impl Into<String>) -> Self {
62        self.category(Category::Linear, Some(settle_coin.into()))
63    }
64
65    /// Inverse contracts (all settle coins).
66    pub fn inverse(self) -> Self {
67        self.category(Category::Inverse, None)
68    }
69
70    /// Spot open orders (spot has no positions).
71    pub fn spot(self) -> Self {
72        self.category(Category::Spot, None)
73    }
74
75    /// Options.
76    pub fn option(self) -> Self {
77        self.category(Category::Option, None)
78    }
79
80    /// Any category, with an optional settle coin filter.
81    pub fn category(mut self, category: Category, settle_coin: Option<String>) -> Self {
82        self.categories.push((category, settle_coin));
83        self
84    }
85
86    /// Also load the wallet balance of `account_type`.
87    pub fn wallet(mut self, account_type: AccountType) -> Self {
88        self.wallet = Some(account_type);
89        self
90    }
91}
92
93/// Maximum number of closed orders remembered to reject late stream updates
94/// and stale snapshot entries for them.
95pub const MAX_CLOSED_ORDERS: usize = 10_000;
96
97/// Identifies an order. Order ids are unique only within a category.
98#[derive(Debug, Clone, PartialEq, Eq, Hash)]
99pub struct OrderKey {
100    /// Product type.
101    pub category: Category,
102    /// Bybit order id.
103    pub order_id: String,
104}
105
106/// Identifies a position: one per symbol in one-way mode, two (buy and sell)
107/// in hedge mode.
108#[derive(Debug, Clone, PartialEq, Eq, Hash)]
109pub struct PositionKey {
110    /// Product type.
111    pub category: Category,
112    /// Symbol, e.g. `BTCUSDT`.
113    pub symbol: String,
114    /// One-way (`OneWay`) or one side of hedge mode (`Buy`/`Sell`).
115    pub position_idx: PositionIdx,
116}
117
118/// What [`AccountState::apply`] and friends changed. Updates that were older than
119/// the stored data are ignored and not reported.
120#[derive(Debug)]
121pub enum Change {
122    /// A new open order appeared.
123    OrderOpened(OrderKey),
124    /// An open order changed (e.g. partially filled or amended).
125    OrderUpdated(OrderKey),
126    /// An order reached a final status and was removed. `order` is its final
127    /// state.
128    OrderClosed {
129        /// Key of the closed order.
130        key: OrderKey,
131        /// Final state of the order.
132        order: Box<Order>,
133    },
134    /// A position was created or changed (including closing to size 0).
135    PositionUpdated(PositionKey),
136    /// The balance of an account type was replaced.
137    WalletUpdated(AccountType),
138}
139
140/// Open orders, positions and wallet balances of one account.
141///
142/// Plain data (`Send + Sync`): it can live in any task. It does no I/O.
143#[derive(Debug, Default)]
144pub struct AccountState {
145    wallets: HashMap<AccountType, WalletBalance>,
146    orders: HashMap<OrderKey, Order>,
147    closed_orders: ClosedOrders,
148    positions: HashMap<PositionKey, Position>,
149}
150
151impl AccountState {
152    /// An empty state; fill it with [`set_orders`](Self::set_orders) and friends or use [`load`](Self::load).
153    pub fn new() -> Self {
154        Self::default()
155    }
156
157    // -- Snapshots ----------------------------------------------------------
158
159    /// Create a state from REST snapshots of everything in `scope`.
160    pub async fn load(client: &Client, scope: &AccountScope) -> Result<Self, Error> {
161        let mut state = Self::new();
162        state.reload(client, scope).await?;
163        Ok(state)
164    }
165
166    /// Fetch fresh REST snapshots of everything in `scope` and merge them
167    /// (see [`set_orders`](Self::set_orders) and
168    /// [`set_positions`](Self::set_positions)). Call it after every
169    /// (re)connect of the private stream and after
170    /// [`Event::Lagged`](crate::ws::Event::Lagged).
171    ///
172    /// The snapshot time is the server clock of `client` before the first
173    /// request, so stream updates made meanwhile are kept. On error the
174    /// categories loaded so far stay merged.
175    pub async fn reload(&mut self, client: &Client, scope: &AccountScope) -> Result<(), Error> {
176        let snapshot_time = client.server_timestamp();
177
178        for (category, settle_coin) in &scope.categories {
179            let mut params = GetOpenClosedOrdersParams::new(*category);
180            if let Some(coin) = settle_coin {
181                params = params.with_settle_coin(coin.clone());
182            }
183            let orders = client.get_open_closed_orders_all(&params).await?;
184            self.set_orders(*category, orders, snapshot_time);
185
186            if *category != Category::Spot {
187                let mut params = GetPositionInfoParams::new(*category);
188                if let Some(coin) = settle_coin {
189                    params = params.with_settle_coin(coin.clone());
190                }
191                let positions = client.get_position_info_all(&params).await?;
192                self.set_positions(*category, positions, snapshot_time);
193            }
194        }
195
196        if let Some(account_type) = scope.wallet {
197            let params = GetWalletBalanceParams::new(account_type);
198            for balance in client.get_wallet_balance(&params).await?.result.list {
199                self.set_wallet(balance);
200            }
201        }
202        Ok(())
203    }
204
205    /// Replace the balance of `balance.account_type` with a REST snapshot.
206    pub fn set_wallet(&mut self, balance: WalletBalance) {
207        self.wallets.insert(balance.account_type, balance);
208    }
209
210    /// Merge a REST snapshot of the open orders of `category`.
211    ///
212    /// `snapshot_time` is the server time of the response
213    /// ([`Response::time`](crate::http::Response::time)), milliseconds.
214    /// - an order from the snapshot replaces the stored one unless the stored
215    ///   one is newer, and is skipped if it was already closed by the stream;
216    /// - a stored order missing from the snapshot is removed, unless it was
217    ///   updated after `snapshot_time` (it was opened after the snapshot).
218    pub fn set_orders(&mut self, category: Category, orders: Vec<Order>, snapshot_time: Timestamp) {
219        let mut in_snapshot = HashSet::with_capacity(orders.len());
220        for order in orders {
221            if !order.is_open_status() {
222                continue;
223            }
224            let key = OrderKey {
225                category,
226                order_id: order.order_id.clone(),
227            };
228            in_snapshot.insert(key.clone());
229            if self.closed_orders.closed_since(&key, order.updated_time) {
230                continue;
231            }
232            match self.orders.get(&key) {
233                Some(stored) if stored.updated_time > order.updated_time => {}
234                _ => {
235                    self.orders.insert(key, order);
236                }
237            }
238        }
239
240        self.orders.retain(|key, order| {
241            key.category != category
242                || in_snapshot.contains(key)
243                || order.updated_time > snapshot_time
244        });
245    }
246
247    /// Merge a REST snapshot of the positions of `category`.
248    ///
249    /// A snapshot position replaces the stored one unless the stored one is
250    /// newer. Stored positions missing from the snapshot are kept only if
251    /// they were updated after `snapshot_time`.
252    pub fn set_positions(
253        &mut self,
254        category: Category,
255        positions: Vec<Position>,
256        snapshot_time: Timestamp,
257    ) {
258        let mut in_snapshot = HashSet::with_capacity(positions.len());
259        for position in positions {
260            let key = PositionKey {
261                category,
262                symbol: position.symbol.clone(),
263                position_idx: position.position_idx,
264            };
265            in_snapshot.insert(key.clone());
266            match self.positions.get(&key) {
267                Some(stored) if is_newer_position(stored, &position) => {}
268                _ => {
269                    self.positions.insert(key, position);
270                }
271            }
272        }
273
274        self.positions.retain(|key, position| {
275            key.category != category
276                || in_snapshot.contains(key)
277                || position.updated_time > snapshot_time
278        });
279    }
280
281    // -- Stream -------------------------------------------------------------
282
283    /// Apply one private stream message. Topics other than `order`,
284    /// `position` and `wallet` are ignored.
285    pub fn apply(&mut self, msg: TopicMessage) -> Vec<Change> {
286        match msg {
287            TopicMessage::Order(msg) => msg
288                .data
289                .into_iter()
290                .filter_map(|order| self.apply_order(order))
291                .collect(),
292            TopicMessage::Position(msg) => msg
293                .data
294                .into_iter()
295                .filter_map(|position| self.apply_position(position))
296                .collect(),
297            TopicMessage::Wallet(msg) => msg
298                .data
299                .into_iter()
300                .map(|wallet| self.apply_wallet(wallet))
301                .collect(),
302            TopicMessage::Execution(_)
303            | TopicMessage::FastExecution(_)
304            | TopicMessage::Greeks(_)
305            | TopicMessage::Dcp(_) => Vec::new(),
306        }
307    }
308
309    /// Apply one `order` stream update. Returns `None` if it was stale.
310    pub fn apply_order(&mut self, msg: OrderMsg) -> Option<Change> {
311        let key = OrderKey {
312            category: msg.category,
313            order_id: msg.order_id.clone(),
314        };
315        if self.closed_orders.closed_since(&key, msg.updated_time) {
316            return None;
317        }
318
319        let open = msg.order_status.is_open();
320        match self.orders.get_mut(&key) {
321            Some(stored) if stored.updated_time > msg.updated_time => None,
322            Some(stored) if open => {
323                stored.update(msg);
324                Some(Change::OrderUpdated(key))
325            }
326            Some(_) => {
327                let mut order = self.orders.remove(&key)?;
328                let updated_time = msg.updated_time;
329                order.update(msg);
330                self.closed_orders.insert(key.clone(), updated_time);
331                Some(Change::OrderClosed {
332                    key,
333                    order: Box::new(order),
334                })
335            }
336            None if open => {
337                self.orders.insert(key.clone(), msg.into());
338                Some(Change::OrderOpened(key))
339            }
340            None => {
341                // Closed before we ever saw it open (e.g. an IOC order).
342                self.closed_orders.insert(key.clone(), msg.updated_time);
343                Some(Change::OrderClosed {
344                    key,
345                    order: Box::new(msg.into()),
346                })
347            }
348        }
349    }
350
351    /// Apply one `position` stream update. Returns `None` if it was stale.
352    pub fn apply_position(&mut self, msg: PositionMsg) -> Option<Change> {
353        let key = PositionKey {
354            category: msg.category,
355            symbol: msg.symbol.clone(),
356            position_idx: msg.position_idx,
357        };
358        let position = Position::from(msg);
359        if let Some(stored) = self.positions.get(&key)
360            && is_newer_position(stored, &position)
361        {
362            return None;
363        }
364        self.positions.insert(key.clone(), position);
365        Some(Change::PositionUpdated(key))
366    }
367
368    /// Apply one `wallet` stream update (always a full balance).
369    pub fn apply_wallet(&mut self, msg: WalletMsg) -> Change {
370        let account_type = msg.account_type;
371        self.wallets.insert(account_type, msg.into());
372        Change::WalletUpdated(account_type)
373    }
374
375    // -- Queries ------------------------------------------------------------
376
377    /// Balance of `account_type`, if loaded or received.
378    pub fn wallet(&self, account_type: AccountType) -> Option<&WalletBalance> {
379        self.wallets.get(&account_type)
380    }
381
382    /// All known wallet balances.
383    pub fn wallets(&self) -> impl Iterator<Item = &WalletBalance> {
384        self.wallets.values()
385    }
386
387    /// Open order by id.
388    pub fn order(&self, category: Category, order_id: &str) -> Option<&Order> {
389        self.orders.get(&OrderKey {
390            category,
391            order_id: order_id.to_owned(),
392        })
393    }
394
395    /// Open order by the client-assigned `orderLinkId`.
396    pub fn order_by_link_id(&self, category: Category, order_link_id: &str) -> Option<&Order> {
397        self.orders.iter().find_map(|(key, order)| {
398            (key.category == category && order.order_link_id.as_deref() == Some(order_link_id))
399                .then_some(order)
400        })
401    }
402
403    /// All open orders.
404    pub fn orders(&self) -> impl Iterator<Item = (&OrderKey, &Order)> {
405        self.orders.iter()
406    }
407
408    /// Open orders of one symbol.
409    pub fn orders_for<'a>(
410        &'a self,
411        category: Category,
412        symbol: &'a str,
413    ) -> impl Iterator<Item = &'a Order> {
414        self.orders
415            .iter()
416            .filter(move |(key, order)| key.category == category && order.symbol == symbol)
417            .map(|(_, order)| order)
418    }
419
420    /// Position by symbol and position index (`OneWay`, or `Buy`/`Sell` in hedge mode).
421    pub fn position(
422        &self,
423        category: Category,
424        symbol: &str,
425        position_idx: PositionIdx,
426    ) -> Option<&Position> {
427        self.positions.get(&PositionKey {
428            category,
429            symbol: symbol.to_owned(),
430            position_idx,
431        })
432    }
433
434    /// All known positions, including closed ones (size 0), which still carry
435    /// the leverage and risk limit settings.
436    pub fn positions(&self) -> impl Iterator<Item = (&PositionKey, &Position)> {
437        self.positions.iter()
438    }
439
440    /// Positions with a non-zero size.
441    pub fn open_positions(&self) -> impl Iterator<Item = (&PositionKey, &Position)> {
442        self.positions
443            .iter()
444            .filter(|(_, position)| !position.size.is_zero())
445    }
446}
447
448/// `true` if `stored` is newer than `incoming` and must be kept.
449///
450/// `updated_time` decides; on a tie the trade sequence `seq` does, when both
451/// are known (Bybit sends `-1` for updates not caused by a trade).
452fn is_newer_position(stored: &Position, incoming: &Position) -> bool {
453    if stored.updated_time != incoming.updated_time {
454        return stored.updated_time > incoming.updated_time;
455    }
456    stored.seq >= 0 && incoming.seq >= 0 && stored.seq > incoming.seq
457}
458
459/// Bounded memory of recently closed orders: key -> `updated_time` of the
460/// closing update. The oldest entries are forgotten first.
461#[derive(Debug, Default)]
462struct ClosedOrders {
463    by_key: HashMap<OrderKey, Timestamp>,
464    order: VecDeque<OrderKey>,
465}
466
467impl ClosedOrders {
468    fn insert(&mut self, key: OrderKey, updated_time: Timestamp) {
469        if self.by_key.insert(key.clone(), updated_time).is_none() {
470            self.order.push_back(key);
471        }
472        while self.order.len() > MAX_CLOSED_ORDERS {
473            if let Some(oldest) = self.order.pop_front() {
474                self.by_key.remove(&oldest);
475            }
476        }
477    }
478
479    /// `true` if the order was closed by an update at least as new as
480    /// `updated_time`, i.e. data with that timestamp is stale.
481    ///
482    /// A conditional order that is triggered (a final status) and then
483    /// continues as a regular order under the same id gets a newer
484    /// `updated_time`, so it is not blocked.
485    fn closed_since(&self, key: &OrderKey, updated_time: Timestamp) -> bool {
486        self.by_key
487            .get(key)
488            .is_some_and(|&closed_at| closed_at >= updated_time)
489    }
490}
491
492#[cfg(test)]
493mod tests {
494    use super::*;
495    use crate::{OrderStatus, ws::IncomingMessage};
496    use rust_decimal::Decimal;
497
498    const LINEAR: Category = Category::Linear;
499
500    fn order_msg(order_id: &str, status: &str, updated_time: Timestamp) -> OrderMsg {
501        let json = format!(
502            r#"{{
503                "category": "linear", "symbol": "BTCUSDT", "orderId": "{order_id}",
504                "orderLinkId": "link-{order_id}", "side": "Buy", "orderType": "Limit",
505                "cancelType": "UNKNOWN", "price": "100", "qty": "2", "orderIv": "",
506                "timeInForce": "GTC", "orderStatus": "{status}", "lastPriceOnCreated": "",
507                "reduceOnly": false, "leavesQty": "", "leavesValue": "", "cumExecQty": "0",
508                "cumExecValue": "0", "avgPrice": "", "blockTradeId": "", "positionIdx": 0,
509                "cumExecFee": "0", "closedPnl": "0", "createdTime": "1000",
510                "updatedTime": "{updated_time}", "rejectReason": "EC_NoError",
511                "stopOrderType": "", "tpslMode": "", "triggerPrice": "", "takeProfit": "",
512                "stopLoss": "", "tpTriggerBy": "", "slTriggerBy": "", "tpLimitPrice": "",
513                "slLimitPrice": "", "triggerDirection": 0, "triggerBy": "",
514                "closeOnTrigger": false, "placeType": "", "smpType": "None", "smpGroup": 0,
515                "smpOrderId": "", "feeCurrency": ""
516            }}"#
517        );
518        serde_json::from_str(&json).unwrap()
519    }
520
521    fn order(order_id: &str, status: &str, updated_time: Timestamp) -> Order {
522        order_msg(order_id, status, updated_time).into()
523    }
524
525    fn position_msg(idx: u8, size: &str, updated_time: Timestamp, seq: i64) -> PositionMsg {
526        let json = format!(
527            r#"{{
528                "category": "linear", "symbol": "BTCUSDT", "positionIdx": {idx},
529                "riskId": 1, "riskLimitValue": "200000", "side": "Buy", "size": "{size}",
530                "entryPrice": "100", "sessionAvgPrice": "", "leverage": "10",
531                "positionValue": "100", "markPrice": "100", "positionIM": "10",
532                "positionMM": "1", "positionIMByMp": "10", "positionMMByMp": "1",
533                "takeProfit": "0", "stopLoss": "0", "trailingStop": "0",
534                "unrealisedPnl": "5", "cumRealisedPnl": "0", "curRealisedPnl": "0",
535                "createdTime": "1000", "updatedTime": "{updated_time}", "liqPrice": "",
536                "positionStatus": "Normal", "adlRankIndicator": 0, "autoAddMargin": 0,
537                "leverageSysUpdatedTime": "", "mmrSysUpdatedTime": "", "seq": {seq},
538                "isReduceOnly": false
539            }}"#
540        );
541        serde_json::from_str(&json).unwrap()
542    }
543
544    fn topic_message(json: &str) -> TopicMessage {
545        match IncomingMessage::from_json(json).unwrap() {
546            IncomingMessage::Topic(msg) => msg,
547            other => panic!("not a private topic message: {other:?}"),
548        }
549    }
550
551    #[test]
552    fn account_state_is_send_and_sync() {
553        fn assert_send_sync<T: Send + Sync>() {}
554        assert_send_sync::<AccountState>();
555    }
556
557    #[test]
558    fn order_lifecycle_open_update_close() {
559        let mut state = AccountState::new();
560
561        let opened = state.apply_order(order_msg("1", "New", 1));
562        let updated = state.apply_order(order_msg("1", "PartiallyFilled", 2));
563        let closed = state.apply_order(order_msg("1", "Filled", 3));
564
565        assert!(matches!(opened, Some(Change::OrderOpened(_))));
566        assert!(matches!(updated, Some(Change::OrderUpdated(_))));
567        assert!(matches!(
568            closed,
569            Some(Change::OrderClosed { ref order, .. }) if order.order_status == OrderStatus::Filled
570        ));
571        assert_eq!(state.orders().count(), 0);
572    }
573
574    #[test]
575    fn open_order_is_queryable() {
576        let mut state = AccountState::new();
577        state.apply_order(order_msg("1", "New", 1));
578
579        assert!(state.order(LINEAR, "1").is_some());
580        assert!(state.order(Category::Spot, "1").is_none());
581        assert!(state.order_by_link_id(LINEAR, "link-1").is_some());
582        assert_eq!(state.orders_for(LINEAR, "BTCUSDT").count(), 1);
583        // leavesQty is derived when the stream omits it.
584        assert_eq!(
585            state.order(LINEAR, "1").unwrap().leaves_qty,
586            Decimal::from(2)
587        );
588    }
589
590    #[test]
591    fn stale_order_update_is_ignored() {
592        let mut state = AccountState::new();
593        state.apply_order(order_msg("1", "PartiallyFilled", 5));
594
595        let change = state.apply_order(order_msg("1", "New", 4));
596
597        assert!(change.is_none());
598        assert_eq!(
599            state.order(LINEAR, "1").unwrap().order_status,
600            OrderStatus::PartiallyFilled
601        );
602    }
603
604    #[test]
605    fn late_update_after_close_does_not_reopen_the_order() {
606        let mut state = AccountState::new();
607        state.apply_order(order_msg("1", "New", 1));
608        state.apply_order(order_msg("1", "Filled", 3));
609
610        let change = state.apply_order(order_msg("1", "PartiallyFilled", 2));
611
612        assert!(change.is_none());
613        assert!(state.order(LINEAR, "1").is_none());
614    }
615
616    #[test]
617    fn order_closed_before_being_seen_open_is_reported_but_not_stored() {
618        let mut state = AccountState::new();
619
620        let change = state.apply_order(order_msg("1", "Cancelled", 1));
621
622        assert!(matches!(change, Some(Change::OrderClosed { .. })));
623        assert!(state.order(LINEAR, "1").is_none());
624    }
625
626    #[test]
627    fn triggered_conditional_order_continues_under_the_same_id() {
628        let mut state = AccountState::new();
629        state.apply_order(order_msg("1", "Untriggered", 1));
630        state.apply_order(order_msg("1", "Triggered", 2));
631
632        let change = state.apply_order(order_msg("1", "New", 3));
633
634        assert!(matches!(change, Some(Change::OrderOpened(_))));
635        assert!(state.order(LINEAR, "1").is_some());
636    }
637
638    #[test]
639    fn snapshot_does_not_resurrect_an_order_closed_by_the_stream() {
640        let mut state = AccountState::new();
641        state.apply_order(order_msg("1", "Filled", 5));
642
643        // The snapshot was taken before the fill.
644        state.set_orders(LINEAR, vec![order("1", "New", 4)], 4);
645
646        assert!(state.order(LINEAR, "1").is_none());
647    }
648
649    #[test]
650    fn snapshot_does_not_overwrite_newer_stream_data() {
651        let mut state = AccountState::new();
652        state.apply_order(order_msg("1", "PartiallyFilled", 5));
653
654        state.set_orders(LINEAR, vec![order("1", "New", 4)], 4);
655
656        assert_eq!(
657            state.order(LINEAR, "1").unwrap().order_status,
658            OrderStatus::PartiallyFilled
659        );
660    }
661
662    #[test]
663    fn snapshot_removes_missing_orders_but_keeps_orders_opened_after_it() {
664        let mut state = AccountState::new();
665        state.apply_order(order_msg("old", "New", 1));
666        state.apply_order(order_msg("new", "New", 20));
667        state.apply_order(order_msg("spot-is-untouched", "New", 1));
668
669        state.set_orders(LINEAR, vec![order("snap", "New", 5)], 10);
670
671        assert!(state.order(LINEAR, "old").is_none());
672        assert!(state.order(LINEAR, "new").is_some());
673        assert!(state.order(LINEAR, "snap").is_some());
674    }
675
676    #[test]
677    fn snapshot_skips_closed_orders() {
678        let mut state = AccountState::new();
679
680        state.set_orders(LINEAR, vec![order("1", "Filled", 5)], 10);
681
682        assert_eq!(state.orders().count(), 0);
683    }
684
685    #[test]
686    fn hedge_mode_positions_are_separate() {
687        let mut state = AccountState::new();
688
689        state.apply_position(position_msg(1, "1", 1, 1));
690        state.apply_position(position_msg(2, "3", 1, 2));
691
692        let buy = state.position(LINEAR, "BTCUSDT", PositionIdx::Buy).unwrap();
693        let sell = state
694            .position(LINEAR, "BTCUSDT", PositionIdx::Sell)
695            .unwrap();
696        assert_eq!(buy.size, Decimal::from(1));
697        assert_eq!(sell.size, Decimal::from(3));
698        // Per-position margin and PnL come from the position message itself.
699        assert_eq!(buy.position_im, Some(Decimal::from(10)));
700        assert_eq!(buy.position_mm, Some(Decimal::from(1)));
701        assert_eq!(buy.unrealised_pnl, Some(Decimal::from(5)));
702    }
703
704    #[test]
705    fn stale_position_update_is_ignored() {
706        let mut state = AccountState::new();
707        state.apply_position(position_msg(0, "2", 5, 10));
708
709        assert!(state.apply_position(position_msg(0, "1", 4, 11)).is_none());
710        // Same time: the trade sequence decides.
711        assert!(state.apply_position(position_msg(0, "1", 5, 9)).is_none());
712        assert!(state.apply_position(position_msg(0, "3", 5, 12)).is_some());
713
714        let position = state
715            .position(LINEAR, "BTCUSDT", PositionIdx::OneWay)
716            .unwrap();
717        assert_eq!(position.size, Decimal::from(3));
718    }
719
720    #[test]
721    fn closed_position_is_kept_but_not_open() {
722        let mut state = AccountState::new();
723        state.apply_position(position_msg(0, "2", 1, 1));
724
725        state.apply_position(position_msg(0, "0", 2, 2));
726
727        assert_eq!(state.positions().count(), 1);
728        assert_eq!(state.open_positions().count(), 0);
729    }
730
731    #[test]
732    fn position_snapshot_merges_by_updated_time() {
733        let mut state = AccountState::new();
734        state.apply_position(position_msg(1, "5", 20, 1));
735        state.apply_position(position_msg(2, "5", 1, 1));
736
737        state.set_positions(
738            LINEAR,
739            vec![Position::from(position_msg(1, "1", 10, 1))],
740            10,
741        );
742
743        // Buy side: the stream version is newer than the snapshot entry.
744        let buy = state.position(LINEAR, "BTCUSDT", PositionIdx::Buy).unwrap();
745        assert_eq!(buy.size, Decimal::from(5));
746        // Sell side: missing from the snapshot and older than it.
747        assert!(
748            state
749                .position(LINEAR, "BTCUSDT", PositionIdx::Sell)
750                .is_none()
751        );
752    }
753
754    #[test]
755    fn apply_dispatches_order_stream_messages() {
756        let json = r#"{
757            "id": "5923240c6880ab-c59f-420b-9adb-3639adc9dd90",
758            "topic": "order",
759            "creationTime": 1672364262474,
760            "data": [
761                {
762                    "symbol": "ETH-30DEC22-1400-C",
763                    "orderId": "5cf98598-39a7-459e-97bf-76ca765ee020",
764                    "side": "Sell",
765                    "orderType": "Market",
766                    "cancelType": "UNKNOWN",
767                    "price": "72.5",
768                    "qty": "1",
769                    "orderIv": "",
770                    "timeInForce": "IOC",
771                    "orderStatus": "Filled",
772                    "orderLinkId": "",
773                    "lastPriceOnCreated": "",
774                    "reduceOnly": false,
775                    "leavesQty": "",
776                    "leavesValue": "",
777                    "cumExecQty": "1",
778                    "cumExecValue": "75",
779                    "avgPrice": "75",
780                    "blockTradeId": "",
781                    "positionIdx": 0,
782                    "cumExecFee": "0.358635",
783                    "closedPnl": "0",
784                    "createdTime": "1672364262444",
785                    "updatedTime": "1672364262457",
786                    "rejectReason": "EC_NoError",
787                    "stopOrderType": "",
788                    "tpslMode": "",
789                    "triggerPrice": "",
790                    "takeProfit": "",
791                    "stopLoss": "",
792                    "tpTriggerBy": "",
793                    "slTriggerBy": "",
794                    "tpLimitPrice": "",
795                    "slLimitPrice": "",
796                    "triggerDirection": 0,
797                    "triggerBy": "",
798                    "closeOnTrigger": false,
799                    "category": "option",
800                    "placeType": "price",
801                    "smpType": "None",
802                    "smpGroup": 0,
803                    "smpOrderId": "",
804                    "feeCurrency": "",
805                    "cumFeeDetail": {
806                        "MNT": "0.00242968"
807                    }
808                }
809            ]
810        }"#;
811        let mut state = AccountState::new();
812
813        let changes = state.apply(topic_message(json));
814
815        // An IOC option order filled at once: reported as closed, not stored.
816        assert!(matches!(
817            changes.as_slice(),
818            [Change::OrderClosed { key, .. }] if key.category == Category::Option
819        ));
820        assert_eq!(state.orders().count(), 0);
821    }
822
823    #[test]
824    fn apply_dispatches_wallet_stream_messages() {
825        let json = r#"{
826            "id": "592324d2bce751-ad38-48eb-8f42-4671d1fb4d4e",
827            "topic": "wallet",
828            "creationTime": 1700034722104,
829            "data": [
830                {
831                    "accountIMRate": "0",
832                    "accountIMRateByMp": "0",
833                    "accountMMRate": "0",
834                    "accountMMRateByMp": "0",
835                    "totalEquity": "10262.91335023",
836                    "totalWalletBalance": "9684.46297164",
837                    "totalMarginBalance": "9684.46297164",
838                    "totalAvailableBalance": "9556.6056555",
839                    "totalPerpUPL": "0",
840                    "totalInitialMargin": "0",
841                    "totalInitialMarginByMp": "0",
842                    "totalMaintenanceMargin": "0",
843                    "totalMaintenanceMarginByMp": "0",
844                    "coin": [
845                        {
846                            "coin": "BTC",
847                            "equity": "0.00102964",
848                            "usdValue": "36.70759517",
849                            "walletBalance": "0.00102964",
850                            "availableToWithdraw": "0.00102964",
851                            "availableToBorrow": "",
852                            "borrowAmount": "0",
853                            "accruedInterest": "0",
854                            "totalOrderIM": "",
855                            "totalPositionIM": "",
856                            "totalPositionMM": "",
857                            "unrealisedPnl": "0",
858                            "cumRealisedPnl": "-0.00000973",
859                            "bonus": "0",
860                            "collateralSwitch": true,
861                            "marginCollateral": true,
862                            "locked": "0",
863                            "spotHedgingQty": "0.01592413",
864                            "spotBorrow": "0"
865                        }
866                    ],
867                    "accountLTV": "0",
868                    "accountType": "UNIFIED"
869                }
870            ]
871        }"#;
872        let mut state = AccountState::new();
873
874        let changes = state.apply(topic_message(json));
875
876        assert!(matches!(
877            changes.as_slice(),
878            [Change::WalletUpdated(AccountType::UNIFIED)]
879        ));
880        assert!(state.wallet(AccountType::UNIFIED).is_some());
881    }
882
883    /// HTTP server answering the snapshot endpoints with fixtures; records
884    /// the request targets.
885    async fn spawn_snapshot_server() -> (String, std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
886        use tokio::io::{AsyncReadExt, AsyncWriteExt};
887
888        const ORDERS: &str = r#"{"retCode": 0, "retMsg": "OK", "result": {"list": [{"orderId": "fd4300ae-7847-404e-b947-b46980a4d140", "orderLinkId": "test-000005", "blockTradeId": "", "symbol": "ETHUSDT", "price": "1600.00", "qty": "0.10", "side": "Buy", "isLeverage": "", "positionIdx": 1, "orderStatus": "New", "cancelType": "UNKNOWN", "rejectReason": "EC_NoError", "avgPrice": "0", "leavesQty": "0.10", "leavesValue": "160", "cumExecQty": "0.00", "cumExecValue": "0", "cumExecFee": "0", "timeInForce": "GTC", "orderType": "Limit", "stopOrderType": "UNKNOWN", "orderIv": "", "triggerPrice": "0.00", "takeProfit": "2500.00", "stopLoss": "1500.00", "tpTriggerBy": "LastPrice", "slTriggerBy": "LastPrice", "triggerDirection": 0, "triggerBy": "UNKNOWN", "lastPriceOnCreated": "", "reduceOnly": false, "closeOnTrigger": false, "smpType": "None", "smpGroup": 0, "smpOrderId": "", "tpslMode": "Full", "tpLimitPrice": "", "slLimitPrice": "", "placeType": "", "createdTime": "1684738540559", "updatedTime": "1684738540561"}], "nextPageCursor": "", "category": "linear"}, "retExtInfo": {}, "time": 1684765770483}"#;
889        const POSITIONS: &str = r#"{"retCode": 0, "retMsg": "OK", "result": {"list": [{"positionIdx": 0, "riskId": 1, "riskLimitValue": "150", "symbol": "BTCUSD", "side": "Sell", "size": "300", "avgPrice": "27464.50441675", "positionValue": "0.01092319", "tradeMode": 0, "positionStatus": "Normal", "autoAddMargin": 1, "adlRankIndicator": 2, "leverage": "10", "positionBalance": "0.00139186", "markPrice": "28224.50", "liqPrice": "", "bustPrice": "999999.00", "positionMM": "0.0000015", "positionMMByMp": "0.0000015", "positionIM": "0.00010923", "positionIMByMp": "0.00010923", "tpslMode": "Full", "takeProfit": "0.00", "stopLoss": "0.00", "trailingStop": "0.00", "unrealisedPnl": "-0.00029413", "curRealisedPnl": "0.00013123", "cumRealisedPnl": "-0.00096902", "seq": 5723621632, "isReduceOnly": false, "mmrSysUpdatedTime": "", "leverageSysUpdatedTime": "", "sessionAvgPrice": "", "createdTime": "1676538056258", "updatedTime": "1697673600012"}], "nextPageCursor": "", "category": "inverse"}, "retExtInfo": {}, "time": 1697684980172}"#;
890        const WALLET: &str = r#"{"retCode": 0, "retMsg": "OK", "result": {"list": [{"totalEquity": "3.31216591", "accountIMRate": "0", "accountIMRateByMp": "0", "totalMarginBalance": "3.00326056", "totalInitialMargin": "0", "totalInitialMarginByMp": "0", "accountType": "UNIFIED", "totalAvailableBalance": "3.00326056", "accountMMRate": "0", "accountMMRateByMp": "0", "totalPerpUPL": "0", "totalWalletBalance": "3.00326056", "accountLTV": "0", "totalMaintenanceMargin": "0", "totalMaintenanceMarginByMp": "0", "coin": [{"availableToBorrow": "3", "bonus": "0", "accruedInterest": "0", "availableToWithdraw": "0", "totalOrderIM": "0", "equity": "0", "totalPositionMM": "0", "usdValue": "0", "spotHedgingQty": "0.01592413", "unrealisedPnl": "0", "collateralSwitch": true, "borrowAmount": "0.0", "totalPositionIM": "0", "walletBalance": "0", "cumRealisedPnl": "0", "locked": "0", "marginCollateral": true, "coin": "BTC", "spotBorrow": "0"}]}]}, "retExtInfo": {}, "time": 1690872862481}"#;
891
892        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
893        let addr = listener.local_addr().unwrap();
894        let requests = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
895        let recorded = requests.clone();
896        tokio::spawn(async move {
897            while let Ok((mut tcp, _)) = listener.accept().await {
898                let mut head = Vec::new();
899                let mut buf = [0u8; 1024];
900                while !head.windows(4).any(|w| w == b"\r\n\r\n") {
901                    match tcp.read(&mut buf).await {
902                        Ok(0) | Err(_) => break,
903                        Ok(n) => head.extend_from_slice(&buf[..n]),
904                    }
905                }
906                let head = String::from_utf8_lossy(&head).into_owned();
907                let target = head
908                    .split_whitespace()
909                    .nth(1)
910                    .unwrap_or_default()
911                    .to_owned();
912                recorded.lock().unwrap().push(target.clone());
913                let body = if target.starts_with("/v5/order/realtime") {
914                    ORDERS
915                } else if target.starts_with("/v5/position/list") {
916                    POSITIONS
917                } else {
918                    WALLET
919                };
920                let response = format!(
921                    "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
922                    body.len()
923                );
924                let _ = tcp.write_all(response.as_bytes()).await;
925            }
926        });
927        (format!("http://{addr}"), requests)
928    }
929
930    #[tokio::test]
931    async fn load_fetches_every_part_of_the_scope() {
932        let (url, requests) = spawn_snapshot_server().await;
933        let config = crate::http::Config::new(url).credentials("key", "secret");
934        let client = Client::new(config).unwrap();
935        let scope = AccountScope::new()
936            .linear("USDT")
937            .spot()
938            .wallet(AccountType::UNIFIED);
939
940        let state = AccountState::load(&client, &scope).await.unwrap();
941
942        // The orders fixture is served for both categories.
943        assert_eq!(state.orders().count(), 2);
944        let order_id = "fd4300ae-7847-404e-b947-b46980a4d140";
945        assert!(state.order(Category::Linear, order_id).is_some());
946        assert!(state.order(Category::Spot, order_id).is_some());
947        // Positions only for linear (spot has none).
948        assert_eq!(state.open_positions().count(), 1);
949        assert!(state.wallet(AccountType::UNIFIED).is_some());
950
951        let requests = requests.lock().unwrap().clone();
952        assert_eq!(requests.len(), 4, "{requests:?}");
953        assert!(requests[0].starts_with("/v5/order/realtime"));
954        assert!(requests[0].contains("category=linear") && requests[0].contains("settleCoin=USDT"));
955        assert!(requests[1].starts_with("/v5/position/list"));
956        assert!(requests[1].contains("settleCoin=USDT"));
957        assert!(requests[2].starts_with("/v5/order/realtime"));
958        assert!(requests[2].contains("category=spot"));
959        assert!(requests[3].starts_with("/v5/account/wallet-balance"));
960    }
961}
bybit/account_state.rs

bybit/
account_state.rs
