1use crate::{
2 Environment, Error, ServerClock, Timestamp,
3 crypto::{SensitiveString, Signer, timestamp},
4 enums::Category,
5 http::{
6 APIErrorResponse, APIKeyInformation, AccountCoinBalance, AccountInfo,
7 AmendOrderBatchRequest, AmendOrderBatchResult, AmendOrderRequest, AmendOrderResponse,
8 AssetInfoResult, BorrowHistoryEntry, CancelAllOrdersRequest, CancelAllOrdersResponse,
9 CancelOrderBatchRequest, CancelOrderBatchResult, CancelOrderRequest, CancelOrderResponse,
10 CancelWithdrawalRequest, CancelWithdrawalResult, ClosedPnl, CoinGreeks, CoinInfoResult,
11 CollateralInfoEntry, CursorPagination, DeliveryPrice, DeliveryRecord, DepositAddress,
12 DepositAllowedCoinInfoResult, DepositRecords, EmptyResult, ExchangeOrderRecords,
13 ExecutionEntry, FeeRateEntry, FundingRateHistory, GetAccountCoinBalanceParams,
14 GetAssetInfoParams, GetBorrowHistoryParams, GetClosedPnlParams, GetCoinGreeksParams,
15 GetCoinInfoParams, GetCollateralInfoParams, GetDeliveryPriceParams,
16 GetDeliveryRecordParams, GetDepositAddressParams, GetDepositAllowedCoinInfoParams,
17 GetDepositRecordsParams, GetExchangeOrderRecordParams, GetExecutionListParams,
18 GetFeeRateParams, GetFundingRateHistoryParams, GetHistoricalVolatilityParams,
19 GetInstrumentsInfoParams, GetInsuranceParams, GetInternalTransferRecordsParams,
20 GetKLinesParams, GetLeverageTokenInfoParams, GetLeverageTokenMarketParams,
21 GetLeverageTokenOrderRecordsParams, GetOpenClosedOrdersParams, GetOpenInterestParams,
22 GetOrderHistoryParams, GetOrderbookParams, GetPositionInfoParams, GetRiskLimitParams,
23 GetSettlementRecordParams, GetSpotBorrowCheckParams, GetSubDepositAddressParams,
24 GetSubDepositRecordsParams, GetTickersParams, GetTradesParams, GetTransactionLogParams,
25 GetTransferableCoinsParams, GetUniversalTransferRecordsParams, GetWalletBalanceParams,
26 GetWithdrawableAmountParams, GetWithdrawalRecordsParams, Headers,
27 HistoricalVolatilityEntry, InstrumentsInfo, Insurance, InternalTransferEntry,
28 InternalTransferRequest, KLine, LeverageTokenInfo, LeverageTokenMarket,
29 LeverageTokenOrderRecord, List, OpenInterest, Order, Orderbook, PlaceOrderBatchRequest,
30 PlaceOrderBatchResult, PlaceOrderRequest, PlaceOrderResponse, Position, PriceKLine,
31 PurchaseLeverageTokenRequest, PurchaseLeverageTokenResult, RedeemLeverageTokenRequest,
32 RedeemLeverageTokenResult, Resp, Response, RiskLimit, SaveTransferSubMemberRequest,
33 ServerTime, SetAutoAddMarginRequest, SetLeverageRequest, SetMarginModeRequest,
34 SetMarginModeResponse, SetRiskLimitRequest, SetRiskLimitResponse,
35 SetSpotMarginLeverageRequest, SetTradingStopRequest, SettlementRecord, SpotBorrowCheck,
36 SwitchCrossIsolatedMarginRequest, SwitchPositionModeRequest, SwitchSpotMarginModeRequest,
37 SwitchSpotMarginModeResult, Ticker, Trade, TransactionLog, TransferResult,
38 TransferableSubMembers, UniversalTransferEntry, UniversalTransferRequest,
39 UpgradeToUtaResult, WalletBalance, WithdrawRecords, WithdrawRequest, WithdrawResult,
40 WithdrawableAmountResult,
41 },
42 serde::{deserialize_json, serialize_json, serialize_query},
43 url::*,
44};
45use reqwest::{
46 self, Method, RequestBuilder, StatusCode,
47 header::{CONTENT_TYPE, HeaderMap, HeaderValue},
48};
49use std::{sync::Arc, time::Duration};
50
51use super::rate_limiter::{RateLimitKey, RateLimiter, RateLimiterConfig, Rejection};
52
53pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
55pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
57pub const DEFAULT_RECV_WINDOW: Timestamp = 5000;
59
60const MAX_ERROR_BODY_LEN: usize = 1024;
62
63pub struct Config {
65 pub base_url: String,
67 pub api_key: Option<SensitiveString>,
69 pub api_secret: Option<SensitiveString>,
71 pub recv_window: Timestamp,
73 pub referer: Option<String>,
75 pub rate_limiter: Option<RateLimiterConfig>,
79 pub timeout: Option<Duration>,
83 pub connect_timeout: Option<Duration>,
85 pub proxy: Option<String>,
89 pub resync_time_on_timestamp_error: bool,
94}
95
96impl Config {
97 pub fn new(base_url: impl Into<String>) -> Self {
99 Self {
100 base_url: base_url.into(),
101 api_key: None,
102 api_secret: None,
103 recv_window: DEFAULT_RECV_WINDOW,
104 referer: None,
105 rate_limiter: None,
106 timeout: Some(DEFAULT_TIMEOUT),
107 connect_timeout: Some(DEFAULT_CONNECT_TIMEOUT),
108 resync_time_on_timestamp_error: true,
109 proxy: None,
110 }
111 }
112
113 pub fn for_env(env: Environment) -> Self {
115 Self::new(env.api_url())
116 }
117
118 pub fn credentials(
120 mut self,
121 api_key: impl Into<SensitiveString>,
122 api_secret: impl Into<SensitiveString>,
123 ) -> Self {
124 self.api_key = Some(api_key.into());
125 self.api_secret = Some(api_secret.into());
126 self
127 }
128
129 pub fn recv_window(mut self, recv_window: Timestamp) -> Self {
131 self.recv_window = recv_window;
132 self
133 }
134
135 pub fn referer(mut self, referer: impl Into<String>) -> Self {
137 self.referer = Some(referer.into());
138 self
139 }
140
141 pub fn rate_limiter(mut self, rate_limiter: RateLimiterConfig) -> Self {
143 self.rate_limiter = Some(rate_limiter);
144 self
145 }
146
147 pub fn timeout(mut self, timeout: Option<Duration>) -> Self {
149 self.timeout = timeout;
150 self
151 }
152
153 pub fn connect_timeout(mut self, connect_timeout: Option<Duration>) -> Self {
155 self.connect_timeout = connect_timeout;
156 self
157 }
158
159 pub fn proxy(mut self, url: impl Into<String>) -> Self {
161 self.proxy = Some(url.into());
162 self
163 }
164
165 pub fn resync_time_on_timestamp_error(mut self, enabled: bool) -> Self {
168 self.resync_time_on_timestamp_error = enabled;
169 self
170 }
171}
172
173#[derive(Debug, Clone)]
179pub struct Client {
180 base_url: String,
181 headers: HeaderMap,
182 client: reqwest::Client,
183 signer: Option<Arc<Signer>>,
184 rate_limiter: Option<Arc<RateLimiter>>,
185 clock: ServerClock,
187 resync_time_on_timestamp_error: bool,
188}
189
190impl Client {
191 pub fn new(cfg: Config) -> Result<Self, Error> {
193 let mut headers = HeaderMap::new();
194
195 if let Some(api_key) = cfg.api_key.as_ref() {
196 let api_key = api_key.expose().parse()?;
197 headers.append(HEADER_X_BAPI_API_KEY, api_key);
198 }
199
200 let recv_window = cfg.recv_window.to_string().parse()?;
201 headers.append(HEADER_X_BAPI_RECV_WINDOW, recv_window);
202
203 if let Some(referer) = cfg.referer {
204 let referer = referer.parse()?;
205 headers.append(HEADER_X_REFERER, referer);
206 }
207
208 let signer = cfg
209 .api_secret
210 .map(|api_secret| -> Result<Arc<Signer>, Error> {
211 let api_key = cfg
212 .api_key
213 .ok_or_else(|| Error::from("api_key is required when api_secret is set"))?;
214 Ok(Arc::new(Signer::new(
215 api_key,
216 api_secret,
217 cfg.recv_window,
218 None,
219 )))
220 })
221 .transpose()?;
222
223 let rate_limiter = cfg
224 .rate_limiter
225 .map(|config| Arc::new(RateLimiter::new(config)));
226
227 let mut default_headers = HeaderMap::new();
230 default_headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
231 let mut builder = reqwest::Client::builder().default_headers(default_headers);
232 if let Some(timeout) = cfg.timeout {
233 builder = builder.timeout(timeout);
234 }
235 if let Some(connect_timeout) = cfg.connect_timeout {
236 builder = builder.connect_timeout(connect_timeout);
237 }
238 if let Some(proxy) = &cfg.proxy {
239 builder = builder.proxy(reqwest::Proxy::all(proxy)?);
240 }
241
242 Ok(Self {
243 base_url: cfg.base_url,
244 headers,
245 client: builder.build()?,
246 signer,
247 rate_limiter,
248 clock: ServerClock::new(),
249 resync_time_on_timestamp_error: cfg.resync_time_on_timestamp_error,
250 })
251 }
252
253 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
264 pub async fn sync_time(&self) -> Result<i64, Error> {
265 let url = format!("{}{}", self.base_url, Path::MarketServerTime);
268 let request = self.client.request(Method::GET, url).build()?;
269 let key = RateLimitKey::new(Path::MarketServerTime);
270
271 let sent_at = timestamp();
272 let response: Response<ServerTime> = self.execute_once(key, 1, request).await?;
273 let received_at = timestamp();
274
275 let server_ms = response.result.time_nano / 1_000_000;
276 let offset = estimate_time_offset(sent_at, received_at, server_ms);
277 self.clock.set_offset(offset);
278 tracing::debug!(offset_ms = offset, "server time synchronized");
279 Ok(offset)
280 }
281
282 pub fn time_offset(&self) -> i64 {
285 self.clock.offset()
286 }
287
288 pub fn clock(&self) -> ServerClock {
292 self.clock.clone()
293 }
294
295 pub fn server_timestamp(&self) -> Timestamp {
299 self.clock.now()
300 }
301
302 fn get_signed_headers(&self, s: &str) -> Result<HeaderMap, Error> {
303 let signer = self.signer.as_ref().ok_or(Error::MissingCredentials)?;
304 let mut headers = self.headers.clone();
305
306 let timestamp = self.server_timestamp();
307 let signature = signer.sign_at(s, timestamp);
308 headers.append(HEADER_X_BAPI_SIGN, signature.parse()?);
309 headers.append(HEADER_X_BAPI_TIMESTAMP, timestamp.to_string().parse()?);
310
311 Ok(headers)
312 }
313}
314
315fn estimate_time_offset(sent_at: Timestamp, received_at: Timestamp, server_ms: Timestamp) -> i64 {
318 let midpoint = sent_at + received_at.saturating_sub(sent_at) / 2;
319 server_ms as i64 - midpoint as i64
320}
321
322impl Client {
324 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
326 pub async fn get_server_time(&self) -> Result<Response<ServerTime>, Error> {
327 let url = format!("{}{}", self.base_url, Path::MarketServerTime);
328
329 let request = self.client.request(Method::GET, url);
330
331 let response = self.send(Path::MarketServerTime, request).await?;
332 Ok(response)
333 }
334
335 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
340 pub async fn get_kline(&self, params: &GetKLinesParams) -> Result<Response<KLine>, Error> {
341 let url = format!("{}{}", self.base_url, Path::MarketKline);
342
343 let request = self.client.request(Method::GET, url).query(params);
344
345 let response = self.send(Path::MarketKline, request).await?;
346 Ok(response)
347 }
348
349 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
354 pub async fn get_mark_price_kline(
355 &self,
356 params: &GetKLinesParams,
357 ) -> Result<Response<PriceKLine>, Error> {
358 let url = format!("{}{}", self.base_url, Path::MarketMarkPriceKline);
359
360 let request = self.client.request(Method::GET, url).query(params);
361
362 let response = self.send(Path::MarketMarkPriceKline, request).await?;
363 Ok(response)
364 }
365
366 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
371 pub async fn get_index_price_kline(
372 &self,
373 params: &GetKLinesParams,
374 ) -> Result<Response<PriceKLine>, Error> {
375 let url = format!("{}{}", self.base_url, Path::MarketIndexPriceKline);
376
377 let request = self.client.request(Method::GET, url).query(params);
378
379 let response = self.send(Path::MarketIndexPriceKline, request).await?;
380 Ok(response)
381 }
382
383 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
388 pub async fn get_premium_index_price_kline(
389 &self,
390 params: &GetKLinesParams,
391 ) -> Result<Response<PriceKLine>, Error> {
392 let url = format!("{}{}", self.base_url, Path::MarketPremiumIndexPriceKline);
393
394 let request = self.client.request(Method::GET, url).query(params);
395
396 let response = self
397 .send(Path::MarketPremiumIndexPriceKline, request)
398 .await?;
399 Ok(response)
400 }
401
402 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
406 pub async fn get_tickers(&self, params: &GetTickersParams) -> Result<Response<Ticker>, Error> {
407 let url = format!("{}{}", self.base_url, Path::MarketTickers);
408
409 let request = self.client.request(Method::GET, url).query(params);
410
411 let response = self.send(Path::MarketTickers, request).await?;
412 Ok(response)
413 }
414
415 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
420 pub async fn get_orderbook(
421 &self,
422 params: &GetOrderbookParams,
423 ) -> Result<Response<Orderbook>, Error> {
424 let url = format!("{}{}", self.base_url, Path::MarketOrderbook);
425
426 let request = self.client.request(Method::GET, url).query(params);
427
428 let response = self.send(Path::MarketOrderbook, request).await?;
429 Ok(response)
430 }
431
432 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
435 pub async fn get_instruments_info(
436 &self,
437 params: &GetInstrumentsInfoParams,
438 ) -> Result<Response<InstrumentsInfo>, Error> {
439 let url = format!("{}{}", self.base_url, Path::MarketInstrumentsInfo);
440
441 let request = self.client.request(Method::GET, url).query(params);
442
443 let response = self.send(Path::MarketInstrumentsInfo, request).await?;
444 Ok(response)
445 }
446
447 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
450 pub async fn get_public_recent_trading_history(
451 &self,
452 params: &GetTradesParams,
453 ) -> Result<Response<Trade>, Error> {
454 let url = format!("{}{}", self.base_url, Path::MarketRecentTrade);
455
456 let request = self.client.request(Method::GET, url).query(params);
457
458 let response = self.send(Path::MarketRecentTrade, request).await?;
459 Ok(response)
460 }
461
462 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
467 pub async fn get_funding_rate_history(
468 &self,
469 params: &GetFundingRateHistoryParams,
470 ) -> Result<Response<FundingRateHistory>, Error> {
471 let url = format!("{}{}", self.base_url, Path::MarketFundingHistory);
472
473 let request = self.client.request(Method::GET, url).query(params);
474
475 let response = self.send(Path::MarketFundingHistory, request).await?;
476 Ok(response)
477 }
478
479 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
484 pub async fn get_open_interest(
485 &self,
486 params: &GetOpenInterestParams,
487 ) -> Result<Response<OpenInterest>, Error> {
488 let url = format!("{}{}", self.base_url, Path::MarketOpenInterest);
489
490 let request = self.client.request(Method::GET, url).query(params);
491
492 let response = self.send(Path::MarketOpenInterest, request).await?;
493 Ok(response)
494 }
495
496 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
504 pub async fn get_historical_volatility(
505 &self,
506 params: &GetHistoricalVolatilityParams,
507 ) -> Result<Response<Vec<HistoricalVolatilityEntry>>, Error> {
508 let url = format!("{}{}", self.base_url, Path::MarketHistoricalVolatility);
509
510 let request = self.client.request(Method::GET, url).query(params);
511
512 let response = self.send(Path::MarketHistoricalVolatility, request).await?;
513 Ok(response)
514 }
515
516 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
519 pub async fn get_insurance(
520 &self,
521 params: &GetInsuranceParams,
522 ) -> Result<Response<Insurance>, Error> {
523 let url = format!("{}{}", self.base_url, Path::MarketInsurance);
524
525 let request = self.client.request(Method::GET, url).query(params);
526
527 let response = self.send(Path::MarketInsurance, request).await?;
528 Ok(response)
529 }
530
531 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
536 pub async fn get_risk_limit(
537 &self,
538 params: &GetRiskLimitParams,
539 ) -> Result<Response<RiskLimit>, Error> {
540 let url = format!("{}{}", self.base_url, Path::MarketRiskLimit);
541
542 let request = self.client.request(Method::GET, url).query(params);
543
544 let response = self.send(Path::MarketRiskLimit, request).await?;
545 Ok(response)
546 }
547
548 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
553 pub async fn get_delivery_price(
554 &self,
555 params: &GetDeliveryPriceParams,
556 ) -> Result<Response<DeliveryPrice>, Error> {
557 let url = format!("{}{}", self.base_url, Path::MarketDeliveryPrice);
558
559 let request = self.client.request(Method::GET, url).query(params);
560
561 let response = self.send(Path::MarketDeliveryPrice, request).await?;
562 Ok(response)
563 }
564}
565
566impl Client {
568 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
615 pub async fn place_order(
616 &self,
617 request: &PlaceOrderRequest,
618 ) -> Result<Response<PlaceOrderResponse>, Error> {
619 let category = request.category;
620 let url = format!("{}{}", self.base_url, Path::TradeOrderCreate);
621 let json = serialize_json(request)?;
622 let headers = self.get_signed_headers(&json)?;
623
624 let request = self
625 .client
626 .request(Method::POST, url)
627 .headers(headers)
628 .body(json);
629
630 let response = self
631 .send_weighted(Path::TradeOrderCreate, Some(category), 1, request)
632 .await?;
633 Ok(response)
634 }
635
636 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
640 pub async fn amend_order(
641 &self,
642 request: &AmendOrderRequest,
643 ) -> Result<Response<AmendOrderResponse>, Error> {
644 let category = request.category;
645 let url = format!("{}{}", self.base_url, Path::TradeOrderAmend);
646 let json = serialize_json(request)?;
647 let headers = self.get_signed_headers(&json)?;
648
649 let request = self
650 .client
651 .request(Method::POST, url)
652 .headers(headers)
653 .body(json);
654
655 let response = self
656 .send_weighted(Path::TradeOrderAmend, Some(category), 1, request)
657 .await?;
658 Ok(response)
659 }
660
661 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
667 pub async fn cancel_order(
668 &self,
669 request: &CancelOrderRequest,
670 ) -> Result<Response<CancelOrderResponse>, Error> {
671 let category = request.category;
672 let url = format!("{}{}", self.base_url, Path::TradeOrderCancel);
673 let json = serialize_json(request)?;
674 let headers = self.get_signed_headers(&json)?;
675
676 let request = self
677 .client
678 .request(Method::POST, url)
679 .headers(headers)
680 .body(json);
681
682 let response = self
683 .send_weighted(Path::TradeOrderCancel, Some(category), 1, request)
684 .await?;
685 Ok(response)
686 }
687
688 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
702 pub async fn get_open_closed_orders(
703 &self,
704 params: &GetOpenClosedOrdersParams,
705 ) -> Result<Response<CursorPagination<Order>>, Error> {
706 let query = serialize_query(params)?;
707 let url = format!("{}{}?{query}", self.base_url, Path::TradeOrderRealtime);
708 let headers = self.get_signed_headers(&query)?;
709
710 let request = self.client.request(Method::GET, url).headers(headers);
711
712 let response = self.send(Path::TradeOrderRealtime, request).await?;
713 Ok(response)
714 }
715
716 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
720 pub async fn get_open_closed_orders_all(
721 &self,
722 params: &GetOpenClosedOrdersParams,
723 ) -> Result<Vec<Order>, Error> {
724 collect_pages(|cursor| {
725 let p = match cursor {
726 Some(cursor) => params.clone().with_cursor(cursor),
727 None => params.clone(),
728 };
729 async move { self.get_open_closed_orders(&p).await.map(|r| r.result) }
730 })
731 .await
732 }
733
734 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
737 pub async fn cancel_all_orders(
738 &self,
739 request: &CancelAllOrdersRequest,
740 ) -> Result<Response<CancelAllOrdersResponse>, Error> {
741 let category = request.category;
742 let url = format!("{}{}", self.base_url, Path::TradeOrderCancelAll);
743 let json = serialize_json(request)?;
744 let headers = self.get_signed_headers(&json)?;
745
746 let request = self
747 .client
748 .request(Method::POST, url)
749 .headers(headers)
750 .body(json);
751
752 let response = self
753 .send_weighted(Path::TradeOrderCancelAll, Some(category), 1, request)
754 .await?;
755 Ok(response)
756 }
757
758 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
762 pub async fn get_order_history(
763 &self,
764 params: &GetOrderHistoryParams,
765 ) -> Result<Response<CursorPagination<Order>>, Error> {
766 let query = serialize_query(params)?;
767 let url = format!("{}{}?{query}", self.base_url, Path::TradeOrderHistory);
768 let headers = self.get_signed_headers(&query)?;
769
770 let request = self.client.request(Method::GET, url).headers(headers);
771
772 let response = self.send(Path::TradeOrderHistory, request).await?;
773 Ok(response)
774 }
775
776 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
778 pub async fn get_order_history_all(
779 &self,
780 params: &GetOrderHistoryParams,
781 ) -> Result<Vec<Order>, Error> {
782 collect_pages(|cursor| {
783 let p = match cursor {
784 Some(cursor) => params.clone().with_cursor(cursor),
785 None => params.clone(),
786 };
787 async move { self.get_order_history(&p).await.map(|r| r.result) }
788 })
789 .await
790 }
791
792 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
796 pub async fn place_orders_batch(
797 &self,
798 request: &PlaceOrderBatchRequest,
799 ) -> Result<Response<List<PlaceOrderBatchResult>>, Error> {
800 let category = request.category;
801 let cost = request.request.len() as u32;
802 let url = format!("{}{}", self.base_url, Path::TradeOrderCreateBatch);
803 let json = serialize_json(request)?;
804 let headers = self.get_signed_headers(&json)?;
805
806 let request = self
807 .client
808 .request(Method::POST, url)
809 .headers(headers)
810 .body(json);
811
812 let response = self
813 .send_weighted(Path::TradeOrderCreateBatch, Some(category), cost, request)
814 .await?;
815 Ok(response)
816 }
817
818 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
822 pub async fn amend_orders_batch(
823 &self,
824 request: &AmendOrderBatchRequest,
825 ) -> Result<Response<List<AmendOrderBatchResult>>, Error> {
826 let category = request.category;
827 let cost = request.request.len() as u32;
828 let url = format!("{}{}", self.base_url, Path::TradeOrderAmendBatch);
829 let json = serialize_json(request)?;
830 let headers = self.get_signed_headers(&json)?;
831
832 let request = self
833 .client
834 .request(Method::POST, url)
835 .headers(headers)
836 .body(json);
837
838 let response = self
839 .send_weighted(Path::TradeOrderAmendBatch, Some(category), cost, request)
840 .await?;
841 Ok(response)
842 }
843
844 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
848 pub async fn cancel_orders_batch(
849 &self,
850 request: &CancelOrderBatchRequest,
851 ) -> Result<Response<List<CancelOrderBatchResult>>, Error> {
852 let category = request.category;
853 let cost = request.request.len() as u32;
854 let url = format!("{}{}", self.base_url, Path::TradeOrderCancelBatch);
855 let json = serialize_json(request)?;
856 let headers = self.get_signed_headers(&json)?;
857
858 let request = self
859 .client
860 .request(Method::POST, url)
861 .headers(headers)
862 .body(json);
863
864 let response = self
865 .send_weighted(Path::TradeOrderCancelBatch, Some(category), cost, request)
866 .await?;
867 Ok(response)
868 }
869
870 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
875 pub async fn get_spot_borrow_check(
876 &self,
877 params: &GetSpotBorrowCheckParams,
878 ) -> Result<Response<SpotBorrowCheck>, Error> {
879 let url = format!("{}{}", self.base_url, Path::TradeOrderSpotBorrowCheck);
880 let query = serialize_query(params)?;
881 let headers = self.get_signed_headers(&query)?;
882
883 let request = self
884 .client
885 .request(Method::GET, url)
886 .headers(headers)
887 .query(params);
888
889 let response = self.send(Path::TradeOrderSpotBorrowCheck, request).await?;
890 Ok(response)
891 }
892}
893
894impl Client {
896 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
907 pub async fn get_position_info(
908 &self,
909 params: &GetPositionInfoParams,
910 ) -> Result<Response<CursorPagination<Position>>, Error> {
911 let query = serialize_query(params)?;
912 let url = format!("{}{}?{query}", self.base_url, Path::PositionList);
913 let headers = self.get_signed_headers(&query)?;
914
915 let request = self.client.request(Method::GET, url).headers(headers);
916
917 let response = self.send(Path::PositionList, request).await?;
918 Ok(response)
919 }
920
921 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
925 pub async fn get_position_info_all(
926 &self,
927 params: &GetPositionInfoParams,
928 ) -> Result<Vec<Position>, Error> {
929 collect_pages(|cursor| {
930 let p = match cursor {
931 Some(cursor) => params.clone().with_cursor(cursor),
932 None => params.clone(),
933 };
934 async move { self.get_position_info(&p).await.map(|r| r.result) }
935 })
936 .await
937 }
938
939 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
942 pub async fn set_leverage(
943 &self,
944 request: &SetLeverageRequest,
945 ) -> Result<Response<EmptyResult>, Error> {
946 let url = format!("{}{}", self.base_url, Path::PositionSetLeverage);
947 let json = serialize_json(request)?;
948 let headers = self.get_signed_headers(&json)?;
949
950 let request = self
951 .client
952 .request(Method::POST, url)
953 .headers(headers)
954 .body(json);
955
956 let response = self.send(Path::PositionSetLeverage, request).await?;
957 Ok(response)
958 }
959
960 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
963 pub async fn set_trading_stop(
964 &self,
965 request: &SetTradingStopRequest,
966 ) -> Result<Response<EmptyResult>, Error> {
967 let url = format!("{}{}", self.base_url, Path::PositionTradingStop);
968 let json = serialize_json(request)?;
969 let headers = self.get_signed_headers(&json)?;
970
971 let request = self
972 .client
973 .request(Method::POST, url)
974 .headers(headers)
975 .body(json);
976
977 let response = self.send(Path::PositionTradingStop, request).await?;
978 Ok(response)
979 }
980
981 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
984 pub async fn switch_cross_isolated_margin(
985 &self,
986 request: &SwitchCrossIsolatedMarginRequest,
987 ) -> Result<Response<EmptyResult>, Error> {
988 let url = format!("{}{}", self.base_url, Path::PositionSwitchIsolated);
989 let json = serialize_json(request)?;
990 let headers = self.get_signed_headers(&json)?;
991
992 let request = self
993 .client
994 .request(Method::POST, url)
995 .headers(headers)
996 .body(json);
997
998 let response = self.send(Path::PositionSwitchIsolated, request).await?;
999 Ok(response)
1000 }
1001
1002 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1005 pub async fn switch_position_mode(
1006 &self,
1007 request: &SwitchPositionModeRequest,
1008 ) -> Result<Response<EmptyResult>, Error> {
1009 let url = format!("{}{}", self.base_url, Path::PositionSwitchMode);
1010 let json = serialize_json(request)?;
1011 let headers = self.get_signed_headers(&json)?;
1012
1013 let request = self
1014 .client
1015 .request(Method::POST, url)
1016 .headers(headers)
1017 .body(json);
1018
1019 let response = self.send(Path::PositionSwitchMode, request).await?;
1020 Ok(response)
1021 }
1022
1023 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1026 pub async fn set_auto_add_margin(
1027 &self,
1028 request: &SetAutoAddMarginRequest,
1029 ) -> Result<Response<EmptyResult>, Error> {
1030 let url = format!("{}{}", self.base_url, Path::PositionSetAutoAddMargin);
1031 let json = serialize_json(request)?;
1032 let headers = self.get_signed_headers(&json)?;
1033
1034 let request = self
1035 .client
1036 .request(Method::POST, url)
1037 .headers(headers)
1038 .body(json);
1039
1040 let response = self.send(Path::PositionSetAutoAddMargin, request).await?;
1041 Ok(response)
1042 }
1043
1044 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1047 pub async fn set_risk_limit(
1048 &self,
1049 request: &SetRiskLimitRequest,
1050 ) -> Result<Response<SetRiskLimitResponse>, Error> {
1051 let url = format!("{}{}", self.base_url, Path::PositionSetRiskLimit);
1052 let json = serialize_json(request)?;
1053 let headers = self.get_signed_headers(&json)?;
1054
1055 let request = self
1056 .client
1057 .request(Method::POST, url)
1058 .headers(headers)
1059 .body(json);
1060
1061 let response = self.send(Path::PositionSetRiskLimit, request).await?;
1062 Ok(response)
1063 }
1064
1065 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1068 pub async fn get_closed_pnl(
1069 &self,
1070 params: &GetClosedPnlParams,
1071 ) -> Result<Response<CursorPagination<ClosedPnl>>, Error> {
1072 let query = serialize_query(params)?;
1073 let url = format!("{}{}?{query}", self.base_url, Path::PositionClosedPnl);
1074 let headers = self.get_signed_headers(&query)?;
1075
1076 let request = self.client.request(Method::GET, url).headers(headers);
1077
1078 let response = self.send(Path::PositionClosedPnl, request).await?;
1079 Ok(response)
1080 }
1081
1082 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1084 pub async fn get_closed_pnl_all(
1085 &self,
1086 params: &GetClosedPnlParams,
1087 ) -> Result<Vec<ClosedPnl>, Error> {
1088 collect_pages(|cursor| {
1089 let p = match cursor {
1090 Some(cursor) => params.clone().with_cursor(cursor),
1091 None => params.clone(),
1092 };
1093 async move { self.get_closed_pnl(&p).await.map(|r| r.result) }
1094 })
1095 .await
1096 }
1097
1098 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1101 pub async fn get_execution_list(
1102 &self,
1103 params: &GetExecutionListParams,
1104 ) -> Result<Response<CursorPagination<ExecutionEntry>>, Error> {
1105 let query = serialize_query(params)?;
1106 let url = format!("{}{}?{query}", self.base_url, Path::ExecutionList);
1107 let headers = self.get_signed_headers(&query)?;
1108
1109 let request = self.client.request(Method::GET, url).headers(headers);
1110
1111 let response = self.send(Path::ExecutionList, request).await?;
1112 Ok(response)
1113 }
1114
1115 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1117 pub async fn get_execution_list_all(
1118 &self,
1119 params: &GetExecutionListParams,
1120 ) -> Result<Vec<ExecutionEntry>, Error> {
1121 collect_pages(|cursor| {
1122 let p = match cursor {
1123 Some(cursor) => params.clone().with_cursor(cursor),
1124 None => params.clone(),
1125 };
1126 async move { self.get_execution_list(&p).await.map(|r| r.result) }
1127 })
1128 .await
1129 }
1130}
1131
1132impl Client {
1134 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1136 pub async fn get_wallet_balance(
1137 &self,
1138 params: &GetWalletBalanceParams,
1139 ) -> Result<Response<List<WalletBalance>>, Error> {
1140 let query = serialize_query(params)?;
1141 let url = format!("{}{}?{query}", self.base_url, Path::AccountWalletBalance);
1142 let headers = self.get_signed_headers(&query)?;
1143
1144 let request = self.client.request(Method::GET, url).headers(headers);
1145
1146 let response = self.send(Path::AccountWalletBalance, request).await?;
1147 Ok(response)
1148 }
1149
1150 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1153 pub async fn get_transaction_log(
1154 &self,
1155 params: &GetTransactionLogParams,
1156 ) -> Result<Response<CursorPagination<TransactionLog>>, Error> {
1157 let query = serialize_query(params)?;
1158 let url = format!("{}{}?{query}", self.base_url, Path::AccountTransactionLog);
1159 let headers = self.get_signed_headers(&query)?;
1160
1161 let request = self.client.request(Method::GET, url).headers(headers);
1162
1163 let response = self.send(Path::AccountTransactionLog, request).await?;
1164 Ok(response)
1165 }
1166
1167 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1171 pub async fn get_transaction_log_all(
1172 &self,
1173 params: &GetTransactionLogParams,
1174 ) -> Result<Vec<TransactionLog>, Error> {
1175 collect_pages(|cursor| {
1176 let p = match cursor {
1177 Some(cursor) => params.clone().with_cursor(cursor),
1178 None => params.clone(),
1179 };
1180 async move { self.get_transaction_log(&p).await.map(|r| r.result) }
1181 })
1182 .await
1183 }
1184
1185 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1187 pub async fn get_account_info(&self) -> Result<Response<AccountInfo>, Error> {
1188 let url = format!("{}{}", self.base_url, Path::AccountInfo);
1189 let query = "";
1190 let headers = self.get_signed_headers(query)?;
1191
1192 let request = self.client.request(Method::GET, url).headers(headers);
1193
1194 let response = self.send(Path::AccountInfo, request).await?;
1195 Ok(response)
1196 }
1197
1198 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1203 pub async fn get_fee_rate(
1204 &self,
1205 params: &GetFeeRateParams,
1206 ) -> Result<Response<List<FeeRateEntry>>, Error> {
1207 let url = format!("{}{}", self.base_url, Path::AccountFeeRate);
1208 let query = serialize_query(params)?;
1209 let headers = self.get_signed_headers(&query)?;
1210
1211 let request = self
1212 .client
1213 .request(Method::GET, url)
1214 .headers(headers)
1215 .query(params);
1216
1217 let response = self.send(Path::AccountFeeRate, request).await?;
1218 Ok(response)
1219 }
1220
1221 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1226 pub async fn set_margin_mode(
1227 &self,
1228 request_body: &SetMarginModeRequest,
1229 ) -> Result<Response<SetMarginModeResponse>, Error> {
1230 let url = format!("{}{}", self.base_url, Path::AccountSetMarginMode);
1231 let body = serialize_json(request_body)?;
1232 let headers = self.get_signed_headers(&body)?;
1233
1234 let request = self
1235 .client
1236 .request(Method::POST, url)
1237 .headers(headers)
1238 .body(body);
1239
1240 let response = self.send(Path::AccountSetMarginMode, request).await?;
1241 Ok(response)
1242 }
1243
1244 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1250 pub async fn upgrade_to_uta(&self) -> Result<Response<UpgradeToUtaResult>, Error> {
1251 let url = format!("{}{}", self.base_url, Path::AccountUpgradeToUta);
1252 let body = "{}".to_owned();
1253 let headers = self.get_signed_headers(&body)?;
1254
1255 let request = self
1256 .client
1257 .request(Method::POST, url)
1258 .headers(headers)
1259 .body(body);
1260
1261 let response = self.send(Path::AccountUpgradeToUta, request).await?;
1262 Ok(response)
1263 }
1264
1265 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1270 pub async fn get_borrow_history(
1271 &self,
1272 params: &GetBorrowHistoryParams,
1273 ) -> Result<Response<CursorPagination<BorrowHistoryEntry>>, Error> {
1274 let query = serialize_query(params)?;
1275 let url = format!("{}{}?{query}", self.base_url, Path::AccountBorrowHistory);
1276 let headers = self.get_signed_headers(&query)?;
1277
1278 let request = self.client.request(Method::GET, url).headers(headers);
1279
1280 let response = self.send(Path::AccountBorrowHistory, request).await?;
1281 Ok(response)
1282 }
1283
1284 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1288 pub async fn get_borrow_history_all(
1289 &self,
1290 params: &GetBorrowHistoryParams,
1291 ) -> Result<Vec<BorrowHistoryEntry>, Error> {
1292 collect_pages(|cursor| {
1293 let p = match cursor {
1294 Some(cursor) => params.clone().with_cursor(cursor),
1295 None => params.clone(),
1296 };
1297 async move { self.get_borrow_history(&p).await.map(|r| r.result) }
1298 })
1299 .await
1300 }
1301
1302 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1308 pub async fn get_collateral_info(
1309 &self,
1310 params: &GetCollateralInfoParams,
1311 ) -> Result<Response<List<CollateralInfoEntry>>, Error> {
1312 let url = format!("{}{}", self.base_url, Path::AccountCollateralInfo);
1313 let query = serialize_query(params)?;
1314 let headers = self.get_signed_headers(&query)?;
1315
1316 let request = self
1317 .client
1318 .request(Method::GET, url)
1319 .headers(headers)
1320 .query(params);
1321
1322 let response = self.send(Path::AccountCollateralInfo, request).await?;
1323 Ok(response)
1324 }
1325}
1326
1327impl Client {
1329 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1332 pub async fn get_api_key_information(&self) -> Result<Response<APIKeyInformation>, Error> {
1333 let url = format!("{}{}", self.base_url, Path::UserQueryApi);
1334 let query = "";
1335 let headers = self.get_signed_headers(query)?;
1336
1337 let request = self.client.request(Method::GET, url).headers(headers);
1338
1339 let response = self.send(Path::UserQueryApi, request).await?;
1340 Ok(response)
1341 }
1342}
1343
1344impl Client {
1346 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1351 pub async fn internal_transfer(
1352 &self,
1353 request: &InternalTransferRequest,
1354 ) -> Result<Response<TransferResult>, Error> {
1355 let url = format!("{}{}", self.base_url, Path::AssetTransferInterTransfer);
1356 let body = serialize_json(request)?;
1357 let headers = self.get_signed_headers(&body)?;
1358
1359 let request = self
1360 .client
1361 .request(Method::POST, url)
1362 .headers(headers)
1363 .body(body);
1364
1365 let response = self.send(Path::AssetTransferInterTransfer, request).await?;
1366 Ok(response)
1367 }
1368
1369 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1374 pub async fn get_internal_transfer_records(
1375 &self,
1376 params: &GetInternalTransferRecordsParams,
1377 ) -> Result<Response<CursorPagination<InternalTransferEntry>>, Error> {
1378 let query = serialize_query(params)?;
1379 let url = format!(
1380 "{}{}?{query}",
1381 self.base_url,
1382 Path::AssetTransferQueryInterTransferList
1383 );
1384 let headers = self.get_signed_headers(&query)?;
1385
1386 let request = self.client.request(Method::GET, url).headers(headers);
1387
1388 let response = self
1389 .send(Path::AssetTransferQueryInterTransferList, request)
1390 .await?;
1391 Ok(response)
1392 }
1393
1394 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1400 pub async fn universal_transfer(
1401 &self,
1402 request: &UniversalTransferRequest,
1403 ) -> Result<Response<TransferResult>, Error> {
1404 let url = format!("{}{}", self.base_url, Path::AssetTransferUniversalTransfer);
1405 let body = serialize_json(request)?;
1406 let headers = self.get_signed_headers(&body)?;
1407
1408 let request = self
1409 .client
1410 .request(Method::POST, url)
1411 .headers(headers)
1412 .body(body);
1413
1414 let response = self
1415 .send(Path::AssetTransferUniversalTransfer, request)
1416 .await?;
1417 Ok(response)
1418 }
1419
1420 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1425 pub async fn get_universal_transfer_records(
1426 &self,
1427 params: &GetUniversalTransferRecordsParams,
1428 ) -> Result<Response<CursorPagination<UniversalTransferEntry>>, Error> {
1429 let query = serialize_query(params)?;
1430 let url = format!(
1431 "{}{}?{query}",
1432 self.base_url,
1433 Path::AssetTransferQueryUniversalTransferList
1434 );
1435 let headers = self.get_signed_headers(&query)?;
1436
1437 let request = self.client.request(Method::GET, url).headers(headers);
1438
1439 let response = self
1440 .send(Path::AssetTransferQueryUniversalTransferList, request)
1441 .await?;
1442 Ok(response)
1443 }
1444
1445 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1450 pub async fn get_transferable_coins(
1451 &self,
1452 params: &GetTransferableCoinsParams,
1453 ) -> Result<Response<List<String>>, Error> {
1454 let query = serialize_query(params)?;
1455 let url = format!(
1456 "{}{}?{query}",
1457 self.base_url,
1458 Path::AssetTransferQueryTransferCoinList
1459 );
1460 let headers = self.get_signed_headers(&query)?;
1461
1462 let request = self.client.request(Method::GET, url).headers(headers);
1463
1464 let response = self
1465 .send(Path::AssetTransferQueryTransferCoinList, request)
1466 .await?;
1467 Ok(response)
1468 }
1469
1470 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1476 pub async fn save_transfer_sub_member(
1477 &self,
1478 request: &SaveTransferSubMemberRequest,
1479 ) -> Result<Response<EmptyResult>, Error> {
1480 let url = format!(
1481 "{}{}",
1482 self.base_url,
1483 Path::AssetTransferSaveTransferSubMember
1484 );
1485 let body = serialize_json(request)?;
1486 let headers = self.get_signed_headers(&body)?;
1487
1488 let request = self
1489 .client
1490 .request(Method::POST, url)
1491 .headers(headers)
1492 .body(body);
1493
1494 let response = self
1495 .send(Path::AssetTransferSaveTransferSubMember, request)
1496 .await?;
1497 Ok(response)
1498 }
1499
1500 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1506 pub async fn get_transferable_sub_members(
1507 &self,
1508 ) -> Result<Response<TransferableSubMembers>, Error> {
1509 let url = format!("{}{}", self.base_url, Path::AssetTransferQuerySubMemberList);
1510 let query = "";
1511 let headers = self.get_signed_headers(query)?;
1512
1513 let request = self.client.request(Method::GET, url).headers(headers);
1514
1515 let response = self
1516 .send(Path::AssetTransferQuerySubMemberList, request)
1517 .await?;
1518 Ok(response)
1519 }
1520
1521 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1527 pub async fn get_account_coin_balance(
1528 &self,
1529 params: &GetAccountCoinBalanceParams,
1530 ) -> Result<Response<AccountCoinBalance>, Error> {
1531 let query = serialize_query(params)?;
1532 let url = format!(
1533 "{}{}?{query}",
1534 self.base_url,
1535 Path::AssetTransferQueryAccountCoinBalance
1536 );
1537 let headers = self.get_signed_headers(&query)?;
1538
1539 let request = self.client.request(Method::GET, url).headers(headers);
1540
1541 let response = self
1542 .send(Path::AssetTransferQueryAccountCoinBalance, request)
1543 .await?;
1544 Ok(response)
1545 }
1546
1547 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1552 pub async fn get_asset_info(
1553 &self,
1554 params: &GetAssetInfoParams,
1555 ) -> Result<Response<AssetInfoResult>, Error> {
1556 let query = serialize_query(params)?;
1557 let url = format!(
1558 "{}{}?{query}",
1559 self.base_url,
1560 Path::AssetTransferQueryAssetInfo
1561 );
1562 let headers = self.get_signed_headers(&query)?;
1563
1564 let request = self.client.request(Method::GET, url).headers(headers);
1565
1566 let response = self
1567 .send(Path::AssetTransferQueryAssetInfo, request)
1568 .await?;
1569 Ok(response)
1570 }
1571
1572 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1576 pub async fn get_deposit_allowed_coin_info(
1577 &self,
1578 params: &GetDepositAllowedCoinInfoParams,
1579 ) -> Result<Response<DepositAllowedCoinInfoResult>, Error> {
1580 let query = serialize_query(params)?;
1581 let url = format!(
1582 "{}{}?{query}",
1583 self.base_url,
1584 Path::AssetDepositQueryAllowedList
1585 );
1586 let headers = self.get_signed_headers(&query)?;
1587
1588 let request = self.client.request(Method::GET, url).headers(headers);
1589
1590 let response = self
1591 .send(Path::AssetDepositQueryAllowedList, request)
1592 .await?;
1593 Ok(response)
1594 }
1595
1596 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1600 pub async fn get_deposit_records(
1601 &self,
1602 params: &GetDepositRecordsParams,
1603 ) -> Result<Response<DepositRecords>, Error> {
1604 let query = serialize_query(params)?;
1605 let url = format!("{}{}?{query}", self.base_url, Path::AssetDepositQueryRecord);
1606 let headers = self.get_signed_headers(&query)?;
1607
1608 let request = self.client.request(Method::GET, url).headers(headers);
1609
1610 let response = self.send(Path::AssetDepositQueryRecord, request).await?;
1611 Ok(response)
1612 }
1613
1614 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1618 pub async fn get_sub_deposit_records(
1619 &self,
1620 params: &GetSubDepositRecordsParams,
1621 ) -> Result<Response<DepositRecords>, Error> {
1622 let query = serialize_query(params)?;
1623 let url = format!(
1624 "{}{}?{query}",
1625 self.base_url,
1626 Path::AssetDepositQuerySubMemberRecord
1627 );
1628 let headers = self.get_signed_headers(&query)?;
1629
1630 let request = self.client.request(Method::GET, url).headers(headers);
1631
1632 let response = self
1633 .send(Path::AssetDepositQuerySubMemberRecord, request)
1634 .await?;
1635 Ok(response)
1636 }
1637
1638 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1642 pub async fn get_deposit_address(
1643 &self,
1644 params: &GetDepositAddressParams,
1645 ) -> Result<Response<DepositAddress>, Error> {
1646 let query = serialize_query(params)?;
1647 let url = format!(
1648 "{}{}?{query}",
1649 self.base_url,
1650 Path::AssetDepositQueryAddress
1651 );
1652 let headers = self.get_signed_headers(&query)?;
1653
1654 let request = self.client.request(Method::GET, url).headers(headers);
1655
1656 let response = self.send(Path::AssetDepositQueryAddress, request).await?;
1657 Ok(response)
1658 }
1659
1660 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1664 pub async fn get_sub_deposit_address(
1665 &self,
1666 params: &GetSubDepositAddressParams,
1667 ) -> Result<Response<DepositAddress>, Error> {
1668 let query = serialize_query(params)?;
1669 let url = format!(
1670 "{}{}?{query}",
1671 self.base_url,
1672 Path::AssetDepositQuerySubMemberAddress
1673 );
1674 let headers = self.get_signed_headers(&query)?;
1675
1676 let request = self.client.request(Method::GET, url).headers(headers);
1677
1678 let response = self
1679 .send(Path::AssetDepositQuerySubMemberAddress, request)
1680 .await?;
1681 Ok(response)
1682 }
1683
1684 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1688 pub async fn get_withdrawal_records(
1689 &self,
1690 params: &GetWithdrawalRecordsParams,
1691 ) -> Result<Response<WithdrawRecords>, Error> {
1692 let query = serialize_query(params)?;
1693 let url = format!(
1694 "{}{}?{query}",
1695 self.base_url,
1696 Path::AssetWithdrawQueryRecord
1697 );
1698 let headers = self.get_signed_headers(&query)?;
1699
1700 let request = self.client.request(Method::GET, url).headers(headers);
1701
1702 let response = self.send(Path::AssetWithdrawQueryRecord, request).await?;
1703 Ok(response)
1704 }
1705
1706 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1712 pub async fn get_withdrawable_amount(
1713 &self,
1714 params: &GetWithdrawableAmountParams,
1715 ) -> Result<Response<WithdrawableAmountResult>, Error> {
1716 let query = serialize_query(params)?;
1717 let url = format!(
1718 "{}{}?{query}",
1719 self.base_url,
1720 Path::AssetWithdrawWithdrawableAmount
1721 );
1722 let headers = self.get_signed_headers(&query)?;
1723
1724 let request = self.client.request(Method::GET, url).headers(headers);
1725
1726 let response = self
1727 .send(Path::AssetWithdrawWithdrawableAmount, request)
1728 .await?;
1729 Ok(response)
1730 }
1731
1732 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1737 pub async fn withdraw(
1738 &self,
1739 request: &WithdrawRequest,
1740 ) -> Result<Response<WithdrawResult>, Error> {
1741 let url = format!("{}{}", self.base_url, Path::AssetWithdrawCreate);
1742 let body = serialize_json(request)?;
1743 let headers = self.get_signed_headers(&body)?;
1744
1745 let request = self
1746 .client
1747 .request(Method::POST, url)
1748 .headers(headers)
1749 .body(body);
1750
1751 let response = self.send(Path::AssetWithdrawCreate, request).await?;
1752 Ok(response)
1753 }
1754
1755 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1760 pub async fn cancel_withdrawal(
1761 &self,
1762 request: &CancelWithdrawalRequest,
1763 ) -> Result<Response<CancelWithdrawalResult>, Error> {
1764 let url = format!("{}{}", self.base_url, Path::AssetWithdrawCancel);
1765 let body = serialize_json(request)?;
1766 let headers = self.get_signed_headers(&body)?;
1767
1768 let request = self
1769 .client
1770 .request(Method::POST, url)
1771 .headers(headers)
1772 .body(body);
1773
1774 let response = self.send(Path::AssetWithdrawCancel, request).await?;
1775 Ok(response)
1776 }
1777
1778 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1783 pub async fn get_coin_info(
1784 &self,
1785 params: &GetCoinInfoParams,
1786 ) -> Result<Response<CoinInfoResult>, Error> {
1787 let url = format!("{}{}", self.base_url, Path::AssetCoinQueryInfo);
1788 let query = serialize_query(params)?;
1789 let headers = self.get_signed_headers(&query)?;
1790
1791 let request = self
1792 .client
1793 .request(Method::GET, url)
1794 .headers(headers)
1795 .query(params);
1796
1797 let response = self.send(Path::AssetCoinQueryInfo, request).await?;
1798 Ok(response)
1799 }
1800
1801 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1805 pub async fn get_exchange_order_record(
1806 &self,
1807 params: &GetExchangeOrderRecordParams,
1808 ) -> Result<Response<ExchangeOrderRecords>, Error> {
1809 let query = serialize_query(params)?;
1810 let url = format!(
1811 "{}{}?{query}",
1812 self.base_url,
1813 Path::AssetExchangeOrderRecord
1814 );
1815 let headers = self.get_signed_headers(&query)?;
1816
1817 let request = self.client.request(Method::GET, url).headers(headers);
1818
1819 let response = self.send(Path::AssetExchangeOrderRecord, request).await?;
1820 Ok(response)
1821 }
1822
1823 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1828 pub async fn get_delivery_record(
1829 &self,
1830 params: &GetDeliveryRecordParams,
1831 ) -> Result<Response<CursorPagination<DeliveryRecord>>, Error> {
1832 let query = serialize_query(params)?;
1833 let url = format!("{}{}?{query}", self.base_url, Path::AssetDeliveryRecord);
1834 let headers = self.get_signed_headers(&query)?;
1835
1836 let request = self.client.request(Method::GET, url).headers(headers);
1837
1838 let response = self.send(Path::AssetDeliveryRecord, request).await?;
1839 Ok(response)
1840 }
1841
1842 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1847 pub async fn get_settlement_record(
1848 &self,
1849 params: &GetSettlementRecordParams,
1850 ) -> Result<Response<CursorPagination<SettlementRecord>>, Error> {
1851 let query = serialize_query(params)?;
1852 let url = format!("{}{}?{query}", self.base_url, Path::AssetSettlementRecord);
1853 let headers = self.get_signed_headers(&query)?;
1854
1855 let request = self.client.request(Method::GET, url).headers(headers);
1856
1857 let response = self.send(Path::AssetSettlementRecord, request).await?;
1858 Ok(response)
1859 }
1860
1861 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1866 pub async fn get_coin_greeks(
1867 &self,
1868 params: &GetCoinGreeksParams,
1869 ) -> Result<Response<List<CoinGreeks>>, Error> {
1870 let query = serialize_query(params)?;
1871 let url = format!("{}{}?{query}", self.base_url, Path::AssetCoinGreeks);
1872 let headers = self.get_signed_headers(&query)?;
1873
1874 let request = self.client.request(Method::GET, url).headers(headers);
1875
1876 let response = self.send(Path::AssetCoinGreeks, request).await?;
1877 Ok(response)
1878 }
1879}
1880
1881impl Client {
1884 #[deprecated(
1889 since = "0.3.0",
1890 note = "Bybit shut down the leveraged token API on 2025-07-04; the endpoint returns HTTP 404"
1891 )]
1892 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1893 pub async fn get_leverage_token_info(
1894 &self,
1895 params: &GetLeverageTokenInfoParams,
1896 ) -> Result<Response<List<LeverageTokenInfo>>, Error> {
1897 let url = format!("{}{}", self.base_url, Path::SpotLeverTokenInfo);
1898
1899 let request = self.client.request(Method::GET, url).query(params);
1900
1901 let response = self.send(Path::SpotLeverTokenInfo, request).await?;
1902 Ok(response)
1903 }
1904
1905 #[deprecated(
1910 since = "0.3.0",
1911 note = "Bybit shut down the leveraged token API on 2025-07-04; the endpoint returns HTTP 404"
1912 )]
1913 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1914 pub async fn get_leverage_token_market(
1915 &self,
1916 params: &GetLeverageTokenMarketParams,
1917 ) -> Result<Response<LeverageTokenMarket>, Error> {
1918 let url = format!("{}{}", self.base_url, Path::SpotLeverTokenReference);
1919
1920 let request = self.client.request(Method::GET, url).query(params);
1921
1922 let response = self.send(Path::SpotLeverTokenReference, request).await?;
1923 Ok(response)
1924 }
1925
1926 #[deprecated(
1931 since = "0.3.0",
1932 note = "Bybit shut down the leveraged token API on 2025-07-04; the endpoint returns HTTP 404"
1933 )]
1934 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1935 pub async fn purchase_leverage_token(
1936 &self,
1937 request: &PurchaseLeverageTokenRequest,
1938 ) -> Result<Response<PurchaseLeverageTokenResult>, Error> {
1939 let url = format!("{}{}", self.base_url, Path::SpotLeverTokenPurchase);
1940 let body = serialize_json(request)?;
1941 let headers = self.get_signed_headers(&body)?;
1942
1943 let request = self
1944 .client
1945 .request(Method::POST, url)
1946 .headers(headers)
1947 .body(body);
1948
1949 let response = self.send(Path::SpotLeverTokenPurchase, request).await?;
1950 Ok(response)
1951 }
1952
1953 #[deprecated(
1958 since = "0.3.0",
1959 note = "Bybit shut down the leveraged token API on 2025-07-04; the endpoint returns HTTP 404"
1960 )]
1961 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1962 pub async fn redeem_leverage_token(
1963 &self,
1964 request: &RedeemLeverageTokenRequest,
1965 ) -> Result<Response<RedeemLeverageTokenResult>, Error> {
1966 let url = format!("{}{}", self.base_url, Path::SpotLeverTokenRedeem);
1967 let body = serialize_json(request)?;
1968 let headers = self.get_signed_headers(&body)?;
1969
1970 let request = self
1971 .client
1972 .request(Method::POST, url)
1973 .headers(headers)
1974 .body(body);
1975
1976 let response = self.send(Path::SpotLeverTokenRedeem, request).await?;
1977 Ok(response)
1978 }
1979
1980 #[deprecated(
1984 since = "0.3.0",
1985 note = "Bybit shut down the leveraged token API on 2025-07-04; the endpoint returns HTTP 404"
1986 )]
1987 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
1988 pub async fn get_leverage_token_order_records(
1989 &self,
1990 params: &GetLeverageTokenOrderRecordsParams,
1991 ) -> Result<Response<List<LeverageTokenOrderRecord>>, Error> {
1992 let query = serialize_query(params)?;
1993 let url = format!(
1994 "{}{}?{query}",
1995 self.base_url,
1996 Path::SpotLeverTokenOrderRecord
1997 );
1998 let headers = self.get_signed_headers(&query)?;
1999
2000 let request = self.client.request(Method::GET, url).headers(headers);
2001
2002 let response = self.send(Path::SpotLeverTokenOrderRecord, request).await?;
2003 Ok(response)
2004 }
2005}
2006
2007impl Client {
2009 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
2015 pub async fn switch_spot_margin_mode(
2016 &self,
2017 request: &SwitchSpotMarginModeRequest,
2018 ) -> Result<Response<SwitchSpotMarginModeResult>, Error> {
2019 let url = format!("{}{}", self.base_url, Path::SpotMarginTradeSwitchMode);
2020 let body = serialize_json(request)?;
2021 let headers = self.get_signed_headers(&body)?;
2022
2023 let request = self
2024 .client
2025 .request(Method::POST, url)
2026 .headers(headers)
2027 .body(body);
2028
2029 let response = self.send(Path::SpotMarginTradeSwitchMode, request).await?;
2030 Ok(response)
2031 }
2032
2033 #[tracing::instrument(level = "debug", skip_all, err(level = "debug"))]
2039 pub async fn set_spot_margin_leverage(
2040 &self,
2041 request: &SetSpotMarginLeverageRequest,
2042 ) -> Result<Response<EmptyResult>, Error> {
2043 let url = format!("{}{}", self.base_url, Path::SpotMarginTradeSetLeverage);
2044 let body = serialize_json(request)?;
2045 let headers = self.get_signed_headers(&body)?;
2046
2047 let request = self
2048 .client
2049 .request(Method::POST, url)
2050 .headers(headers)
2051 .body(body);
2052
2053 let response = self.send(Path::SpotMarginTradeSetLeverage, request).await?;
2054 Ok(response)
2055 }
2056}
2057
2058impl Client {
2059 async fn send<T>(&self, path: Path, request: RequestBuilder) -> Result<Response<T>, Error>
2061 where
2062 T: serde::de::DeserializeOwned,
2063 {
2064 self.send_weighted(path, None, 1, request).await
2065 }
2066
2067 async fn send_weighted<T>(
2071 &self,
2072 path: Path,
2073 category: Option<Category>,
2074 cost: u32,
2075 request: RequestBuilder,
2076 ) -> Result<Response<T>, Error>
2077 where
2078 T: serde::de::DeserializeOwned,
2079 {
2080 let key = RateLimitKey { path, category };
2081 let request = request.build()?;
2082 let signed = request.headers().contains_key(HEADER_X_BAPI_SIGN);
2083 let retry = (signed && self.resync_time_on_timestamp_error)
2084 .then(|| request.try_clone())
2085 .flatten();
2086
2087 match self.execute_once(key, cost, request).await {
2088 Err(e) if e.is_timestamp_error() && retry.is_some() => {
2089 let mut retry = retry.expect("checked above");
2090 let offset_ms = self.sync_time().await?;
2091 tracing::debug!(%path, offset_ms, "timestamp rejected, retrying with server time");
2092 self.resign(&mut retry)?;
2093 self.execute_once(key, cost, retry).await
2094 }
2095 result => result,
2096 }
2097 }
2098
2099 async fn execute_once<T>(
2102 &self,
2103 key: RateLimitKey,
2104 cost: u32,
2105 request: reqwest::Request,
2106 ) -> Result<Response<T>, Error>
2107 where
2108 T: serde::de::DeserializeOwned,
2109 {
2110 let path = key.path;
2111 if let Some(rl) = &self.rate_limiter {
2112 rl.check(key, cost).map_err(|rejection| match rejection {
2113 Rejection::RetryAfter(retry_after_ms) => Error::RateLimited { retry_after_ms },
2114 Rejection::Unsatisfiable { cost, burst } => {
2115 Error::RateLimitUnsatisfiable { cost, burst }
2116 }
2117 })?;
2118 }
2119
2120 let start = std::time::Instant::now();
2121 let response = self.client.execute(request).await?;
2122 let elapsed_ms = start.elapsed().as_millis();
2123 let status = response.status();
2124 let headers = parse_headers(response.headers());
2125
2126 if let Some(rl) = &self.rate_limiter {
2127 rl.update(
2128 key,
2129 headers.api_limit_status,
2130 headers.api_limit_reset_timestamp,
2131 );
2132 }
2133
2134 let body = response.text().await?;
2135 let result = parse_response(status, headers, &body);
2136 match &result {
2137 Ok(response) => tracing::debug!(
2138 %path,
2139 elapsed_ms,
2140 api_limit = response.headers.api_limit,
2141 api_limit_status = response.headers.api_limit_status,
2142 "api call completed"
2143 ),
2144 Err(e) => tracing::debug!(%path, elapsed_ms, error = %e, "api call failed"),
2145 }
2146 result
2147 }
2148
2149 fn resign(&self, request: &mut reqwest::Request) -> Result<(), Error> {
2152 let signer = self.signer.as_ref().ok_or(Error::MissingCredentials)?;
2153 let payload = match request.body().and_then(|body| body.as_bytes()) {
2154 Some(body) => String::from_utf8_lossy(body).into_owned(),
2155 None => request.url().query().unwrap_or_default().to_owned(),
2156 };
2157 let timestamp = self.server_timestamp();
2158 let signature = signer.sign_at(&payload, timestamp);
2159 let headers = request.headers_mut();
2160 headers.insert(HEADER_X_BAPI_SIGN, signature.parse()?);
2161 headers.insert(HEADER_X_BAPI_TIMESTAMP, timestamp.to_string().parse()?);
2162 Ok(())
2163 }
2164}
2165
2166pub const MAX_PAGES: usize = 1000;
2168
2169async fn collect_pages<T, F, Fut>(mut fetch: F) -> Result<Vec<T>, Error>
2176where
2177 F: FnMut(Option<String>) -> Fut,
2178 Fut: std::future::Future<Output = Result<CursorPagination<T>, Error>>,
2179{
2180 let mut all = Vec::new();
2181 let mut cursor: Option<String> = None;
2182 for _ in 0..MAX_PAGES {
2183 let page = fetch(cursor.clone()).await?;
2184 let empty = page.list.is_empty();
2185 all.extend(page.list);
2186 match page.next_page_cursor {
2187 Some(next) if !empty && cursor.as_ref() != Some(&next) => cursor = Some(next),
2188 _ => return Ok(all),
2189 }
2190 }
2191 Err(Error::TooManyPages { max: MAX_PAGES })
2192}
2193
2194fn parse_response<T>(status: StatusCode, headers: Headers, body: &str) -> Result<Response<T>, Error>
2200where
2201 T: serde::de::DeserializeOwned,
2202{
2203 let envelope = deserialize_json::<APIErrorResponse>(body);
2204
2205 if !status.is_success() {
2206 return Err(match envelope {
2207 Ok(envelope) if envelope.ret_code != 0 => envelope.into(),
2208 _ => Error::Http {
2209 status: status.as_u16(),
2210 body: truncate(body, MAX_ERROR_BODY_LEN).to_owned(),
2211 },
2212 });
2213 }
2214
2215 let envelope = envelope?;
2216 if envelope.ret_code != 0 {
2217 return Err(envelope.into());
2218 }
2219
2220 let response: Resp<T> = deserialize_json(body)?;
2221 Ok(Response {
2222 result: response.result,
2223 time: response.time,
2224 headers,
2225 ret_ext_info: response.ret_ext_info,
2226 })
2227}
2228
2229fn truncate(s: &str, max: usize) -> &str {
2231 if s.len() <= max {
2232 return s;
2233 }
2234 let mut end = max;
2235 while !s.is_char_boundary(end) {
2236 end -= 1;
2237 }
2238 &s[..end]
2239}
2240
2241fn parse_headers(headers: &HeaderMap) -> Headers {
2243 let ret_code = headers
2244 .get(HEADER_RET_CODE)
2245 .and_then(|h| h.to_str().unwrap_or_default().parse().ok());
2246 let trace_id = headers
2247 .get(HEADER_TRACE_ID)
2248 .and_then(|h| h.to_str().map(|str| str.into()).ok());
2249 let time_now = headers
2250 .get(HEADER_TIME_NOW)
2251 .and_then(|h| h.to_str().unwrap_or_default().parse().ok());
2252
2253 let api_limit = headers
2254 .get(HEADER_X_BAPI_LIMIT)
2255 .and_then(|h| h.to_str().unwrap_or_default().parse().ok());
2256 let api_limit_status = headers
2257 .get(HEADER_X_BAPI_LIMIT_STATUS)
2258 .and_then(|h| h.to_str().unwrap_or_default().parse().ok());
2259 let api_limit_reset_timestamp = headers
2260 .get(HEADER_X_BAPI_LIMIT_RESET_TIMESTAMP)
2261 .and_then(|h| h.to_str().unwrap_or_default().parse().ok());
2262
2263 Headers {
2264 ret_code,
2265 trace_id,
2266 time_now,
2267 api_limit,
2268 api_limit_status,
2269 api_limit_reset_timestamp,
2270 }
2271}
2272
2273#[cfg(test)]
2274mod tests {
2275 use super::*;
2276
2277 fn headers() -> Headers {
2278 parse_headers(&HeaderMap::new())
2279 }
2280
2281 #[test]
2282 fn parse_response_success() {
2283 let body = r#"{"retCode":0,"retMsg":"OK","result":{"timeSecond":"1688639403","timeNano":"1688639403423213947"},"retExtInfo":{},"time":1688639403423}"#;
2284
2285 let response: Response<ServerTime> =
2286 parse_response(StatusCode::OK, headers(), body).unwrap();
2287
2288 assert_eq!(response.time, Some(1688639403423));
2289 }
2290
2291 #[test]
2292 fn parse_response_api_error_with_http_200() {
2293 let body = r#"{"retCode":10001,"retMsg":"params error","result":{},"retExtInfo":{},"time":1688639403423}"#;
2294
2295 let err = parse_response::<ServerTime>(StatusCode::OK, headers(), body).unwrap_err();
2296
2297 assert!(matches!(err, Error::Api { code: 10001, ref msg } if msg == "params error"));
2298 }
2299
2300 #[test]
2301 fn parse_response_api_error_with_http_error_status() {
2302 let body = r#"{"retCode":10006,"retMsg":"Too many visits!","result":{},"retExtInfo":{},"time":1688639403423}"#;
2303
2304 let err = parse_response::<ServerTime>(StatusCode::TOO_MANY_REQUESTS, headers(), body)
2305 .unwrap_err();
2306
2307 assert!(matches!(err, Error::Api { code: 10006, .. }));
2308 }
2309
2310 #[test]
2311 fn parse_response_non_json_error_page() {
2312 let body = "<html><body>403 Forbidden</body></html>";
2313
2314 let err = parse_response::<ServerTime>(StatusCode::FORBIDDEN, headers(), body).unwrap_err();
2315
2316 assert!(matches!(err, Error::Http { status: 403, body: ref b } if b == body));
2317 }
2318
2319 #[test]
2320 fn truncate_respects_char_boundaries() {
2321 assert_eq!(truncate("abc", 10), "abc");
2322 assert_eq!(truncate("ab\u{00e9}c", 3), "ab");
2323 }
2324
2325 fn page(list: Vec<u32>, cursor: Option<&str>) -> CursorPagination<u32> {
2326 CursorPagination {
2327 category: None,
2328 next_page_cursor: cursor.map(String::from),
2329 list,
2330 }
2331 }
2332
2333 async fn collect_from(
2335 pages: Vec<CursorPagination<u32>>,
2336 ) -> (Result<Vec<u32>, Error>, Vec<Option<String>>) {
2337 let mut pages = pages.into_iter();
2338 let mut requested = Vec::new();
2339 let result = collect_pages(|cursor| {
2340 requested.push(cursor);
2341 let page = pages.next().expect("unexpected extra request");
2342 async move { Ok(page) }
2343 })
2344 .await;
2345 (result, requested)
2346 }
2347
2348 #[tokio::test]
2349 async fn collect_pages_follows_cursor_until_none() {
2350 let (result, requested) = collect_from(vec![
2351 page(vec![1, 2], Some("a")),
2352 page(vec![3], Some("b")),
2353 page(vec![4], None),
2354 ])
2355 .await;
2356
2357 assert_eq!(result.unwrap(), vec![1, 2, 3, 4]);
2358 assert_eq!(requested, vec![None, Some("a".into()), Some("b".into())]);
2359 }
2360
2361 #[tokio::test]
2362 async fn collect_pages_stops_on_repeated_cursor() {
2363 let (result, _) =
2364 collect_from(vec![page(vec![1], Some("a")), page(vec![2], Some("a"))]).await;
2365
2366 assert_eq!(result.unwrap(), vec![1, 2]);
2367 }
2368
2369 #[tokio::test]
2370 async fn collect_pages_stops_on_empty_page_with_cursor() {
2371 let (result, _) =
2372 collect_from(vec![page(vec![1], Some("a")), page(vec![], Some("b"))]).await;
2373
2374 assert_eq!(result.unwrap(), vec![1]);
2375 }
2376
2377 #[tokio::test]
2378 async fn collect_pages_fails_after_max_pages() {
2379 let mut n = 0;
2380 let result = collect_pages(|_| {
2381 n += 1;
2382 let cursor = n.to_string();
2383 async move { Ok(page(vec![1], Some(&cursor))) }
2384 })
2385 .await;
2386
2387 assert!(matches!(
2388 result,
2389 Err(Error::TooManyPages { max: MAX_PAGES })
2390 ));
2391 assert_eq!(n, MAX_PAGES);
2392 }
2393
2394 #[test]
2395 fn time_offset_is_measured_from_the_middle_of_the_round_trip() {
2396 assert_eq!(estimate_time_offset(1000, 1100, 61_050), 60_000);
2398 assert_eq!(estimate_time_offset(1000, 1000, 400), -600);
2400 }
2401
2402 async fn spawn_http_server(
2407 body_for: fn(&str) -> String,
2408 ) -> (String, std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
2409 use tokio::io::{AsyncReadExt, AsyncWriteExt};
2410
2411 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
2412 let addr = listener.local_addr().unwrap();
2413 let requests = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
2414 let recorded = requests.clone();
2415 tokio::spawn(async move {
2416 while let Ok((mut tcp, _)) = listener.accept().await {
2417 let mut head = Vec::new();
2418 let mut buf = [0u8; 1024];
2419 while !head.windows(4).any(|w| w == b"\r\n\r\n") {
2420 match tcp.read(&mut buf).await {
2421 Ok(0) | Err(_) => break,
2422 Ok(n) => head.extend_from_slice(&buf[..n]),
2423 }
2424 }
2425 let text = String::from_utf8_lossy(&head).into_owned();
2426 let content_length: usize = text
2427 .lines()
2428 .find_map(|line| {
2429 let (name, value) = line.split_once(':')?;
2430 name.eq_ignore_ascii_case("content-length")
2431 .then(|| value.trim().parse().unwrap_or(0))
2432 })
2433 .unwrap_or(0);
2434 let header_len = head
2435 .windows(4)
2436 .position(|w| w == b"\r\n\r\n")
2437 .map_or(head.len(), |i| i + 4);
2438 while head.len() < header_len + content_length {
2439 match tcp.read(&mut buf).await {
2440 Ok(0) | Err(_) => break,
2441 Ok(n) => head.extend_from_slice(&buf[..n]),
2442 }
2443 }
2444 let request = String::from_utf8_lossy(&head).into_owned();
2445 recorded.lock().unwrap().push(request.clone());
2446 let body = body_for(&request);
2447 let response = format!(
2448 "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
2449 body.len()
2450 );
2451 let _ = tcp.write_all(response.as_bytes()).await;
2452 }
2453 });
2454 (format!("http://{addr}"), requests)
2455 }
2456
2457 const SERVER_AHEAD_MS: u64 = 60_000;
2459
2460 fn request_path(request: &str) -> &str {
2461 request.split_whitespace().nth(1).unwrap_or_default()
2462 }
2463
2464 fn header<'a>(request: &'a str, name: &str) -> Option<&'a str> {
2465 request.lines().find_map(|line| {
2466 let (n, value) = line.split_once(':')?;
2467 n.eq_ignore_ascii_case(name).then(|| value.trim())
2468 })
2469 }
2470
2471 fn server_time_body() -> String {
2472 let server_ms = timestamp() + SERVER_AHEAD_MS;
2473 format!(
2474 r#"{{"retCode":0,"retMsg":"OK","result":{{"timeSecond":"{}","timeNano":"{}"}},"retExtInfo":{{}},"time":{server_ms}}}"#,
2475 server_ms / 1000,
2476 server_ms * 1_000_000
2477 )
2478 }
2479
2480 fn strict_bybit(request: &str) -> String {
2484 let path = request_path(request);
2485 if path.starts_with("/v5/market/time") {
2486 return server_time_body();
2487 }
2488 let server_ms = (timestamp() + SERVER_AHEAD_MS) as i64;
2489 let ts: i64 = header(request, HEADER_X_BAPI_TIMESTAMP)
2490 .unwrap()
2491 .parse()
2492 .unwrap();
2493 let payload = match path.split_once('?') {
2494 Some((_, query)) => query.to_owned(),
2495 None => request
2496 .split_once("\r\n\r\n")
2497 .map(|(_, b)| b.to_owned())
2498 .unwrap_or_default(),
2499 };
2500 let expected = crate::hmac_sha256("secret", format!("{ts}key5000{payload}"));
2501 let sign_ok = header(request, HEADER_X_BAPI_SIGN) == Some(expected.as_str());
2502 if (ts - server_ms).abs() > 5_000 {
2503 r#"{"retCode":10002,"retMsg":"timestamp","result":{},"retExtInfo":{},"time":1}"#
2504 .to_owned()
2505 } else if !sign_ok {
2506 r#"{"retCode":10004,"retMsg":"sign","result":{},"retExtInfo":{},"time":1}"#.to_owned()
2507 } else {
2508 r#"{"retCode":0,"retMsg":"OK","result":{"list":[],"nextPageCursor":""},"retExtInfo":{},"time":1}"#.to_owned()
2509 }
2510 }
2511
2512 fn fake_bybit(request: &str) -> String {
2513 if request_path(request).starts_with("/v5/market/time") {
2514 server_time_body()
2515 } else {
2516 r#"{"retCode":10001,"retMsg":"test","result":{},"retExtInfo":{},"time":1}"#.to_owned()
2517 }
2518 }
2519
2520 fn paths(requests: &std::sync::Mutex<Vec<String>>) -> Vec<String> {
2521 requests
2522 .lock()
2523 .unwrap()
2524 .iter()
2525 .map(|r| request_path(r).split('?').next().unwrap().to_owned())
2526 .collect()
2527 }
2528
2529 #[tokio::test]
2530 async fn get_is_resigned_and_retried_after_a_timestamp_error() {
2531 let (url, requests) = spawn_http_server(strict_bybit).await;
2532 let client = Client::new(Config::new(url).credentials("key", "secret")).unwrap();
2533
2534 let params =
2535 GetOpenClosedOrdersParams::new(Category::Linear).with_symbol(String::from("BTCUSDT"));
2536 let result = client.get_open_closed_orders(¶ms).await;
2537
2538 assert!(result.is_ok(), "{result:?}");
2539 assert_eq!(
2540 paths(&requests),
2541 [
2542 "/v5/order/realtime",
2543 "/v5/market/time",
2544 "/v5/order/realtime"
2545 ]
2546 );
2547 assert!((client.time_offset() - SERVER_AHEAD_MS as i64).abs() < 1000);
2548 }
2549
2550 #[tokio::test]
2551 async fn post_is_resigned_and_retried_after_a_timestamp_error() {
2552 let (url, requests) = spawn_http_server(strict_bybit).await;
2553 let client = Client::new(Config::new(url).credentials("key", "secret")).unwrap();
2554
2555 let request = SetLeverageRequest::new(Category::Linear, String::from("BTCUSDT"), 10.into());
2556 let result = client.set_leverage(&request).await;
2557
2558 assert!(result.is_ok(), "{result:?}");
2559 assert_eq!(
2560 paths(&requests),
2561 [
2562 "/v5/position/set-leverage",
2563 "/v5/market/time",
2564 "/v5/position/set-leverage"
2565 ]
2566 );
2567 }
2568
2569 #[tokio::test]
2570 async fn timestamp_errors_are_returned_when_resync_is_disabled() {
2571 let (url, requests) = spawn_http_server(strict_bybit).await;
2572 let config = Config::new(url)
2573 .credentials("key", "secret")
2574 .resync_time_on_timestamp_error(false);
2575 let client = Client::new(config).unwrap();
2576
2577 let params =
2578 GetOpenClosedOrdersParams::new(Category::Linear).with_symbol(String::from("BTCUSDT"));
2579 let err = client.get_open_closed_orders(¶ms).await.unwrap_err();
2580
2581 assert!(err.is_timestamp_error());
2582 assert_eq!(paths(&requests), ["/v5/order/realtime"]);
2583 }
2584
2585 #[tokio::test]
2586 async fn signed_requests_use_the_synchronized_server_time() {
2587 let (url, requests) = spawn_http_server(fake_bybit).await;
2588 let client = Client::new(Config::new(url).credentials("key", "secret")).unwrap();
2589
2590 let offset = client.sync_time().await.unwrap();
2591 let _ = client.get_api_key_information().await;
2592
2593 let expected = SERVER_AHEAD_MS as i64;
2594 assert!((offset - expected).abs() < 1000, "offset {offset}");
2595 assert_eq!(client.time_offset(), offset);
2596
2597 let head = requests.lock().unwrap().last().cloned().unwrap();
2598 let sent: i64 = head
2599 .lines()
2600 .find_map(|line| {
2601 let (name, value) = line.split_once(':')?;
2602 name.eq_ignore_ascii_case(HEADER_X_BAPI_TIMESTAMP)
2603 .then(|| value.trim().parse().unwrap())
2604 })
2605 .expect("no X-BAPI-TIMESTAMP header");
2606 let skew = sent - (timestamp() as i64 + expected);
2607 assert!(
2608 skew.abs() < 1000,
2609 "signed timestamp is {skew} ms off the server clock"
2610 );
2611 }
2612
2613 #[tokio::test]
2614 async fn clones_share_the_clock_offset_and_the_rate_limiter() {
2615 let (url, _) = spawn_http_server(fake_bybit).await;
2616 let config = Config::new(url).rate_limiter(RateLimiterConfig::uniform(0.0, 2));
2617 let client = Client::new(config).unwrap();
2618 let clone = client.clone();
2619
2620 clone.sync_time().await.unwrap();
2621 assert_ne!(client.time_offset(), 0);
2622 assert_eq!(client.time_offset(), clone.time_offset());
2623
2624 client.get_server_time().await.unwrap();
2626 assert!(matches!(
2627 clone.get_server_time().await,
2628 Err(Error::RateLimitUnsatisfiable { .. })
2629 ));
2630 }
2631
2632 #[tokio::test]
2633 async fn requests_go_through_the_configured_proxy() {
2634 fn answer_time(_: &str) -> String {
2635 server_time_body()
2636 }
2637 let (proxy, requests) = spawn_http_server(answer_time).await;
2638 let config = Config::new("http://bybit.invalid").proxy(proxy);
2640 let client = Client::new(config).unwrap();
2641
2642 client.get_server_time().await.unwrap();
2643
2644 let request_line = requests.lock().unwrap()[0]
2645 .lines()
2646 .next()
2647 .unwrap()
2648 .to_owned();
2649 assert_eq!(
2650 request_line,
2651 "GET http://bybit.invalid/v5/market/time HTTP/1.1"
2652 );
2653 }
2654
2655 #[test]
2656 fn invalid_proxy_url_is_an_error() {
2657 let config = Config::new("http://bybit.invalid").proxy("not a url");
2658
2659 assert!(Client::new(config).is_err());
2660 }
2661
2662 #[tokio::test]
2663 async fn private_endpoint_without_credentials_returns_error() {
2664 let client = Client::new(Config::new("http://127.0.0.1:1")).unwrap();
2665
2666 let err = client.get_api_key_information().await.unwrap_err();
2667
2668 assert!(matches!(err, Error::MissingCredentials));
2669 }
2670}