bybit/ws/
trade.rs

1//! Order entry over WebSocket (`/v5/trade`).
2
3use std::{
4    collections::HashMap,
5    sync::{
6        Arc, Mutex,
7        atomic::{AtomicBool, AtomicU64, Ordering},
8    },
9    time::Duration,
10};
11
12use serde::de::DeserializeOwned;
13use tokio::{
14    sync::{mpsc, oneshot},
15    time::timeout,
16};
17
18use crate::{
19    ServerClock, Timestamp,
20    http::{
21        AmendOrderBatchRequest, AmendOrderBatchResult, AmendOrderRequest, AmendOrderResponse,
22        CancelOrderBatchRequest, CancelOrderBatchResult, CancelOrderRequest, CancelOrderResponse,
23        List, PlaceOrderBatchRequest, PlaceOrderBatchResult, PlaceOrderRequest, PlaceOrderResponse,
24        RetExtInfo,
25    },
26    timestamp,
27    ws::{
28        Config, Error, Event, Handle, IncomingMessage, OutgoingMessage, Stream, TradeHeader,
29        TradeReply, TradeReplyHeader,
30    },
31};
32
33/// A successful reply of the order entry stream.
34///
35/// An accepted request only means Bybit took it: the order stream
36/// (`Topic::Order*`) confirms the order status.
37#[derive(Debug, Clone, PartialEq)]
38pub struct TradeResponse<T> {
39    /// Same as `result` of the REST endpoint.
40    pub data: T,
41    /// Per-item results of batch requests, in request order.
42    pub ret_ext_info: Option<RetExtInfo>,
43    /// Rate limit status of the op.
44    pub header: Option<TradeReplyHeader>,
45}
46
47/// Places, amends and cancels orders over the order entry stream
48/// (`/v5/trade`), matching replies to requests by `reqId`.
49///
50/// Built on [`Stream`]: it authenticates on every (re)connect with the
51/// credentials and [`ServerClock`] of the [`Config`]; requests are accepted
52/// once [`Event::Authenticated`] arrived. Requests in flight when the
53/// connection drops fail with [`Error::ConnectionLost`] and are not resent:
54/// check the order (e.g. by `orderLinkId`) before retrying.
55///
56/// The order entry stream is not available on demo trading; use testnet.
57///
58/// ```no_run
59/// use bybit::{
60///     Category, Environment, OrderType, Side,
61///     http::PlaceOrderRequest,
62///     ws::{self, TradeClient},
63/// };
64/// use rust_decimal::dec;
65///
66/// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
67/// let config = ws::Config::trade(Environment::Testnet)?.credentials("API_KEY", "API_SECRET");
68/// let (trade, mut events) = TradeClient::new(config)?;
69/// trade.connect().await?;
70/// while let Some(event) = events.recv().await {
71///     if matches!(event, ws::Event::Authenticated) {
72///         break;
73///     }
74/// }
75///
76/// let mut order = PlaceOrderRequest::new(
77///     Category::Linear,
78///     "BTCUSDT",
79///     Side::Buy,
80///     OrderType::Limit,
81///     dec!(0.001),
82/// );
83/// order.price = Some(dec!(10000));
84/// let reply = trade.place_order(&order).await?;
85/// println!("order id: {}", reply.data.order_id);
86/// # Ok(())
87/// # }
88/// ```
89#[derive(Debug, Clone)]
90pub struct TradeClient {
91    handle: Handle,
92    shared: Arc<Shared>,
93    clock: ServerClock,
94    recv_window: Timestamp,
95    request_timeout: Duration,
96}
97
98#[derive(Debug, Default)]
99struct Shared {
100    /// Waiting requests by `reqId`.
101    pending: Mutex<HashMap<String, oneshot::Sender<TradeReply>>>,
102    /// Authenticated on the current connection.
103    ready: AtomicBool,
104    next_id: AtomicU64,
105    /// Prefix that keeps `reqId`s unique across clients and restarts.
106    id_prefix: String,
107}
108
109impl Shared {
110    /// Drop every waiting request; their callers get `ConnectionLost`.
111    fn fail_pending(&self) {
112        self.pending.lock().unwrap().clear();
113    }
114}
115
116impl TradeClient {
117    /// Spawn the stream driver (see [`Stream::new`]) and the reply router.
118    /// Returns the client and the events that are not replies to its
119    /// requests. Fails without credentials: the order entry stream requires
120    /// authentication.
121    pub fn new(config: Config) -> Result<(Self, mpsc::Receiver<Event>), Error> {
122        if config.api_key.is_none() || config.api_secret.is_none() {
123            return Err(Error::MissingCredentials);
124        }
125        let clock = config.server_clock.clone();
126        let recv_window = config.auth_recv_window;
127        let request_timeout = config.request_timeout;
128        let event_queue_size = config.event_queue_size;
129
130        let shared = Arc::new(Shared {
131            id_prefix: format!("t{:x}", timestamp()),
132            ..Default::default()
133        });
134        let (handle, driver_events) = Stream::new(config);
135        let (tx, rx) = mpsc::channel(event_queue_size);
136        tokio::spawn(route(driver_events, tx, shared.clone()));
137
138        let client = Self {
139            handle,
140            shared,
141            clock,
142            recv_window,
143            request_timeout,
144        };
145        Ok((client, rx))
146    }
147
148    /// The command handle of the underlying stream.
149    pub fn handle(&self) -> &Handle {
150        &self.handle
151    }
152
153    /// Open the connection; requests are accepted after
154    /// [`Event::Authenticated`].
155    pub async fn connect(&self) -> Result<(), Error> {
156        self.handle.connect().await
157    }
158
159    /// Close the connection.
160    pub async fn disconnect(&self) -> Result<(), Error> {
161        self.handle.disconnect().await
162    }
163
164    /// `true` while connected and authenticated.
165    pub fn is_ready(&self) -> bool {
166        self.shared.ready.load(Ordering::Relaxed)
167    }
168
169    /// Create an order (`order.create`).
170    pub async fn place_order(
171        &self,
172        request: &PlaceOrderRequest,
173    ) -> Result<TradeResponse<PlaceOrderResponse>, Error> {
174        self.call(|req_id, header| OutgoingMessage::OrderCreate {
175            req_id,
176            header,
177            args: vec![request.clone()],
178        })
179        .await
180    }
181
182    /// Amend an order (`order.amend`).
183    pub async fn amend_order(
184        &self,
185        request: &AmendOrderRequest,
186    ) -> Result<TradeResponse<AmendOrderResponse>, Error> {
187        self.call(|req_id, header| OutgoingMessage::OrderAmend {
188            req_id,
189            header,
190            args: vec![request.clone()],
191        })
192        .await
193    }
194
195    /// Cancel an order (`order.cancel`).
196    pub async fn cancel_order(
197        &self,
198        request: &CancelOrderRequest,
199    ) -> Result<TradeResponse<CancelOrderResponse>, Error> {
200        self.call(|req_id, header| OutgoingMessage::OrderCancel {
201            req_id,
202            header,
203            args: vec![request.clone()],
204        })
205        .await
206    }
207
208    /// Create orders in a batch (`order.create-batch`); per-order results are
209    /// in [`TradeResponse::ret_ext_info`].
210    pub async fn place_order_batch(
211        &self,
212        request: &PlaceOrderBatchRequest,
213    ) -> Result<TradeResponse<List<PlaceOrderBatchResult>>, Error> {
214        self.call(|req_id, header| OutgoingMessage::OrderCreateBatch {
215            req_id,
216            header,
217            args: vec![request.clone()],
218        })
219        .await
220    }
221
222    /// Amend orders in a batch (`order.amend-batch`).
223    pub async fn amend_order_batch(
224        &self,
225        request: &AmendOrderBatchRequest,
226    ) -> Result<TradeResponse<List<AmendOrderBatchResult>>, Error> {
227        self.call(|req_id, header| OutgoingMessage::OrderAmendBatch {
228            req_id,
229            header,
230            args: vec![request.clone()],
231        })
232        .await
233    }
234
235    /// Cancel orders in a batch (`order.cancel-batch`).
236    pub async fn cancel_order_batch(
237        &self,
238        request: &CancelOrderBatchRequest,
239    ) -> Result<TradeResponse<List<CancelOrderBatchResult>>, Error> {
240        self.call(|req_id, header| OutgoingMessage::OrderCancelBatch {
241            req_id,
242            header,
243            args: vec![request.clone()],
244        })
245        .await
246    }
247
248    async fn call<T: DeserializeOwned>(
249        &self,
250        build: impl FnOnce(String, TradeHeader) -> OutgoingMessage,
251    ) -> Result<TradeResponse<T>, Error> {
252        if !self.is_ready() {
253            return Err(Error::NotConnected);
254        }
255        let n = self.shared.next_id.fetch_add(1, Ordering::Relaxed);
256        let req_id = format!("{}-{n}", self.shared.id_prefix);
257        let (tx, rx) = oneshot::channel();
258        self.shared
259            .pending
260            .lock()
261            .unwrap()
262            .insert(req_id.clone(), tx);
263
264        let header = TradeHeader {
265            timestamp: self.clock.now().to_string(),
266            recv_window: self.recv_window.to_string(),
267            referer: None,
268        };
269        if let Err(e) = self
270            .handle
271            .send_command(build(req_id.clone(), header))
272            .await
273        {
274            self.shared.pending.lock().unwrap().remove(&req_id);
275            return Err(e);
276        }
277
278        let reply = match timeout(self.request_timeout, rx).await {
279            Ok(Ok(reply)) => reply,
280            Ok(Err(_)) => return Err(Error::ConnectionLost),
281            Err(_) => {
282                self.shared.pending.lock().unwrap().remove(&req_id);
283                return Err(Error::Timeout);
284            }
285        };
286        if reply.ret_code != 0 {
287            return Err(Error::Api {
288                code: reply.ret_code,
289                msg: reply.ret_msg,
290            });
291        }
292        let data = serde_json::from_value(reply.data.unwrap_or_default())
293            .map_err(|e| Error::InvalidReply(e.to_string()))?;
294        let ret_ext_info = reply
295            .ret_ext_info
296            .and_then(|value| serde_json::from_value(value).ok());
297        Ok(TradeResponse {
298            data,
299            ret_ext_info,
300            header: reply.header,
301        })
302    }
303}
304
305/// Deliver replies to waiting requests and forward every other event.
306async fn route(
307    mut driver_events: mpsc::Receiver<Event>,
308    tx: mpsc::Sender<Event>,
309    shared: Arc<Shared>,
310) {
311    while let Some(event) = driver_events.recv().await {
312        match &event {
313            Event::Authenticated => shared.ready.store(true, Ordering::Relaxed),
314            Event::Connected
315            | Event::Reconnecting { .. }
316            | Event::Disconnected { .. }
317            | Event::AuthFailed { .. } => {
318                shared.ready.store(false, Ordering::Relaxed);
319                shared.fail_pending();
320            }
321            Event::Message(IncomingMessage::TradeReply(reply)) => {
322                let waiting = reply
323                    .req_id
324                    .as_ref()
325                    .and_then(|id| shared.pending.lock().unwrap().remove(id));
326                if let Some(waiting) = waiting {
327                    let _ = waiting.send((**reply).clone());
328                    continue;
329                }
330            }
331            _ => {}
332        }
333        // A dropped receiver must not stop the routing of replies.
334        let _ = tx.send(event).await;
335    }
336    shared.ready.store(false, Ordering::Relaxed);
337    shared.fail_pending();
338}
339
340#[cfg(test)]
341mod tests {
342    use std::time::Duration;
343
344    use futures_util::{SinkExt, StreamExt};
345    use rust_decimal::dec;
346    use tokio::net::TcpListener;
347    use tokio_tungstenite::tungstenite::Message;
348
349    use super::*;
350    use crate::{Category, OrderType, Side};
351
352    /// How the fake order entry stream answers `order.*` requests.
353    #[derive(Clone, Copy)]
354    enum Answer {
355        Accept,
356        Reject,
357        Silent,
358        /// Close the connection instead of answering.
359        Drop,
360    }
361
362    /// Received request frames.
363    type Received = Arc<Mutex<Vec<serde_json::Value>>>;
364
365    /// Local server speaking the order entry protocol: `auth` succeeds,
366    /// `ping` gets an order-entry style `pong`, `order.*` per `answer`.
367    async fn spawn_trade_server(answer: Answer) -> (String, Received) {
368        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
369        let addr = listener.local_addr().unwrap();
370        let received: Received = Default::default();
371        let recorded = received.clone();
372        tokio::spawn(async move {
373            while let Ok((tcp, _)) = listener.accept().await {
374                let recorded = recorded.clone();
375                tokio::spawn(async move {
376                    let Ok(mut ws) = tokio_tungstenite::accept_async(tcp).await else {
377                        return;
378                    };
379                    while let Some(Ok(Message::Text(text))) = ws.next().await {
380                        let json: serde_json::Value = serde_json::from_str(&text).unwrap();
381                        recorded.lock().unwrap().push(json.clone());
382                        let op = json["op"].as_str().unwrap_or_default().to_owned();
383                        let reply = match op.as_str() {
384                            "auth" => serde_json::json!({
385                                "retCode": 0, "retMsg": "OK", "op": "auth", "connId": "c"
386                            }),
387                            "ping" => serde_json::json!({
388                                "retCode": 0, "retMsg": "OK", "op": "pong", "connId": "c"
389                            }),
390                            _ => match answer {
391                                Answer::Accept => serde_json::json!({
392                                    "reqId": json["reqId"], "retCode": 0, "retMsg": "OK",
393                                    "op": op, "data": {"orderId": "order-1", "orderLinkId": ""},
394                                    "retExtInfo": {},
395                                    "header": {"X-Bapi-Limit": "10", "X-Bapi-Limit-Status": "9"},
396                                    "connId": "c"
397                                }),
398                                Answer::Reject => serde_json::json!({
399                                    "reqId": json["reqId"], "retCode": 10001,
400                                    "retMsg": "params error", "op": op, "connId": "c"
401                                }),
402                                Answer::Silent => continue,
403                                Answer::Drop => return,
404                            },
405                        };
406                        if ws
407                            .send(Message::Text(reply.to_string().into()))
408                            .await
409                            .is_err()
410                        {
411                            return;
412                        }
413                    }
414                });
415            }
416        });
417        (format!("ws://{addr}"), received)
418    }
419
420    fn config(url: String) -> Config {
421        Config::new(url)
422            .credentials("key", "secret")
423            .ping_interval(None)
424            .max_reconnect_attempts(0)
425            .request_timeout(Duration::from_millis(300))
426    }
427
428    /// Connect and wait until requests are accepted.
429    async fn connected(config: Config) -> (TradeClient, mpsc::Receiver<Event>) {
430        let (trade, mut events) = TradeClient::new(config).unwrap();
431        trade.connect().await.unwrap();
432        timeout(Duration::from_secs(10), async {
433            while let Some(event) = events.recv().await {
434                if matches!(event, Event::Authenticated) {
435                    return;
436                }
437            }
438        })
439        .await
440        .expect("not authenticated");
441        assert!(trade.is_ready());
442        (trade, events)
443    }
444
445    fn order() -> PlaceOrderRequest {
446        let mut order = PlaceOrderRequest::new(
447            Category::Linear,
448            "BTCUSDT",
449            Side::Buy,
450            OrderType::Limit,
451            dec!(0.001),
452        );
453        order.price = Some(dec!(10000));
454        order
455    }
456
457    #[tokio::test]
458    async fn accepted_order_returns_the_reply_data() {
459        let (url, received) = spawn_trade_server(Answer::Accept).await;
460        let (trade, _events) = connected(config(url)).await;
461
462        let reply = trade.place_order(&order()).await.unwrap();
463
464        assert_eq!(reply.data.order_id, "order-1");
465        assert_eq!(reply.header.unwrap().limit_status, Some(9));
466        let request = received.lock().unwrap().last().cloned().unwrap();
467        assert_eq!(request["op"], "order.create");
468        assert_eq!(request["args"][0]["symbol"], "BTCUSDT");
469        assert_eq!(request["header"]["X-BAPI-RECV-WINDOW"], "5000");
470        let sent: i64 = request["header"]["X-BAPI-TIMESTAMP"]
471            .as_str()
472            .unwrap()
473            .parse()
474            .unwrap();
475        assert!((sent - timestamp() as i64).abs() < 1000);
476        assert!(request["reqId"].as_str().unwrap().len() <= 36);
477    }
478
479    #[tokio::test]
480    async fn concurrent_requests_get_their_own_replies() {
481        let (url, received) = spawn_trade_server(Answer::Accept).await;
482        let (trade, _events) = connected(config(url)).await;
483
484        let order = order();
485        let (a, b) = tokio::join!(trade.place_order(&order), trade.place_order(&order));
486
487        assert!(a.is_ok() && b.is_ok());
488        let ids: Vec<_> = received
489            .lock()
490            .unwrap()
491            .iter()
492            .filter(|r| r["op"] == "order.create")
493            .map(|r| r["reqId"].as_str().unwrap().to_owned())
494            .collect();
495        assert_eq!(ids.len(), 2);
496        assert_ne!(ids[0], ids[1]);
497    }
498
499    #[tokio::test]
500    async fn rejected_order_is_an_api_error() {
501        let (url, _) = spawn_trade_server(Answer::Reject).await;
502        let (trade, _events) = connected(config(url)).await;
503
504        let err = trade.place_order(&order()).await.unwrap_err();
505
506        assert!(matches!(err, Error::Api { code: 10001, .. }), "{err:?}");
507    }
508
509    #[tokio::test]
510    async fn missing_reply_times_out() {
511        let (url, _) = spawn_trade_server(Answer::Silent).await;
512        let (trade, _events) = connected(config(url)).await;
513
514        let err = trade.place_order(&order()).await.unwrap_err();
515
516        assert!(matches!(err, Error::Timeout), "{err:?}");
517        assert!(trade.shared.pending.lock().unwrap().is_empty());
518    }
519
520    #[tokio::test]
521    async fn lost_connection_fails_the_waiting_request() {
522        let (url, _) = spawn_trade_server(Answer::Drop).await;
523        let (trade, _events) =
524            connected(config(url).request_timeout(Duration::from_secs(10))).await;
525
526        let err = trade.place_order(&order()).await.unwrap_err();
527
528        assert!(matches!(err, Error::ConnectionLost), "{err:?}");
529        assert!(!trade.is_ready());
530    }
531
532    #[tokio::test]
533    async fn requests_before_authentication_are_refused() {
534        let (url, _) = spawn_trade_server(Answer::Accept).await;
535        let (trade, _events) = TradeClient::new(config(url)).unwrap();
536
537        let err = trade.place_order(&order()).await.unwrap_err();
538
539        assert!(matches!(err, Error::NotConnected), "{err:?}");
540    }
541
542    #[tokio::test]
543    async fn credentials_are_required() {
544        let result = TradeClient::new(Config::new("ws://127.0.0.1:1"));
545
546        assert!(matches!(result, Err(Error::MissingCredentials)));
547    }
548
549    #[tokio::test]
550    async fn order_entry_heartbeat_keeps_the_connection() {
551        let (url, received) = spawn_trade_server(Answer::Accept).await;
552        let config = config(url)
553            .ping_interval(Some(Duration::from_millis(50)))
554            .pong_timeout(Duration::from_millis(200));
555        let (_trade, mut events) = connected(config).await;
556
557        tokio::time::sleep(Duration::from_millis(600)).await;
558
559        let pings = received
560            .lock()
561            .unwrap()
562            .iter()
563            .filter(|r| r["op"] == "ping")
564            .count();
565        assert!(pings >= 3, "only {pings} pings");
566        while let Ok(event) = events.try_recv() {
567            assert!(
568                !matches!(
569                    event,
570                    Event::Reconnecting { .. } | Event::Disconnected { .. }
571                ),
572                "{event:?}"
573            );
574        }
575    }
576}
bybit/ws/trade.rs

bybit/ws/
trade.rs
