brokapi_bybit_protocol/http/client.rs
1//! The authorized HTTP client: signing, the exchange clock offset and the mapping of error replies.
2//!
3//! The generic methods are here ([`BybitHttpClient::get_public`], [`BybitHttpClient::get_private`],
4//! [`BybitHttpClient::post_command`]). The endpoint methods (`get_order_book`, `create_order`, ...) are in
5//! [`super::endpoints`].
6
7use std::sync::Arc;
8use std::sync::atomic::{AtomicI64, Ordering};
9use std::time::{SystemTime, UNIX_EPOCH};
10
11use reqwest::{Client, Method, StatusCode};
12use serde::de::DeserializeOwned;
13use tracing::debug;
14
15use super::models::Envelope;
16use crate::config::BybitConfig;
17use crate::error::{BybitApiError, is_auth_code};
18use crate::sign::{rest_payload, sign};
19
20struct Inner {
21 api_key: String,
22 api_secret: String,
23 recv_window_ms: u64,
24 /// The exchange clock minus the local clock, milliseconds.
25 offset_ms: AtomicI64,
26}
27
28/// The HTTP client of the Bybit API V5.
29///
30/// It signs every private request with the key pair of the config, and keeps the offset between the clock of the
31/// exchange and the local clock ([`sync_time`](BybitHttpClient::sync_time)), so the timestamp is always inside the
32/// receive window.
33///
34/// * A **read** is sent once. The client does not retry: the caller decides.
35/// * A **command** (create, amend, cancel) is sent once too, and never repeated: a repeat is the one way to place an
36/// order twice. A lost answer is an unknown outcome, as for every command.
37///
38/// The client is cheap to clone. The limit of the exchange (per second and per endpoint) is not enforced here: the
39/// connector throttles through the shared limiter.
40///
41/// # Examples
42///
43/// ```
44/// # #[tokio::main(flavor = "current_thread")]
45/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
46/// # let server = brokapi_mocks::bybit::MockServer::start(Default::default()).await;
47/// # let cfg = brokapi_bybit_protocol::config::BybitConfig::for_mock("mock-key", "mock-secret", &server.http_base);
48/// use brokapi_bybit_protocol::http::client::BybitHttpClient;
49///
50/// let client = BybitHttpClient::new(cfg)?;
51/// let offset = client.sync_time().await?;
52/// assert!(offset.abs() < 5_000, "the clock of the imitation is the local one");
53/// # Ok(())
54/// # }
55/// ```
56#[derive(Clone)]
57pub struct BybitHttpClient {
58 client: Client,
59 inner: Arc<Inner>,
60 http_base_url: Arc<str>,
61 config: Arc<BybitConfig>,
62}
63
64fn local_ms() -> i64 {
65 SystemTime::now()
66 .duration_since(UNIX_EPOCH)
67 .map(|d| d.as_millis() as i64)
68 .unwrap_or(0)
69}
70
71impl BybitHttpClient {
72 /// Creates a client from a configuration. No network call is made.
73 ///
74 /// # Errors
75 ///
76 /// [`BybitApiError::Http`] (no status) if the HTTP stack cannot be built.
77 ///
78 /// # Examples
79 ///
80 /// ```
81 /// use brokapi_bybit_protocol::config::BybitConfig;
82 /// use brokapi_bybit_protocol::http::client::BybitHttpClient;
83 ///
84 /// let client = BybitHttpClient::new(BybitConfig::new("key", "secret"))?;
85 /// assert_eq!(client.base_url(), "https://api.bybit.com");
86 /// # Ok::<(), brokapi_bybit_protocol::error::BybitApiError>(())
87 /// ```
88 pub fn new(config: BybitConfig) -> Result<Self, BybitApiError> {
89 let client = Client::builder()
90 .timeout(config.request_timeout)
91 .gzip(true)
92 .build()?;
93 Ok(Self {
94 client,
95 inner: Arc::new(Inner {
96 api_key: config.api_key.clone(),
97 api_secret: config.api_secret.clone(),
98 recv_window_ms: config.recv_window_ms,
99 offset_ms: AtomicI64::new(0),
100 }),
101 http_base_url: Arc::from(config.http_base_url.trim_end_matches('/')),
102 config: Arc::new(config),
103 })
104 }
105
106 /// The base URL requests go to.
107 pub fn base_url(&self) -> &str {
108 &self.http_base_url
109 }
110
111 /// The configuration the client was made from.
112 pub fn config(&self) -> &BybitConfig {
113 &self.config
114 }
115
116 /// The timestamp to put in a request: the local clock moved by the offset of the exchange clock.
117 pub fn timestamp_ms(&self) -> i64 {
118 local_ms() + self.inner.offset_ms.load(Ordering::Relaxed)
119 }
120
121 /// Reads the clock of the exchange and keeps its offset from the local clock. Returns the offset in
122 /// milliseconds (positive: the exchange is ahead).
123 ///
124 /// The offset is taken at the middle of the round trip. A request is signed with the corrected time from then on.
125 ///
126 /// # Errors
127 ///
128 /// As [`BybitHttpClient::get_public`].
129 pub async fn sync_time(&self) -> Result<i64, BybitApiError> {
130 let before = local_ms();
131 let t = self.get_server_time().await?;
132 let after = local_ms();
133 let exchange_ms = t.time_nano / 1_000_000;
134 let offset = exchange_ms - (before + after) / 2;
135 self.inner.offset_ms.store(offset, Ordering::Relaxed);
136 debug!("the exchange clock is {offset} ms from the local one");
137 Ok(offset)
138 }
139
140 /// `GET`s a public path (no key) and returns the `result` of the reply parsed as `T`.
141 ///
142 /// # Errors
143 ///
144 /// * [`BybitApiError::Api`]: the exchange refused the request (`retCode` is not zero).
145 /// * [`BybitApiError::RateLimitExceeded`]: HTTP 429 or `retCode` 10006.
146 /// * [`BybitApiError::Http`]: another status, a transport failure, or a timeout.
147 /// * [`BybitApiError::Parse`]: the reply is a success but not the JSON that `T` expects.
148 ///
149 /// # Examples
150 ///
151 /// ```
152 /// # #[tokio::main(flavor = "current_thread")]
153 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
154 /// # let server = brokapi_mocks::bybit::MockServer::start(Default::default()).await;
155 /// # let cfg = brokapi_bybit_protocol::config::BybitConfig::for_mock("mock-key", "mock-secret", &server.http_base);
156 /// use brokapi_bybit_protocol::error::BybitApiError;
157 /// use brokapi_bybit_protocol::http::client::BybitHttpClient;
158 ///
159 /// let client = BybitHttpClient::new(cfg)?;
160 /// let time: serde_json::Value = client.get_public("/v5/market/time", &[]).await?;
161 /// assert!(time["timeSecond"].is_string());
162 ///
163 /// // A refusal of the exchange keeps its code.
164 /// let err = client
165 /// .get_public::<serde_json::Value>("/v5/market/orderbook", &[("category", "linear".into()), ("symbol", "NOPEUSDT".into())])
166 /// .await
167 /// .unwrap_err();
168 /// assert!(matches!(err, BybitApiError::Api { code: 10001, .. }));
169 /// # Ok(())
170 /// # }
171 /// ```
172 pub async fn get_public<T: DeserializeOwned>(
173 &self,
174 path: &str,
175 query: &[(&str, String)],
176 ) -> Result<T, BybitApiError> {
177 let url = self.url(path, &encode(query));
178 let resp = self.client.get(url).send().await?;
179 self.handle(resp).await
180 }
181
182 /// `GET`s a private path: signed with the key. Same errors as [`BybitHttpClient::get_public`], and
183 /// [`BybitApiError::Auth`] when the key is refused.
184 ///
185 /// # Examples
186 ///
187 /// ```
188 /// # #[tokio::main(flavor = "current_thread")]
189 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
190 /// # let server = brokapi_mocks::bybit::MockServer::start(Default::default()).await;
191 /// # let cfg = brokapi_bybit_protocol::config::BybitConfig::for_mock("mock-key", "mock-secret", &server.http_base);
192 /// use brokapi_bybit_protocol::http::client::BybitHttpClient;
193 ///
194 /// let client = BybitHttpClient::new(cfg)?;
195 /// let wallet: serde_json::Value = client
196 /// .get_private("/v5/account/wallet-balance", &[("accountType", "UNIFIED".into())])
197 /// .await?;
198 /// assert!(wallet["list"].is_array());
199 /// # Ok(())
200 /// # }
201 /// ```
202 pub async fn get_private<T: DeserializeOwned>(
203 &self,
204 path: &str,
205 query: &[(&str, String)],
206 ) -> Result<T, BybitApiError> {
207 let qs = encode(query);
208 let req = self.client.get(self.url(path, &qs));
209 let resp = self.sign_request(req, &qs).send().await?;
210 self.handle(resp).await
211 }
212
213 /// `POST`s a command (create, amend, cancel an order) with a JSON body. **Never repeats the request.**
214 ///
215 /// The body is signed as the exact text that is sent. A transport error or a timeout is an unknown outcome: the
216 /// caller must find out with a read.
217 ///
218 /// # Errors
219 ///
220 /// The same as [`BybitHttpClient::get_private`].
221 pub async fn post_command<T, B>(&self, path: &str, body: &B) -> Result<T, BybitApiError>
222 where
223 T: DeserializeOwned,
224 B: serde::Serialize,
225 {
226 let text = serde_json::to_string(body)?;
227 let req = self
228 .client
229 .request(Method::POST, self.url(path, ""))
230 .header("Content-Type", "application/json");
231 let resp = self.sign_request(req, &text).body(text).send().await?;
232 self.handle(resp).await
233 }
234
235 /// `GET`s a private path and returns the status and the body text, without parsing and without mapping errors.
236 /// An error status is not an `Err` here. Only a transport failure is.
237 ///
238 /// # Errors
239 ///
240 /// [`BybitApiError::Http`] (no status) on a transport failure.
241 pub async fn read_raw(
242 &self,
243 method: Method,
244 path: &str,
245 query: &[(&str, String)],
246 body: Option<&serde_json::Value>,
247 signed: bool,
248 ) -> Result<(StatusCode, String), BybitApiError> {
249 let (qs, text) = if method == Method::GET {
250 (encode(query), String::new())
251 } else {
252 (
253 String::new(),
254 body.map(|b| b.to_string()).unwrap_or_else(|| "{}".into()),
255 )
256 };
257 let mut req = self.client.request(method.clone(), self.url(path, &qs));
258 if method != Method::GET {
259 req = req.header("Content-Type", "application/json");
260 }
261 let signed_payload = if method == Method::GET { &qs } else { &text };
262 if signed {
263 req = self.sign_request(req, signed_payload);
264 }
265 if method != Method::GET {
266 req = req.body(text);
267 }
268 let resp = req.send().await?;
269 let status = resp.status();
270 Ok((status, resp.text().await?))
271 }
272
273 fn url(&self, path: &str, query: &str) -> String {
274 if query.is_empty() {
275 format!("{}{}", self.http_base_url, path)
276 } else {
277 format!("{}{}?{}", self.http_base_url, path, query)
278 }
279 }
280
281 fn sign_request(&self, req: reqwest::RequestBuilder, payload: &str) -> reqwest::RequestBuilder {
282 let ts = self.timestamp_ms();
283 let signature = sign(
284 &self.inner.api_secret,
285 &rest_payload(ts, &self.inner.api_key, self.inner.recv_window_ms, payload),
286 );
287 req.header("X-BAPI-API-KEY", &self.inner.api_key)
288 .header("X-BAPI-TIMESTAMP", ts.to_string())
289 .header("X-BAPI-RECV-WINDOW", self.inner.recv_window_ms.to_string())
290 .header("X-BAPI-SIGN", signature)
291 }
292
293 async fn handle<T: DeserializeOwned>(
294 &self,
295 resp: reqwest::Response,
296 ) -> Result<T, BybitApiError> {
297 let status = resp.status();
298 let text = resp.text().await?;
299 if !status.is_success() {
300 return Err(match status.as_u16() {
301 429 => BybitApiError::RateLimitExceeded,
302 401 => BybitApiError::Auth {
303 message: "HTTP 401: the key, the signature or the contour (main, demo, test) is wrong".into(),
304 },
305 s => BybitApiError::Http {
306 status: Some(s),
307 message: error_text(&text),
308 },
309 });
310 }
311 let env: Envelope<T> = serde_json::from_str(&text).map_err(|e| {
312 BybitApiError::Parse(format!("the reply is not an answer of the API: {e}"))
313 })?;
314 match env.ret_code {
315 0 => env
316 .result
317 .ok_or_else(|| BybitApiError::Parse("the reply has no result".into())),
318 10006 => Err(BybitApiError::RateLimitExceeded),
319 code if is_auth_code(code) => Err(BybitApiError::Auth {
320 message: format!("{code}: {}", env.ret_msg),
321 }),
322 code => Err(BybitApiError::Api {
323 code,
324 message: env.ret_msg,
325 }),
326 }
327 }
328}
329
330/// The query as the text that is sent and signed.
331///
332/// The value of `cursor` is written as it came in the previous answer: the exchange hands the token out already
333/// percent-encoded, and encoding it a second time would change it.
334fn encode(query: &[(&str, String)]) -> String {
335 query
336 .iter()
337 .map(|(k, v)| {
338 if *k == "cursor" {
339 format!("cursor={v}")
340 } else {
341 serde_urlencoded::to_string([(k, v)]).unwrap_or_default()
342 }
343 })
344 .collect::<Vec<_>>()
345 .join("&")
346}
347
348fn error_text(body: &str) -> String {
349 let body = body.trim();
350 if body.is_empty() {
351 return "empty body".to_string();
352 }
353 body.chars().take(300).collect()
354}
355
356#[cfg(test)]
357mod tests {
358 use super::*;
359
360 #[test]
361 fn the_query_is_one_text_in_the_order_given() {
362 let q = [
363 ("category", "linear".to_string()),
364 ("symbol", "BTC USDT".to_string()),
365 ];
366 assert_eq!(encode(&q), "category=linear&symbol=BTC+USDT");
367 assert_eq!(encode(&[]), "");
368 }
369
370 #[test]
371 fn the_timestamp_follows_the_offset() {
372 let c = BybitHttpClient::new(BybitConfig::new("k", "s")).unwrap();
373 let a = c.timestamp_ms();
374 c.inner.offset_ms.store(60_000, Ordering::Relaxed);
375 assert!(c.timestamp_ms() >= a + 59_000);
376 }
377}