1use std::{
21 cmp::Reverse,
22 collections::HashMap,
23 fmt::{Debug, Display},
24 num::NonZeroU32,
25 sync::{
26 Arc, LazyLock,
27 atomic::{AtomicBool, Ordering},
28 },
29};
30
31use ahash::{AHashMap, AHashSet};
32use chrono::{DateTime, Utc};
33use nautilus_common::cache::InstrumentLookupError;
34use nautilus_core::{
35 AtomicMap, AtomicTime, consts::NAUTILUS_USER_AGENT, env::get_or_env_var_opt, nanos::UnixNanos,
36 time::get_atomic_clock_realtime,
37};
38use nautilus_model::{
39 data::{Bar, BarType, FundingRateUpdate, OrderBookDeltas, TradeTick},
40 enums::{MarketStatusAction, OrderSide, OrderType, PositionSideSpecified, TimeInForce},
41 events::account::state::AccountState,
42 identifiers::{AccountId, ClientOrderId, InstrumentId, Symbol, VenueOrderId},
43 instruments::{Instrument, InstrumentAny},
44 reports::{FillReport, OrderStatusReport, PositionStatusReport},
45 types::{Price, Quantity},
46};
47use nautilus_network::{
48 http::{HttpClient, Method, USER_AGENT},
49 ratelimiter::quota::Quota,
50 retry::{RetryConfig, RetryManager},
51};
52use rust_decimal::Decimal;
53use serde::{Serialize, de::DeserializeOwned};
54use tokio_util::sync::CancellationToken;
55use ustr::Ustr;
56
57use super::{
58 error::{BybitCancelOrderError, BybitHttpError, BybitModifyOrderError, BybitSubmitOrderError},
59 models::{
60 BybitAccountDetailsResponse, BybitAccountInfoResponse, BybitBorrowResponse,
61 BybitEscrowSubMembersResponse, BybitFeeRate, BybitFeeRateResponse, BybitFundingResponse,
62 BybitInstrumentInverse, BybitInstrumentInverseResponse, BybitInstrumentLinear,
63 BybitInstrumentLinearResponse, BybitInstrumentOption, BybitInstrumentOptionResponse,
64 BybitInstrumentSpot, BybitInstrumentSpotResponse, BybitKlinesResponse,
65 BybitNoConvertRepayResponse, BybitOpenOrdersResponse, BybitOrder,
66 BybitOrderHistoryResponse, BybitOrderbookResponse, BybitPlaceOrderResponse,
67 BybitPositionListResponse, BybitServerTimeResponse, BybitSetLeverageResponse,
68 BybitSetMarginModeResponse, BybitSetTradingStopResponse, BybitSubApiKeyInfo,
69 BybitSubApiKeysResponse, BybitSubMember, BybitSubMembersPagedResponse,
70 BybitSubMembersResponse, BybitSwitchModeResponse, BybitTickerData, BybitTickerOption,
71 BybitTickersOptionResponse, BybitTradeHistoryResponse, BybitTradesResponse,
72 BybitUpdateMasterApiResponse, BybitUpdateSubApiResponse, BybitWalletBalanceResponse,
73 },
74 query::{
75 BybitAmendOrderParamsBuilder, BybitBatchAmendOrderEntryBuilder,
76 BybitBatchCancelOrderEntryBuilder, BybitBatchCancelOrderParamsBuilder,
77 BybitBatchPlaceOrderEntryBuilder, BybitBorrowParamsBuilder,
78 BybitCancelAllOrdersParamsBuilder, BybitCancelOrderParamsBuilder, BybitFeeRateParams,
79 BybitFeeRateParamsBuilder, BybitFundingParams, BybitFundingParamsBuilder,
80 BybitInstrumentsInfoParams, BybitKlinesParams, BybitKlinesParamsBuilder,
81 BybitNativeTpSlParams, BybitNoConvertRepayParamsBuilder, BybitOpenOrdersParamsBuilder,
82 BybitOrderHistoryParamsBuilder, BybitOrderbookParams, BybitOrderbookParamsBuilder,
83 BybitPlaceOrderParamsBuilder, BybitPositionListParams, BybitSetLeverageParamsBuilder,
84 BybitSetMarginModeParamsBuilder, BybitSetTradingStopParams, BybitSubApiKeysParams,
85 BybitSubMembersPageParams, BybitSwitchModeParamsBuilder, BybitTickersParams,
86 BybitTradeHistoryParams, BybitTradesParams, BybitTradesParamsBuilder,
87 BybitUpdateMasterApiParams, BybitUpdateSubApiParams, BybitWalletBalanceParams,
88 },
89};
90use crate::common::{
91 consts::{BYBIT_NAUTILUS_BROKER_ID, BYBIT_VENUE},
92 credential::{Credential, credential_env_vars},
93 enums::{
94 BybitAccountType, BybitBboSideType, BybitContractType, BybitEnvironment, BybitMarginMode,
95 BybitOpenOnly, BybitOrderFilter, BybitOrderSide, BybitOrderType, BybitPositionIdx,
96 BybitPositionMode, BybitProductType, BybitTpSlMode,
97 },
98 models::{BybitCursorListResponse, BybitErrorCheck, BybitResponseCheck},
99 parse::{
100 bar_spec_to_bybit_interval, make_bybit_symbol, map_time_in_force, parse_account_state,
101 parse_fill_report, parse_funding_rate, parse_inverse_instrument, parse_kline_bar,
102 parse_linear_instrument, parse_option_instrument, parse_order_status_report,
103 parse_orderbook, parse_position_status_report, parse_spot_instrument, parse_trade_tick,
104 spot_leverage, spot_market_unit, trigger_direction,
105 },
106 symbol::BybitSymbol,
107 urls::bybit_http_base_url,
108};
109
110const DEFAULT_RECV_WINDOW_MS: u64 = 5_000;
111
112trait BuilderResultExt<T> {
113 fn build_anyhow(self) -> anyhow::Result<T>;
114}
115
116impl<T, E: Display> BuilderResultExt<T> for Result<T, E> {
117 fn build_anyhow(self) -> anyhow::Result<T> {
118 self.map_err(|e| anyhow::anyhow!("{e}"))
119 }
120}
121
122const BYBIT_ORDER_REALTIME: &str = "/v5/order/realtime";
123const BYBIT_ORDER_HISTORY: &str = "/v5/order/history";
124
125pub static BYBIT_REST_QUOTA: LazyLock<Quota> = LazyLock::new(|| {
130 Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant")
131});
132
133pub static BYBIT_REPAY_QUOTA: LazyLock<Quota> = LazyLock::new(|| {
137 Quota::per_second(NonZeroU32::new(1).expect("non-zero")).expect("valid constant")
138});
139
140const BYBIT_GLOBAL_RATE_KEY: &str = "bybit:global";
141const BYBIT_REPAY_ROUTE_KEY: &str = "bybit:/v5/account/no-convert-repay";
142
143#[cfg_attr(
148 feature = "python",
149 pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.bybit", from_py_object)
150)]
151#[cfg_attr(
152 feature = "python",
153 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.bybit")
154)]
155#[derive(Clone)]
156pub struct BybitRawHttpClient {
157 base_url: String,
158 client: HttpClient,
159 credential: Option<Credential>,
160 recv_window_ms: u64,
161 retry_manager: RetryManager<BybitHttpError>,
162 cancellation_token: Arc<std::sync::Mutex<CancellationToken>>,
163}
164
165impl Default for BybitRawHttpClient {
166 fn default() -> Self {
167 Self::new(None, 60, 3, 1000, 10_000, DEFAULT_RECV_WINDOW_MS, None)
168 .expect("Failed to create default BybitRawHttpClient")
169 }
170}
171
172impl Debug for BybitRawHttpClient {
173 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
174 f.debug_struct(stringify!(BybitRawHttpClient))
175 .field("base_url", &self.base_url)
176 .field("has_credentials", &self.credential.is_some())
177 .field("recv_window_ms", &self.recv_window_ms)
178 .finish()
179 }
180}
181
182impl BybitRawHttpClient {
183 #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
185 pub fn cancel_all_requests(&self) {
186 self.cancellation_token
187 .lock()
188 .expect("cancellation token lock poisoned")
189 .cancel();
190 }
191
192 #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
195 pub fn reset_cancellation_token(&self) {
196 let mut guard = self
197 .cancellation_token
198 .lock()
199 .expect("cancellation token lock poisoned");
200 *guard = CancellationToken::new();
201 }
202
203 #[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
205 pub fn cancellation_token(&self) -> CancellationToken {
206 self.cancellation_token
207 .lock()
208 .expect("cancellation token lock poisoned")
209 .clone()
210 }
211
212 pub fn new(
218 base_url: Option<String>,
219 timeout_secs: u64,
220 max_retries: u32,
221 retry_delay_ms: u64,
222 retry_delay_max_ms: u64,
223 recv_window_ms: u64,
224 proxy_url: Option<String>,
225 ) -> Result<Self, BybitHttpError> {
226 let retry_config = RetryConfig {
227 max_retries,
228 initial_delay_ms: retry_delay_ms,
229 max_delay_ms: retry_delay_max_ms,
230 backoff_factor: 2.0,
231 jitter_ms: 1000,
232 operation_timeout_ms: Some(60_000),
233 immediate_first: false,
234 max_elapsed_ms: Some(180_000),
235 };
236
237 let retry_manager = RetryManager::new(retry_config);
238
239 Ok(Self {
240 base_url: base_url
241 .unwrap_or_else(|| bybit_http_base_url(BybitEnvironment::Mainnet).to_string()),
242 client: HttpClient::new(
243 Self::default_headers(),
244 vec![],
245 Self::rate_limiter_quotas(),
246 Some(*BYBIT_REST_QUOTA),
247 Some(timeout_secs),
248 proxy_url,
249 )
250 .map_err(|e| {
251 BybitHttpError::NetworkError(format!("Failed to create HTTP client: {e}"))
252 })?,
253 credential: None,
254 recv_window_ms,
255 retry_manager,
256 cancellation_token: Arc::new(std::sync::Mutex::new(CancellationToken::new())),
257 })
258 }
259
260 #[expect(clippy::too_many_arguments)]
266 pub fn with_credentials(
267 api_key: String,
268 api_secret: String,
269 base_url: Option<String>,
270 timeout_secs: u64,
271 max_retries: u32,
272 retry_delay_ms: u64,
273 retry_delay_max_ms: u64,
274 recv_window_ms: u64,
275 proxy_url: Option<String>,
276 ) -> Result<Self, BybitHttpError> {
277 let retry_config = RetryConfig {
278 max_retries,
279 initial_delay_ms: retry_delay_ms,
280 max_delay_ms: retry_delay_max_ms,
281 backoff_factor: 2.0,
282 jitter_ms: 1000,
283 operation_timeout_ms: Some(60_000),
284 immediate_first: false,
285 max_elapsed_ms: Some(180_000),
286 };
287
288 let retry_manager = RetryManager::new(retry_config);
289
290 Ok(Self {
291 base_url: base_url
292 .unwrap_or_else(|| bybit_http_base_url(BybitEnvironment::Mainnet).to_string()),
293 client: HttpClient::new(
294 Self::default_headers(),
295 vec![],
296 Self::rate_limiter_quotas(),
297 Some(*BYBIT_REST_QUOTA),
298 Some(timeout_secs),
299 proxy_url,
300 )
301 .map_err(|e| {
302 BybitHttpError::NetworkError(format!("Failed to create HTTP client: {e}"))
303 })?,
304 credential: Some(Credential::new(api_key, api_secret)),
305 recv_window_ms,
306 retry_manager,
307 cancellation_token: Arc::new(std::sync::Mutex::new(CancellationToken::new())),
308 })
309 }
310
311 #[expect(clippy::too_many_arguments)]
323 pub fn new_with_env(
324 api_key: Option<String>,
325 api_secret: Option<String>,
326 base_url: Option<String>,
327 demo: bool,
328 testnet: bool,
329 timeout_secs: u64,
330 max_retries: u32,
331 retry_delay_ms: u64,
332 retry_delay_max_ms: u64,
333 recv_window_ms: u64,
334 proxy_url: Option<String>,
335 ) -> Result<Self, BybitHttpError> {
336 let environment = if demo {
337 BybitEnvironment::Demo
338 } else if testnet {
339 BybitEnvironment::Testnet
340 } else {
341 BybitEnvironment::Mainnet
342 };
343 let (key_var, secret_var) = credential_env_vars(environment);
344 let key = get_or_env_var_opt(api_key, key_var);
345 let secret = get_or_env_var_opt(api_secret, secret_var);
346
347 if let (Some(k), Some(s)) = (key, secret) {
348 Self::with_credentials(
349 k,
350 s,
351 base_url,
352 timeout_secs,
353 max_retries,
354 retry_delay_ms,
355 retry_delay_max_ms,
356 recv_window_ms,
357 proxy_url,
358 )
359 } else {
360 Self::new(
361 base_url,
362 timeout_secs,
363 max_retries,
364 retry_delay_ms,
365 retry_delay_max_ms,
366 recv_window_ms,
367 proxy_url,
368 )
369 }
370 }
371
372 fn default_headers() -> HashMap<String, String> {
373 HashMap::from([
374 (USER_AGENT.to_string(), NAUTILUS_USER_AGENT.to_string()),
375 (
376 "X-Referer".to_string(),
377 BYBIT_NAUTILUS_BROKER_ID.to_string(),
378 ),
379 ])
380 }
381
382 fn rate_limiter_quotas() -> Vec<(String, Quota)> {
383 vec![
384 (BYBIT_GLOBAL_RATE_KEY.to_string(), *BYBIT_REST_QUOTA),
385 (BYBIT_REPAY_ROUTE_KEY.to_string(), *BYBIT_REPAY_QUOTA),
386 ]
387 }
388
389 fn rate_limit_keys(endpoint: &str) -> Vec<String> {
390 let normalized = endpoint.split('?').next().unwrap_or(endpoint);
391 let route = format!("bybit:{normalized}");
392
393 vec![BYBIT_GLOBAL_RATE_KEY.to_string(), route]
394 }
395
396 fn sign_request(
397 &self,
398 timestamp: &str,
399 params: Option<&str>,
400 ) -> Result<HashMap<String, String>, BybitHttpError> {
401 let credential = self
402 .credential
403 .as_ref()
404 .ok_or(BybitHttpError::MissingCredentials)?;
405
406 let signature = credential.sign_with_payload(timestamp, self.recv_window_ms, params);
407
408 let mut headers = HashMap::new();
409 headers.insert(
410 "X-BAPI-API-KEY".to_string(),
411 credential.api_key().to_string(),
412 );
413 headers.insert("X-BAPI-TIMESTAMP".to_string(), timestamp.to_string());
414 headers.insert("X-BAPI-SIGN".to_string(), signature);
415 headers.insert(
416 "X-BAPI-RECV-WINDOW".to_string(),
417 self.recv_window_ms.to_string(),
418 );
419
420 Ok(headers)
421 }
422
423 async fn send_request<T: DeserializeOwned + BybitResponseCheck, P: Serialize>(
424 &self,
425 method: Method,
426 endpoint: &str,
427 params: Option<&P>,
428 body: Option<Vec<u8>>,
429 authenticate: bool,
430 ) -> Result<T, BybitHttpError> {
431 let endpoint = endpoint.to_string();
432 let url = format!("{}{endpoint}", self.base_url);
433 let method_clone = method.clone();
434 let body_clone = body.clone();
435
436 let params_str = if method == Method::GET {
438 params
439 .map(serde_urlencoded::to_string)
440 .transpose()
441 .map_err(|e| {
442 BybitHttpError::JsonError(format!("Failed to serialize params: {e}"))
443 })?
444 } else {
445 None
446 };
447
448 let operation = || {
449 let url = url.clone();
450 let method = method_clone.clone();
451 let body = body_clone.clone();
452 let endpoint = endpoint.clone();
453 let params_str = params_str.clone();
454
455 async move {
456 let mut headers = Self::default_headers();
457
458 if authenticate {
459 let timestamp = get_atomic_clock_realtime().get_time_ms().to_string();
460
461 let sign_payload = if method == Method::GET {
462 params_str.as_deref()
463 } else {
464 body.as_ref().and_then(|b| std::str::from_utf8(b).ok())
465 };
466
467 let auth_headers = self.sign_request(×tamp, sign_payload)?;
468 headers.extend(auth_headers);
469 }
470
471 if method == Method::POST || method == Method::PUT {
472 headers.insert("Content-Type".to_string(), "application/json".to_string());
473 }
474
475 let full_url = if let Some(ref query) = params_str {
476 if query.is_empty() {
477 url
478 } else {
479 format!("{url}?{query}")
480 }
481 } else {
482 url
483 };
484
485 let rate_limit_keys = Self::rate_limit_keys(&endpoint);
486
487 let response = self
488 .client
489 .request(
490 method,
491 full_url,
492 None,
493 Some(headers),
494 body,
495 None,
496 Some(rate_limit_keys),
497 )
498 .await?;
499
500 if response.status.as_u16() >= 400 {
501 let body = String::from_utf8_lossy(&response.body).to_string();
502 return Err(BybitHttpError::UnexpectedStatus {
503 status: response.status.as_u16(),
504 body,
505 });
506 }
507
508 match serde_json::from_slice::<T>(&response.body) {
510 Ok(result) => {
511 if result.ret_code() != 0 {
513 return Err(BybitHttpError::BybitError {
514 error_code: result.ret_code() as i32,
515 message: result.ret_msg().to_string(),
516 });
517 }
518 Ok(result)
519 }
520 Err(json_err) => {
521 if let Ok(error_check) =
524 serde_json::from_slice::<BybitErrorCheck>(&response.body)
525 && error_check.ret_code != 0
526 {
527 return Err(BybitHttpError::BybitError {
528 error_code: error_check.ret_code as i32,
529 message: error_check.ret_msg,
530 });
531 }
532 Err(json_err.into())
534 }
535 }
536 }
537 };
538
539 let should_retry = |error: &BybitHttpError| -> bool {
540 match error {
541 BybitHttpError::NetworkError(_) => true,
542 BybitHttpError::UnexpectedStatus { status, .. } => *status == 429 || *status >= 500,
543 _ => false,
544 }
545 };
546
547 let create_error = |msg: String| -> BybitHttpError {
548 if msg == "canceled" {
549 BybitHttpError::Canceled("Adapter disconnecting or shutting down".to_string())
550 } else {
551 BybitHttpError::NetworkError(msg)
552 }
553 };
554
555 let token = self.cancellation_token();
556
557 self.retry_manager
558 .execute_with_retry_with_cancel(
559 endpoint.as_str(),
560 operation,
561 should_retry,
562 create_error,
563 &token,
564 )
565 .await
566 }
567
568 #[cfg(test)]
569 fn build_path<S: Serialize>(base: &str, params: &S) -> Result<String, BybitHttpError> {
570 let query = serde_urlencoded::to_string(params)
571 .map_err(|e| BybitHttpError::JsonError(e.to_string()))?;
572
573 if query.is_empty() {
574 Ok(base.to_owned())
575 } else {
576 Ok(format!("{base}?{query}"))
577 }
578 }
579
580 pub async fn get_server_time(&self) -> Result<BybitServerTimeResponse, BybitHttpError> {
590 self.send_request::<_, ()>(Method::GET, "/v5/market/time", None, None, false)
591 .await
592 }
593
594 pub async fn get_instruments<T: DeserializeOwned + BybitResponseCheck>(
604 &self,
605 params: &BybitInstrumentsInfoParams,
606 ) -> Result<T, BybitHttpError> {
607 self.send_request(
608 Method::GET,
609 "/v5/market/instruments-info",
610 Some(params),
611 None,
612 false,
613 )
614 .await
615 }
616
617 pub async fn get_instruments_spot(
627 &self,
628 params: &BybitInstrumentsInfoParams,
629 ) -> Result<BybitInstrumentSpotResponse, BybitHttpError> {
630 self.get_instruments(params).await
631 }
632
633 pub async fn get_instruments_linear(
643 &self,
644 params: &BybitInstrumentsInfoParams,
645 ) -> Result<BybitInstrumentLinearResponse, BybitHttpError> {
646 self.get_instruments(params).await
647 }
648
649 pub async fn get_instruments_inverse(
659 &self,
660 params: &BybitInstrumentsInfoParams,
661 ) -> Result<BybitInstrumentInverseResponse, BybitHttpError> {
662 self.get_instruments(params).await
663 }
664
665 pub async fn get_instruments_option(
675 &self,
676 params: &BybitInstrumentsInfoParams,
677 ) -> Result<BybitInstrumentOptionResponse, BybitHttpError> {
678 self.get_instruments(params).await
679 }
680
681 pub async fn get_klines(
691 &self,
692 params: &BybitKlinesParams,
693 ) -> Result<BybitKlinesResponse, BybitHttpError> {
694 self.send_request(Method::GET, "/v5/market/kline", Some(params), None, false)
695 .await
696 }
697
698 pub async fn get_recent_trades(
708 &self,
709 params: &BybitTradesParams,
710 ) -> Result<BybitTradesResponse, BybitHttpError> {
711 self.send_request(
712 Method::GET,
713 "/v5/market/recent-trade",
714 Some(params),
715 None,
716 false,
717 )
718 .await
719 }
720
721 pub async fn get_funding_history(
731 &self,
732 params: &BybitFundingParams,
733 ) -> Result<BybitFundingResponse, BybitHttpError> {
734 self.send_request(
735 Method::GET,
736 "/v5/market/funding/history",
737 Some(params),
738 None,
739 false,
740 )
741 .await
742 }
743
744 pub async fn get_orderbook(
754 &self,
755 params: &BybitOrderbookParams,
756 ) -> Result<BybitOrderbookResponse, BybitHttpError> {
757 self.send_request(
758 Method::GET,
759 "/v5/market/orderbook",
760 Some(params),
761 None,
762 false,
763 )
764 .await
765 }
766
767 #[expect(clippy::too_many_arguments)]
781 pub async fn get_open_orders(
782 &self,
783 category: BybitProductType,
784 symbol: Option<String>,
785 base_coin: Option<String>,
786 settle_coin: Option<String>,
787 order_id: Option<String>,
788 order_link_id: Option<String>,
789 open_only: Option<BybitOpenOnly>,
790 order_filter: Option<BybitOrderFilter>,
791 limit: Option<u32>,
792 cursor: Option<String>,
793 ) -> Result<BybitOpenOrdersResponse, BybitHttpError> {
794 let mut builder = BybitOpenOrdersParamsBuilder::default();
795 builder.category(category);
796
797 if let Some(s) = symbol {
798 builder.symbol(s);
799 }
800
801 if let Some(bc) = base_coin {
802 builder.base_coin(bc);
803 }
804
805 if let Some(sc) = settle_coin {
806 builder.settle_coin(sc);
807 }
808
809 if let Some(oi) = order_id {
810 builder.order_id(oi);
811 }
812
813 if let Some(ol) = order_link_id {
814 builder.order_link_id(ol);
815 }
816
817 if let Some(oo) = open_only {
818 builder.open_only(oo);
819 }
820
821 if let Some(of) = order_filter {
822 builder.order_filter(of);
823 }
824
825 if let Some(l) = limit {
826 builder.limit(l);
827 }
828
829 if let Some(c) = cursor {
830 builder.cursor(c);
831 }
832
833 let params = builder
834 .build()
835 .expect("Failed to build BybitOpenOrdersParams");
836
837 self.send_request(Method::GET, BYBIT_ORDER_REALTIME, Some(¶ms), None, true)
838 .await
839 }
840
841 pub async fn place_order(
851 &self,
852 request: &serde_json::Value,
853 ) -> Result<BybitPlaceOrderResponse, BybitHttpError> {
854 let body = serde_json::to_vec(request)?;
855 self.send_request::<_, ()>(Method::POST, "/v5/order/create", None, Some(body), true)
856 .await
857 }
858
859 pub async fn get_wallet_balance(
869 &self,
870 params: &BybitWalletBalanceParams,
871 ) -> Result<BybitWalletBalanceResponse, BybitHttpError> {
872 self.send_request(
873 Method::GET,
874 "/v5/account/wallet-balance",
875 Some(params),
876 None,
877 true,
878 )
879 .await
880 }
881
882 pub async fn get_account_info(&self) -> Result<BybitAccountInfoResponse, BybitHttpError> {
892 self.send_request::<_, ()>(Method::GET, "/v5/account/info", None, None, true)
893 .await
894 }
895
896 pub async fn get_account_details(&self) -> Result<BybitAccountDetailsResponse, BybitHttpError> {
906 self.send_request::<_, ()>(Method::GET, "/v5/user/query-api", None, None, true)
907 .await
908 }
909
910 pub async fn update_sub_api_key(
920 &self,
921 params: &BybitUpdateSubApiParams,
922 ) -> Result<BybitUpdateSubApiResponse, BybitHttpError> {
923 let body = serde_json::to_vec(params)?;
924 self.send_request::<_, ()>(
925 Method::POST,
926 "/v5/user/update-sub-api",
927 None,
928 Some(body),
929 true,
930 )
931 .await
932 }
933
934 pub async fn update_master_api_key(
944 &self,
945 params: &BybitUpdateMasterApiParams,
946 ) -> Result<BybitUpdateMasterApiResponse, BybitHttpError> {
947 let body = serde_json::to_vec(params)?;
948 self.send_request::<_, ()>(Method::POST, "/v5/user/update-api", None, Some(body), true)
949 .await
950 }
951
952 pub async fn get_sub_members(&self) -> Result<BybitSubMembersResponse, BybitHttpError> {
962 self.send_request::<_, ()>(Method::GET, "/v5/user/query-sub-members", None, None, true)
963 .await
964 }
965
966 pub async fn get_sub_members_paged(
976 &self,
977 params: &BybitSubMembersPageParams,
978 ) -> Result<BybitSubMembersPagedResponse, BybitHttpError> {
979 self.send_request(Method::GET, "/v5/user/submembers", Some(params), None, true)
980 .await
981 }
982
983 pub async fn get_escrow_sub_members(
993 &self,
994 params: &BybitSubMembersPageParams,
995 ) -> Result<BybitEscrowSubMembersResponse, BybitHttpError> {
996 self.send_request(
997 Method::GET,
998 "/v5/user/escrow_sub_members",
999 Some(params),
1000 None,
1001 true,
1002 )
1003 .await
1004 }
1005
1006 pub async fn get_sub_api_keys(
1016 &self,
1017 params: &BybitSubApiKeysParams,
1018 ) -> Result<BybitSubApiKeysResponse, BybitHttpError> {
1019 self.send_request(
1020 Method::GET,
1021 "/v5/user/sub-apikeys",
1022 Some(params),
1023 None,
1024 true,
1025 )
1026 .await
1027 }
1028
1029 pub async fn fetch_all_sub_members_paged(
1036 &self,
1037 page_size: Option<u32>,
1038 ) -> Result<Vec<BybitSubMember>, BybitHttpError> {
1039 let mut members = Vec::new();
1040 let mut cursor: Option<String> = None;
1041
1042 loop {
1043 let params = BybitSubMembersPageParams {
1044 page_size,
1045 next_cursor: cursor.take(),
1046 };
1047 let mut page = self.get_sub_members_paged(¶ms).await?;
1048 let next = page.result.continuation_cursor().map(str::to_owned);
1049 members.append(&mut page.result.sub_members);
1050
1051 match next {
1052 Some(c) => cursor = Some(c),
1053 None => break,
1054 }
1055 }
1056
1057 Ok(members)
1058 }
1059
1060 pub async fn fetch_all_escrow_sub_members(
1068 &self,
1069 page_size: Option<u32>,
1070 ) -> Result<Vec<BybitSubMember>, BybitHttpError> {
1071 let mut members = Vec::new();
1072 let mut cursor: Option<String> = None;
1073
1074 loop {
1075 let params = BybitSubMembersPageParams {
1076 page_size,
1077 next_cursor: cursor.take(),
1078 };
1079 let mut page = self.get_escrow_sub_members(¶ms).await?;
1080 let next = page.result.continuation_cursor().map(str::to_owned);
1081 members.append(&mut page.result.sub_members);
1082
1083 match next {
1084 Some(c) => cursor = Some(c),
1085 None => break,
1086 }
1087 }
1088
1089 Ok(members)
1090 }
1091
1092 pub async fn fetch_all_sub_api_keys(
1100 &self,
1101 sub_member_id: impl Into<String>,
1102 limit: Option<u32>,
1103 ) -> Result<Vec<BybitSubApiKeyInfo>, BybitHttpError> {
1104 let sub_member_id = sub_member_id.into();
1105 let mut keys = Vec::new();
1106 let mut cursor: Option<String> = None;
1107
1108 loop {
1109 let params = BybitSubApiKeysParams {
1110 sub_member_id: sub_member_id.clone(),
1111 limit,
1112 cursor: cursor.take(),
1113 };
1114 let mut page = self.get_sub_api_keys(¶ms).await?;
1115 let next = page.result.continuation_cursor().map(str::to_owned);
1116 keys.append(&mut page.result.keys);
1117
1118 match next {
1119 Some(c) => cursor = Some(c),
1120 None => break,
1121 }
1122 }
1123
1124 Ok(keys)
1125 }
1126
1127 pub async fn get_fee_rate(
1137 &self,
1138 params: &BybitFeeRateParams,
1139 ) -> Result<BybitFeeRateResponse, BybitHttpError> {
1140 self.send_request(
1141 Method::GET,
1142 "/v5/account/fee-rate",
1143 Some(params),
1144 None,
1145 true,
1146 )
1147 .await
1148 }
1149
1150 pub async fn set_margin_mode(
1167 &self,
1168 margin_mode: BybitMarginMode,
1169 ) -> Result<BybitSetMarginModeResponse, BybitHttpError> {
1170 let params = BybitSetMarginModeParamsBuilder::default()
1171 .set_margin_mode(margin_mode)
1172 .build()
1173 .expect("Failed to build BybitSetMarginModeParams");
1174
1175 let body = serde_json::to_vec(¶ms)?;
1176 self.send_request::<_, ()>(
1177 Method::POST,
1178 "/v5/account/set-margin-mode",
1179 None,
1180 Some(body),
1181 true,
1182 )
1183 .await
1184 }
1185
1186 pub async fn set_leverage(
1203 &self,
1204 product_type: BybitProductType,
1205 symbol: &str,
1206 buy_leverage: &str,
1207 sell_leverage: &str,
1208 ) -> Result<BybitSetLeverageResponse, BybitHttpError> {
1209 let params = BybitSetLeverageParamsBuilder::default()
1210 .category(product_type)
1211 .symbol(symbol.to_string())
1212 .buy_leverage(buy_leverage.to_string())
1213 .sell_leverage(sell_leverage.to_string())
1214 .build()
1215 .expect("Failed to build BybitSetLeverageParams");
1216
1217 let body = serde_json::to_vec(¶ms)?;
1218 self.send_request::<_, ()>(
1219 Method::POST,
1220 "/v5/position/set-leverage",
1221 None,
1222 Some(body),
1223 true,
1224 )
1225 .await
1226 }
1227
1228 pub async fn switch_mode(
1245 &self,
1246 product_type: BybitProductType,
1247 mode: BybitPositionMode,
1248 symbol: Option<String>,
1249 coin: Option<String>,
1250 ) -> Result<BybitSwitchModeResponse, BybitHttpError> {
1251 let mut builder = BybitSwitchModeParamsBuilder::default();
1252 builder.category(product_type);
1253 builder.mode(mode);
1254
1255 if let Some(s) = symbol {
1256 builder.symbol(s);
1257 }
1258
1259 if let Some(c) = coin {
1260 builder.coin(c);
1261 }
1262
1263 let params = builder
1264 .build()
1265 .expect("Failed to build BybitSwitchModeParams");
1266
1267 let body = serde_json::to_vec(¶ms)?;
1268 self.send_request::<_, ()>(
1269 Method::POST,
1270 "/v5/position/switch-mode",
1271 None,
1272 Some(body),
1273 true,
1274 )
1275 .await
1276 }
1277
1278 pub async fn set_trading_stop(
1291 &self,
1292 params: &BybitSetTradingStopParams,
1293 ) -> Result<BybitSetTradingStopResponse, BybitHttpError> {
1294 let body = serde_json::to_vec(params)?;
1295 self.send_request::<_, ()>(
1296 Method::POST,
1297 "/v5/position/trading-stop",
1298 None,
1299 Some(body),
1300 true,
1301 )
1302 .await
1303 }
1304
1305 pub async fn borrow(
1322 &self,
1323 coin: &str,
1324 amount: &str,
1325 ) -> Result<BybitBorrowResponse, BybitHttpError> {
1326 let params = BybitBorrowParamsBuilder::default()
1327 .coin(coin.to_string())
1328 .amount(amount.to_string())
1329 .build()
1330 .expect("Failed to build BybitBorrowParams");
1331
1332 let body = serde_json::to_vec(¶ms)?;
1333 self.send_request::<_, ()>(Method::POST, "/v5/account/borrow", None, Some(body), true)
1334 .await
1335 }
1336
1337 pub async fn no_convert_repay(
1355 &self,
1356 coin: &str,
1357 amount: Option<&str>,
1358 ) -> Result<BybitNoConvertRepayResponse, BybitHttpError> {
1359 let mut builder = BybitNoConvertRepayParamsBuilder::default();
1360 builder.coin(coin.to_string());
1361
1362 if let Some(amt) = amount {
1363 builder.amount(amt.to_string());
1364 }
1365
1366 let params = builder
1367 .build()
1368 .expect("Failed to build BybitNoConvertRepayParams");
1369
1370 if let Ok(params_json) = serde_json::to_string(¶ms) {
1371 log::debug!("Repay request params: {params_json}");
1372 }
1373
1374 let body = serde_json::to_vec(¶ms)?;
1375 let result = self
1376 .send_request::<_, ()>(
1377 Method::POST,
1378 "/v5/account/no-convert-repay",
1379 None,
1380 Some(body),
1381 true,
1382 )
1383 .await;
1384
1385 if let Err(ref e) = result
1386 && let Ok(params_json) = serde_json::to_string(¶ms)
1387 {
1388 log::error!("Repay request failed with params {params_json}: {e}");
1389 }
1390
1391 result
1392 }
1393
1394 pub async fn get_tickers<T: DeserializeOwned + BybitResponseCheck>(
1404 &self,
1405 params: &BybitTickersParams,
1406 ) -> Result<T, BybitHttpError> {
1407 self.send_request(Method::GET, "/v5/market/tickers", Some(params), None, false)
1408 .await
1409 }
1410
1411 pub async fn get_trade_history(
1421 &self,
1422 params: &BybitTradeHistoryParams,
1423 ) -> Result<BybitTradeHistoryResponse, BybitHttpError> {
1424 self.send_request(Method::GET, "/v5/execution/list", Some(params), None, true)
1425 .await
1426 }
1427
1428 pub async fn get_positions(
1441 &self,
1442 params: &BybitPositionListParams,
1443 ) -> Result<BybitPositionListResponse, BybitHttpError> {
1444 self.send_request(Method::GET, "/v5/position/list", Some(params), None, true)
1445 .await
1446 }
1447
1448 #[must_use]
1450 pub fn base_url(&self) -> &str {
1451 &self.base_url
1452 }
1453
1454 #[must_use]
1456 pub fn recv_window_ms(&self) -> u64 {
1457 self.recv_window_ms
1458 }
1459
1460 #[must_use]
1462 pub fn credential(&self) -> Option<&Credential> {
1463 self.credential.as_ref()
1464 }
1465}
1466
1467#[cfg_attr(
1469 feature = "python",
1470 pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.bybit", from_py_object)
1471)]
1472#[cfg_attr(
1473 feature = "python",
1474 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.bybit")
1475)]
1476pub struct BybitHttpClient {
1481 pub(crate) inner: Arc<BybitRawHttpClient>,
1482 pub(crate) instruments_cache: Arc<AtomicMap<Ustr, InstrumentAny>>,
1483 clock: &'static AtomicTime,
1484 cache_initialized: Arc<AtomicBool>,
1485 use_spot_position_reports: Arc<AtomicBool>,
1486}
1487
1488impl Clone for BybitHttpClient {
1489 fn clone(&self) -> Self {
1490 Self {
1491 inner: self.inner.clone(),
1492 instruments_cache: self.instruments_cache.clone(),
1493 cache_initialized: self.cache_initialized.clone(),
1494 use_spot_position_reports: self.use_spot_position_reports.clone(),
1495 clock: self.clock,
1496 }
1497 }
1498}
1499
1500impl Default for BybitHttpClient {
1501 fn default() -> Self {
1502 Self::new(None, 60, 3, 1000, 10_000, DEFAULT_RECV_WINDOW_MS, None)
1503 .expect("Failed to create default BybitHttpClient")
1504 }
1505}
1506
1507impl Debug for BybitHttpClient {
1508 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1509 f.debug_struct(stringify!(BybitHttpClient))
1510 .field("inner", &self.inner)
1511 .finish()
1512 }
1513}
1514
1515impl BybitHttpClient {
1516 pub fn new(
1522 base_url: Option<String>,
1523 timeout_secs: u64,
1524 max_retries: u32,
1525 retry_delay_ms: u64,
1526 retry_delay_max_ms: u64,
1527 recv_window_ms: u64,
1528 proxy_url: Option<String>,
1529 ) -> Result<Self, BybitHttpError> {
1530 Ok(Self {
1531 inner: Arc::new(BybitRawHttpClient::new(
1532 base_url,
1533 timeout_secs,
1534 max_retries,
1535 retry_delay_ms,
1536 retry_delay_max_ms,
1537 recv_window_ms,
1538 proxy_url,
1539 )?),
1540 instruments_cache: Arc::new(AtomicMap::new()),
1541 cache_initialized: Arc::new(AtomicBool::new(false)),
1542 use_spot_position_reports: Arc::new(AtomicBool::new(false)),
1543 clock: get_atomic_clock_realtime(),
1544 })
1545 }
1546
1547 #[expect(clippy::too_many_arguments)]
1553 pub fn with_credentials(
1554 api_key: String,
1555 api_secret: String,
1556 base_url: Option<String>,
1557 timeout_secs: u64,
1558 max_retries: u32,
1559 retry_delay_ms: u64,
1560 retry_delay_max_ms: u64,
1561 recv_window_ms: u64,
1562 proxy_url: Option<String>,
1563 ) -> Result<Self, BybitHttpError> {
1564 Ok(Self {
1565 inner: Arc::new(BybitRawHttpClient::with_credentials(
1566 api_key,
1567 api_secret,
1568 base_url,
1569 timeout_secs,
1570 max_retries,
1571 retry_delay_ms,
1572 retry_delay_max_ms,
1573 recv_window_ms,
1574 proxy_url,
1575 )?),
1576 instruments_cache: Arc::new(AtomicMap::new()),
1577 cache_initialized: Arc::new(AtomicBool::new(false)),
1578 use_spot_position_reports: Arc::new(AtomicBool::new(false)),
1579 clock: get_atomic_clock_realtime(),
1580 })
1581 }
1582
1583 #[expect(clippy::too_many_arguments)]
1596 pub fn new_with_env(
1597 api_key: Option<String>,
1598 api_secret: Option<String>,
1599 base_url: Option<String>,
1600 demo: bool,
1601 testnet: bool,
1602 timeout_secs: u64,
1603 max_retries: u32,
1604 retry_delay_ms: u64,
1605 retry_delay_max_ms: u64,
1606 recv_window_ms: u64,
1607 proxy_url: Option<String>,
1608 ) -> Result<Self, BybitHttpError> {
1609 let environment = if demo {
1610 BybitEnvironment::Demo
1611 } else if testnet {
1612 BybitEnvironment::Testnet
1613 } else {
1614 BybitEnvironment::Mainnet
1615 };
1616 let (key_var, secret_var) = credential_env_vars(environment);
1617 let key = get_or_env_var_opt(api_key, key_var);
1618 let secret = get_or_env_var_opt(api_secret, secret_var);
1619
1620 match (key, secret) {
1621 (Some(k), Some(s)) => Self::with_credentials(
1622 k,
1623 s,
1624 base_url,
1625 timeout_secs,
1626 max_retries,
1627 retry_delay_ms,
1628 retry_delay_max_ms,
1629 recv_window_ms,
1630 proxy_url,
1631 ),
1632 _ => Self::new(
1633 base_url,
1634 timeout_secs,
1635 max_retries,
1636 retry_delay_ms,
1637 retry_delay_max_ms,
1638 recv_window_ms,
1639 proxy_url,
1640 ),
1641 }
1642 }
1643
1644 #[must_use]
1645 pub fn base_url(&self) -> &str {
1646 self.inner.base_url()
1647 }
1648
1649 #[must_use]
1650 pub fn recv_window_ms(&self) -> u64 {
1651 self.inner.recv_window_ms()
1652 }
1653
1654 #[must_use]
1655 pub fn credential(&self) -> Option<&Credential> {
1656 self.inner.credential()
1657 }
1658
1659 pub fn set_use_spot_position_reports(&self, use_spot_position_reports: bool) {
1660 self.use_spot_position_reports
1661 .store(use_spot_position_reports, Ordering::Relaxed);
1662 }
1663
1664 pub fn cancel_all_requests(&self) {
1665 self.inner.cancel_all_requests();
1666 }
1667
1668 pub fn reset_cancellation_token(&self) {
1669 self.inner.reset_cancellation_token();
1670 }
1671
1672 pub fn cancellation_token(&self) -> CancellationToken {
1673 self.inner.cancellation_token()
1674 }
1675
1676 pub fn cache_instrument(&self, instrument: InstrumentAny) {
1678 self.instruments_cache
1679 .insert(instrument.symbol().inner(), instrument);
1680 self.cache_initialized.store(true, Ordering::Release);
1681 }
1682
1683 pub fn cache_instruments(&self, instruments: &[InstrumentAny]) {
1685 self.instruments_cache.rcu(|m| {
1686 for instrument in instruments {
1687 m.insert(instrument.symbol().inner(), instrument.clone());
1688 }
1689 });
1690 self.cache_initialized.store(true, Ordering::Release);
1691 }
1692
1693 pub fn get_instrument(&self, symbol: &Ustr) -> Option<InstrumentAny> {
1694 self.instruments_cache.get_cloned(symbol)
1695 }
1696
1697 fn instrument_from_cache(&self, symbol: &Symbol) -> anyhow::Result<InstrumentAny> {
1698 self.get_instrument(&symbol.inner()).ok_or_else(|| {
1699 anyhow::anyhow!(
1700 "Instrument {symbol} not found in cache, ensure instruments loaded first"
1701 )
1702 })
1703 }
1704
1705 #[must_use]
1706 fn generate_ts_init(&self) -> UnixNanos {
1707 self.clock.get_time_ns()
1708 }
1709
1710 pub async fn get_server_time(&self) -> Result<BybitServerTimeResponse, BybitHttpError> {
1722 self.inner.get_server_time().await
1723 }
1724
1725 pub async fn get_instruments<T: DeserializeOwned + BybitResponseCheck>(
1737 &self,
1738 params: &BybitInstrumentsInfoParams,
1739 ) -> Result<T, BybitHttpError> {
1740 self.inner.get_instruments(params).await
1741 }
1742
1743 pub async fn get_instruments_spot(
1755 &self,
1756 params: &BybitInstrumentsInfoParams,
1757 ) -> Result<BybitInstrumentSpotResponse, BybitHttpError> {
1758 self.inner.get_instruments_spot(params).await
1759 }
1760
1761 pub async fn get_instruments_linear(
1773 &self,
1774 params: &BybitInstrumentsInfoParams,
1775 ) -> Result<BybitInstrumentLinearResponse, BybitHttpError> {
1776 self.inner.get_instruments_linear(params).await
1777 }
1778
1779 pub async fn get_instruments_inverse(
1791 &self,
1792 params: &BybitInstrumentsInfoParams,
1793 ) -> Result<BybitInstrumentInverseResponse, BybitHttpError> {
1794 self.inner.get_instruments_inverse(params).await
1795 }
1796
1797 pub async fn get_instruments_option(
1809 &self,
1810 params: &BybitInstrumentsInfoParams,
1811 ) -> Result<BybitInstrumentOptionResponse, BybitHttpError> {
1812 self.inner.get_instruments_option(params).await
1813 }
1814
1815 pub async fn get_klines(
1827 &self,
1828 params: &BybitKlinesParams,
1829 ) -> Result<BybitKlinesResponse, BybitHttpError> {
1830 self.inner.get_klines(params).await
1831 }
1832
1833 pub async fn get_recent_trades(
1845 &self,
1846 params: &BybitTradesParams,
1847 ) -> Result<BybitTradesResponse, BybitHttpError> {
1848 self.inner.get_recent_trades(params).await
1849 }
1850
1851 #[expect(clippy::too_many_arguments)]
1863 pub async fn get_open_orders(
1864 &self,
1865 category: BybitProductType,
1866 symbol: Option<String>,
1867 base_coin: Option<String>,
1868 settle_coin: Option<String>,
1869 order_id: Option<String>,
1870 order_link_id: Option<String>,
1871 open_only: Option<BybitOpenOnly>,
1872 order_filter: Option<BybitOrderFilter>,
1873 limit: Option<u32>,
1874 cursor: Option<String>,
1875 ) -> Result<BybitOpenOrdersResponse, BybitHttpError> {
1876 self.inner
1877 .get_open_orders(
1878 category,
1879 symbol,
1880 base_coin,
1881 settle_coin,
1882 order_id,
1883 order_link_id,
1884 open_only,
1885 order_filter,
1886 limit,
1887 cursor,
1888 )
1889 .await
1890 }
1891
1892 pub async fn place_order(
1904 &self,
1905 request: &serde_json::Value,
1906 ) -> Result<BybitPlaceOrderResponse, BybitHttpError> {
1907 self.inner.place_order(request).await
1908 }
1909
1910 pub async fn get_wallet_balance(
1922 &self,
1923 params: &BybitWalletBalanceParams,
1924 ) -> Result<BybitWalletBalanceResponse, BybitHttpError> {
1925 self.inner.get_wallet_balance(params).await
1926 }
1927
1928 pub async fn get_account_info(&self) -> Result<BybitAccountInfoResponse, BybitHttpError> {
1940 self.inner.get_account_info().await
1941 }
1942
1943 pub async fn get_account_details(&self) -> Result<BybitAccountDetailsResponse, BybitHttpError> {
1955 self.inner.get_account_details().await
1956 }
1957
1958 pub async fn update_sub_api_key(
1970 &self,
1971 params: &BybitUpdateSubApiParams,
1972 ) -> Result<BybitUpdateSubApiResponse, BybitHttpError> {
1973 self.inner.update_sub_api_key(params).await
1974 }
1975
1976 pub async fn update_master_api_key(
1988 &self,
1989 params: &BybitUpdateMasterApiParams,
1990 ) -> Result<BybitUpdateMasterApiResponse, BybitHttpError> {
1991 self.inner.update_master_api_key(params).await
1992 }
1993
1994 pub async fn get_sub_members(&self) -> Result<BybitSubMembersResponse, BybitHttpError> {
2006 self.inner.get_sub_members().await
2007 }
2008
2009 pub async fn get_sub_members_paged(
2021 &self,
2022 params: &BybitSubMembersPageParams,
2023 ) -> Result<BybitSubMembersPagedResponse, BybitHttpError> {
2024 self.inner.get_sub_members_paged(params).await
2025 }
2026
2027 pub async fn get_escrow_sub_members(
2039 &self,
2040 params: &BybitSubMembersPageParams,
2041 ) -> Result<BybitEscrowSubMembersResponse, BybitHttpError> {
2042 self.inner.get_escrow_sub_members(params).await
2043 }
2044
2045 pub async fn get_sub_api_keys(
2057 &self,
2058 params: &BybitSubApiKeysParams,
2059 ) -> Result<BybitSubApiKeysResponse, BybitHttpError> {
2060 self.inner.get_sub_api_keys(params).await
2061 }
2062
2063 pub async fn get_positions(
2076 &self,
2077 params: &BybitPositionListParams,
2078 ) -> Result<BybitPositionListResponse, BybitHttpError> {
2079 self.inner.get_positions(params).await
2080 }
2081
2082 pub async fn get_fee_rate(
2095 &self,
2096 params: &BybitFeeRateParams,
2097 ) -> Result<BybitFeeRateResponse, BybitHttpError> {
2098 self.inner.get_fee_rate(params).await
2099 }
2100
2101 pub async fn set_margin_mode(
2114 &self,
2115 margin_mode: BybitMarginMode,
2116 ) -> Result<BybitSetMarginModeResponse, BybitHttpError> {
2117 self.inner.set_margin_mode(margin_mode).await
2118 }
2119
2120 pub async fn set_leverage(
2133 &self,
2134 product_type: BybitProductType,
2135 symbol: &str,
2136 buy_leverage: &str,
2137 sell_leverage: &str,
2138 ) -> Result<BybitSetLeverageResponse, BybitHttpError> {
2139 self.inner
2140 .set_leverage(product_type, symbol, buy_leverage, sell_leverage)
2141 .await
2142 }
2143
2144 pub async fn switch_mode(
2157 &self,
2158 product_type: BybitProductType,
2159 mode: BybitPositionMode,
2160 symbol: Option<String>,
2161 coin: Option<String>,
2162 ) -> Result<BybitSwitchModeResponse, BybitHttpError> {
2163 self.inner
2164 .switch_mode(product_type, mode, symbol, coin)
2165 .await
2166 }
2167
2168 pub async fn set_trading_stop(
2181 &self,
2182 params: &BybitSetTradingStopParams,
2183 ) -> Result<BybitSetTradingStopResponse, BybitHttpError> {
2184 self.inner.set_trading_stop(params).await
2185 }
2186
2187 pub async fn get_spot_borrow_amount(&self, coin: &str) -> anyhow::Result<Decimal> {
2202 let params = BybitWalletBalanceParams {
2203 account_type: BybitAccountType::Unified,
2204 coin: Some(coin.to_string()),
2205 };
2206
2207 let response = self.inner.get_wallet_balance(¶ms).await?;
2208
2209 let borrow_amount = response
2210 .result
2211 .list
2212 .first()
2213 .and_then(|wallet| wallet.coin.iter().find(|c| c.coin.as_str() == coin))
2214 .map_or(Decimal::ZERO, |balance| balance.spot_borrow);
2215
2216 Ok(borrow_amount)
2217 }
2218
2219 pub async fn borrow_spot(
2235 &self,
2236 coin: &str,
2237 amount: Quantity,
2238 ) -> anyhow::Result<BybitBorrowResponse> {
2239 let amount_str = amount.to_string();
2240 self.inner
2241 .borrow(coin, &amount_str)
2242 .await
2243 .map_err(|e| anyhow::anyhow!("Failed to borrow {amount} {coin}: {e}"))
2244 }
2245
2246 pub async fn repay_spot_borrow(
2263 &self,
2264 coin: &str,
2265 amount: Option<Quantity>,
2266 ) -> anyhow::Result<BybitNoConvertRepayResponse> {
2267 let amount_str = amount.as_ref().map(|q| q.to_string());
2268 self.inner
2269 .no_convert_repay(coin, amount_str.as_deref())
2270 .await
2271 .map_err(|e| anyhow::anyhow!("Failed to repay spot borrow for {coin}: {e}"))
2272 }
2273
2274 async fn generate_spot_position_reports_from_wallet(
2282 &self,
2283 account_id: AccountId,
2284 instrument_id: Option<InstrumentId>,
2285 ) -> anyhow::Result<Vec<PositionStatusReport>> {
2286 let params = BybitWalletBalanceParams {
2287 account_type: BybitAccountType::Unified,
2288 coin: None,
2289 };
2290
2291 let response = self.inner.get_wallet_balance(¶ms).await?;
2292 let ts_init = self.generate_ts_init();
2293
2294 let mut wallet_by_coin: HashMap<Ustr, Decimal> = HashMap::new();
2295
2296 for wallet in &response.result.list {
2297 for coin_balance in &wallet.coin {
2298 let balance = coin_balance.wallet_balance - coin_balance.spot_borrow;
2299 *wallet_by_coin
2300 .entry(coin_balance.coin)
2301 .or_insert(Decimal::ZERO) += balance;
2302 }
2303 }
2304
2305 let mut reports = Vec::new();
2306
2307 if let Some(instrument_id) = instrument_id {
2308 if let Some(instrument) = self
2309 .instruments_cache
2310 .get_cloned(&instrument_id.symbol.inner())
2311 {
2312 let base_currency = instrument
2313 .base_currency()
2314 .expect("SPOT instrument should have base currency");
2315 let coin = base_currency.code;
2316 let wallet_balance = wallet_by_coin.get(&coin).copied().unwrap_or(Decimal::ZERO);
2317
2318 let side = if wallet_balance > Decimal::ZERO {
2319 PositionSideSpecified::Long
2320 } else if wallet_balance < Decimal::ZERO {
2321 PositionSideSpecified::Short
2322 } else {
2323 PositionSideSpecified::Flat
2324 };
2325
2326 let abs_balance = wallet_balance.abs();
2327 let quantity = Quantity::from_decimal_dp(abs_balance, instrument.size_precision())?;
2328
2329 let report = PositionStatusReport::new(
2330 account_id,
2331 instrument_id,
2332 side,
2333 quantity,
2334 ts_init,
2335 ts_init,
2336 None,
2337 None,
2338 None,
2339 );
2340
2341 reports.push(report);
2342 }
2343 } else {
2344 let instruments_guard = self.instruments_cache.load();
2346 for (symbol, instrument) in instruments_guard.iter() {
2347 if !symbol.as_str().ends_with("-SPOT") {
2349 continue;
2350 }
2351
2352 let base_currency = match instrument.base_currency() {
2353 Some(currency) => currency,
2354 None => continue,
2355 };
2356
2357 let coin = base_currency.code;
2358 let wallet_balance = wallet_by_coin.get(&coin).copied().unwrap_or(Decimal::ZERO);
2359
2360 if wallet_balance.is_zero() {
2361 continue;
2362 }
2363
2364 let side = if wallet_balance > Decimal::ZERO {
2365 PositionSideSpecified::Long
2366 } else if wallet_balance < Decimal::ZERO {
2367 PositionSideSpecified::Short
2368 } else {
2369 PositionSideSpecified::Flat
2370 };
2371
2372 let abs_balance = wallet_balance.abs();
2373 let quantity = Quantity::from_decimal_dp(abs_balance, instrument.size_precision())?;
2374
2375 if quantity.is_zero() {
2376 continue;
2377 }
2378
2379 let report = PositionStatusReport::new(
2380 account_id,
2381 instrument.id(),
2382 side,
2383 quantity,
2384 ts_init,
2385 ts_init,
2386 None,
2387 None,
2388 None,
2389 );
2390
2391 reports.push(report);
2392 }
2393 }
2394
2395 Ok(reports)
2396 }
2397
2398 #[expect(clippy::too_many_arguments)]
2409 pub async fn submit_order(
2410 &self,
2411 account_id: AccountId,
2412 product_type: BybitProductType,
2413 instrument_id: InstrumentId,
2414 client_order_id: ClientOrderId,
2415 order_side: OrderSide,
2416 order_type: OrderType,
2417 quantity: Quantity,
2418 time_in_force: Option<TimeInForce>,
2419 price: Option<Price>,
2420 trigger_price: Option<Price>,
2421 post_only: Option<bool>,
2422 reduce_only: bool,
2423 is_quote_quantity: bool,
2424 is_leverage: bool,
2425 position_idx: Option<BybitPositionIdx>,
2426 bbo_side_type: Option<BybitBboSideType>,
2427 bbo_level: Option<String>,
2428 native_tp_sl: Option<&BybitNativeTpSlParams>,
2429 ) -> anyhow::Result<OrderStatusReport> {
2430 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
2431 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2432
2433 let bybit_side = match order_side {
2434 OrderSide::Buy => BybitOrderSide::Buy,
2435 OrderSide::Sell => BybitOrderSide::Sell,
2436 _ => anyhow::bail!("Invalid order side: {order_side:?}"),
2437 };
2438
2439 let (bybit_order_type, is_stop_order) = match order_type {
2441 OrderType::Market => (BybitOrderType::Market, false),
2442 OrderType::Limit => (BybitOrderType::Limit, false),
2443 OrderType::StopMarket | OrderType::MarketIfTouched => (BybitOrderType::Market, true),
2444 OrderType::StopLimit | OrderType::LimitIfTouched => (BybitOrderType::Limit, true),
2445 _ => anyhow::bail!("Unsupported order type: {order_type:?}"),
2446 };
2447
2448 let bybit_tif = map_time_in_force(bybit_order_type, time_in_force, post_only)
2449 .map_err(|tif| anyhow::anyhow!("Unsupported time in force: {tif:?}"))?;
2450 let market_unit = spot_market_unit(product_type, bybit_order_type, is_quote_quantity);
2451 let trigger_dir = trigger_direction(order_type, order_side, is_stop_order);
2452
2453 let mut order_entry = BybitBatchPlaceOrderEntryBuilder::default();
2454 order_entry.symbol(bybit_symbol.raw_symbol().to_string());
2455 order_entry.side(bybit_side);
2456 order_entry.order_type(bybit_order_type);
2457 order_entry.qty(quantity.to_string());
2458 order_entry.time_in_force(bybit_tif);
2459 order_entry.order_link_id(client_order_id.to_string());
2460 order_entry.market_unit(market_unit);
2461 order_entry.trigger_direction(trigger_dir);
2462
2463 if bbo_side_type.is_none()
2464 && let Some(price) = price
2465 {
2466 order_entry.price(Some(price.to_string()));
2467 }
2468
2469 if let Some(trigger_price) = trigger_price {
2470 order_entry.trigger_price(Some(trigger_price.to_string()));
2471 }
2472
2473 if reduce_only {
2474 order_entry.reduce_only(Some(true));
2475 }
2476
2477 order_entry.is_leverage(spot_leverage(product_type, is_leverage));
2478
2479 if let Some(idx) = position_idx {
2480 order_entry.position_idx(Some(idx));
2481 }
2482
2483 order_entry.bbo_side_type(bbo_side_type);
2484 order_entry.bbo_level(bbo_level);
2485
2486 if let Some(tp_sl) = native_tp_sl {
2487 if let Some(ref tp) = tp_sl.take_profit {
2488 order_entry.take_profit(Some(tp.clone()));
2489 }
2490
2491 if let Some(ref sl) = tp_sl.stop_loss {
2492 order_entry.stop_loss(Some(sl.clone()));
2493 }
2494
2495 if let Some(tp_trigger) = tp_sl.tp_trigger_by {
2496 order_entry.tp_trigger_by(Some(tp_trigger));
2497 }
2498
2499 if let Some(sl_trigger) = tp_sl.sl_trigger_by {
2500 order_entry.sl_trigger_by(Some(sl_trigger));
2501 }
2502
2503 if let Some(tp_ot) = tp_sl.tp_order_type {
2504 order_entry.tp_order_type(Some(tp_ot));
2505 }
2506
2507 if let Some(sl_ot) = tp_sl.sl_order_type {
2508 order_entry.sl_order_type(Some(sl_ot));
2509 }
2510
2511 if let Some(ref tp_lp) = tp_sl.tp_limit_price {
2512 order_entry.tp_limit_price(Some(tp_lp.clone()));
2513 }
2514
2515 if let Some(ref sl_lp) = tp_sl.sl_limit_price {
2516 order_entry.sl_limit_price(Some(sl_lp.clone()));
2517 }
2518
2519 let mode = tp_sl.tpsl_mode.or_else(|| {
2522 (tp_sl.take_profit.is_some() || tp_sl.stop_loss.is_some())
2523 .then_some(BybitTpSlMode::Full)
2524 });
2525
2526 if let Some(m) = mode {
2527 order_entry.tpsl_mode(Some(m));
2528 }
2529
2530 if let Some(close) = tp_sl.close_on_trigger {
2531 order_entry.close_on_trigger(Some(close));
2532 }
2533
2534 if let Some(ref iv) = tp_sl.order_iv {
2535 order_entry.order_iv(Some(iv.clone()));
2536 }
2537
2538 if let Some(mmp) = tp_sl.mmp {
2539 order_entry.mmp(Some(mmp));
2540 }
2541 }
2542
2543 let order_entry = order_entry.build().build_anyhow()?;
2544
2545 let mut params = BybitPlaceOrderParamsBuilder::default();
2546 params.category(product_type);
2547 params.order(order_entry);
2548
2549 let params = params.build().build_anyhow()?;
2550
2551 let body = serde_json::to_value(¶ms)?;
2552 let response = self.inner.place_order(&body).await?;
2553
2554 let order_id = response
2555 .result
2556 .order_id
2557 .ok_or(BybitSubmitOrderError::MissingOrderId)?;
2558
2559 let order = self
2560 .query_order_by_id(
2561 product_type,
2562 order_id.as_str(),
2563 BYBIT_ORDER_REALTIME,
2564 "after submission",
2565 )
2566 .await
2567 .map_err(|source| BybitSubmitOrderError::PostSubmitLookup { source })?;
2568
2569 if order.order_status == crate::common::enums::BybitOrderStatus::Rejected
2572 && (order.cum_exec_qty.as_str() == "0" || order.cum_exec_qty.is_empty())
2573 {
2574 return Err(BybitSubmitOrderError::Rejected {
2575 reason: order.reject_reason.to_string(),
2576 }
2577 .into());
2578 }
2579
2580 let ts_init = self.generate_ts_init();
2581
2582 parse_order_status_report(&order, &instrument, account_id, ts_init)
2583 }
2584
2585 pub async fn cancel_order(
2595 &self,
2596 account_id: AccountId,
2597 product_type: BybitProductType,
2598 instrument_id: InstrumentId,
2599 client_order_id: Option<ClientOrderId>,
2600 venue_order_id: Option<VenueOrderId>,
2601 ) -> anyhow::Result<OrderStatusReport> {
2602 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
2603 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2604
2605 let mut cancel_entry = BybitBatchCancelOrderEntryBuilder::default();
2606 cancel_entry.symbol(bybit_symbol.raw_symbol().to_string());
2607
2608 if let Some(venue_order_id) = venue_order_id {
2609 cancel_entry.order_id(venue_order_id.to_string());
2610 } else if let Some(client_order_id) = client_order_id {
2611 cancel_entry.order_link_id(client_order_id.to_string());
2612 } else {
2613 anyhow::bail!("Either client_order_id or venue_order_id must be provided");
2614 }
2615
2616 let cancel_entry = cancel_entry.build().build_anyhow()?;
2617
2618 let mut params = BybitCancelOrderParamsBuilder::default();
2619 params.category(product_type);
2620 params.order(cancel_entry);
2621
2622 let params = params.build().build_anyhow()?;
2623 let body = serde_json::to_vec(¶ms)?;
2624
2625 let response: BybitPlaceOrderResponse = self
2626 .inner
2627 .send_request::<_, ()>(Method::POST, "/v5/order/cancel", None, Some(body), true)
2628 .await?;
2629
2630 let order_id = response
2631 .result
2632 .order_id
2633 .ok_or(BybitCancelOrderError::MissingOrderId)?;
2634
2635 let order = self
2636 .query_order_by_id(
2637 product_type,
2638 order_id.as_str(),
2639 BYBIT_ORDER_HISTORY,
2640 "after cancellation",
2641 )
2642 .await
2643 .map_err(|source| BybitCancelOrderError::PostCancelLookup { source })?;
2644
2645 let ts_init = self.generate_ts_init();
2646
2647 parse_order_status_report(&order, &instrument, account_id, ts_init)
2648 }
2649
2650 pub async fn batch_cancel_orders(
2660 &self,
2661 account_id: AccountId,
2662 product_type: BybitProductType,
2663 instrument_ids: Vec<InstrumentId>,
2664 client_order_ids: Vec<Option<ClientOrderId>>,
2665 venue_order_ids: Vec<Option<VenueOrderId>>,
2666 ) -> anyhow::Result<Vec<OrderStatusReport>> {
2667 if instrument_ids.len() != client_order_ids.len()
2668 || instrument_ids.len() != venue_order_ids.len()
2669 {
2670 anyhow::bail!(
2671 "instrument_ids, client_order_ids, and venue_order_ids must have the same length"
2672 );
2673 }
2674
2675 if instrument_ids.is_empty() {
2676 return Ok(Vec::new());
2677 }
2678
2679 if instrument_ids.len() > 20 {
2680 anyhow::bail!("Batch cancel limit is 20 orders per request");
2681 }
2682
2683 let mut cancel_entries = Vec::new();
2684
2685 for ((instrument_id, client_order_id), venue_order_id) in instrument_ids
2686 .iter()
2687 .zip(client_order_ids.iter())
2688 .zip(venue_order_ids.iter())
2689 {
2690 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2691 let mut cancel_entry = BybitBatchCancelOrderEntryBuilder::default();
2692 cancel_entry.symbol(bybit_symbol.raw_symbol().to_string());
2693
2694 if let Some(venue_order_id) = venue_order_id {
2695 cancel_entry.order_id(venue_order_id.to_string());
2696 } else if let Some(client_order_id) = client_order_id {
2697 cancel_entry.order_link_id(client_order_id.to_string());
2698 } else {
2699 anyhow::bail!(
2700 "Either client_order_id or venue_order_id must be provided for each order"
2701 );
2702 }
2703
2704 cancel_entries.push(cancel_entry.build().build_anyhow()?);
2705 }
2706
2707 let mut params = BybitBatchCancelOrderParamsBuilder::default();
2708 params.category(product_type);
2709 params.request(cancel_entries);
2710
2711 let params = params.build().build_anyhow()?;
2712 let body = serde_json::to_vec(¶ms)?;
2713
2714 let _response: BybitPlaceOrderResponse = self
2715 .inner
2716 .send_request::<_, ()>(
2717 Method::POST,
2718 "/v5/order/cancel-batch",
2719 None,
2720 Some(body),
2721 true,
2722 )
2723 .await?;
2724
2725 let mut reports = Vec::new();
2727
2728 for (instrument_id, (client_order_id, venue_order_id)) in instrument_ids
2729 .iter()
2730 .zip(client_order_ids.iter().zip(venue_order_ids.iter()))
2731 {
2732 let Ok(instrument) = self.instrument_from_cache(&instrument_id.symbol) else {
2733 log::debug!(
2734 "Skipping cancelled order report for instrument not in cache: symbol={}",
2735 instrument_id.symbol
2736 );
2737 continue;
2738 };
2739
2740 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2741
2742 let mut query_params = BybitOpenOrdersParamsBuilder::default();
2743 query_params.category(product_type);
2744 query_params.symbol(bybit_symbol.raw_symbol().to_string());
2745
2746 if let Some(venue_order_id) = venue_order_id {
2747 query_params.order_id(venue_order_id.to_string());
2748 } else if let Some(client_order_id) = client_order_id {
2749 query_params.order_link_id(client_order_id.to_string());
2750 }
2751
2752 let query_params = query_params.build().build_anyhow()?;
2753 let order_response: BybitOrderHistoryResponse = self
2754 .inner
2755 .send_request(
2756 Method::GET,
2757 BYBIT_ORDER_HISTORY,
2758 Some(&query_params),
2759 None,
2760 true,
2761 )
2762 .await?;
2763
2764 if let Some(order) = order_response.result.list.into_iter().next() {
2765 let ts_init = self.generate_ts_init();
2766 let report = parse_order_status_report(&order, &instrument, account_id, ts_init)?;
2767 reports.push(report);
2768 }
2769 }
2770
2771 Ok(reports)
2772 }
2773
2774 pub async fn cancel_all_orders(
2783 &self,
2784 account_id: AccountId,
2785 product_type: BybitProductType,
2786 instrument_id: InstrumentId,
2787 ) -> anyhow::Result<Vec<OrderStatusReport>> {
2788 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
2789 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2790
2791 let mut params = BybitCancelAllOrdersParamsBuilder::default();
2792 params.category(product_type);
2793 params.symbol(bybit_symbol.raw_symbol().to_string());
2794
2795 let params = params.build().build_anyhow()?;
2796 let body = serde_json::to_vec(¶ms)?;
2797
2798 let _response: crate::common::models::BybitListResponse<serde_json::Value> = self
2799 .inner
2800 .send_request::<_, ()>(Method::POST, "/v5/order/cancel-all", None, Some(body), true)
2801 .await?;
2802
2803 let mut query_params = BybitOrderHistoryParamsBuilder::default();
2805 query_params.category(product_type);
2806 query_params.symbol(bybit_symbol.raw_symbol().to_string());
2807 query_params.limit(50u32);
2808
2809 let query_params = query_params.build().build_anyhow()?;
2810 let order_response: BybitOrderHistoryResponse = self
2811 .inner
2812 .send_request(
2813 Method::GET,
2814 BYBIT_ORDER_HISTORY,
2815 Some(&query_params),
2816 None,
2817 true,
2818 )
2819 .await?;
2820
2821 let ts_init = self.generate_ts_init();
2822
2823 let mut reports = Vec::new();
2824
2825 for order in order_response.result.list {
2826 if let Ok(report) = parse_order_status_report(&order, &instrument, account_id, ts_init)
2827 {
2828 reports.push(report);
2829 }
2830 }
2831
2832 Ok(reports)
2833 }
2834
2835 #[expect(clippy::too_many_arguments)]
2846 pub async fn modify_order(
2847 &self,
2848 account_id: AccountId,
2849 product_type: BybitProductType,
2850 instrument_id: InstrumentId,
2851 client_order_id: Option<ClientOrderId>,
2852 venue_order_id: Option<VenueOrderId>,
2853 quantity: Option<Quantity>,
2854 price: Option<Price>,
2855 ) -> anyhow::Result<OrderStatusReport> {
2856 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
2857 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2858
2859 let mut amend_entry = BybitBatchAmendOrderEntryBuilder::default();
2860 amend_entry.symbol(bybit_symbol.raw_symbol().to_string());
2861
2862 if let Some(venue_order_id) = venue_order_id {
2863 amend_entry.order_id(venue_order_id.to_string());
2864 } else if let Some(client_order_id) = client_order_id {
2865 amend_entry.order_link_id(client_order_id.to_string());
2866 } else {
2867 anyhow::bail!("Either client_order_id or venue_order_id must be provided");
2868 }
2869
2870 if let Some(quantity) = quantity {
2871 amend_entry.qty(Some(quantity.to_string()));
2872 }
2873
2874 if let Some(price) = price {
2875 amend_entry.price(Some(price.to_string()));
2876 }
2877
2878 let amend_entry = amend_entry.build().build_anyhow()?;
2879
2880 let mut params = BybitAmendOrderParamsBuilder::default();
2881 params.category(product_type);
2882 params.order(amend_entry);
2883
2884 let params = params.build().build_anyhow()?;
2885 let body = serde_json::to_vec(¶ms)?;
2886
2887 let response: BybitPlaceOrderResponse = self
2888 .inner
2889 .send_request::<_, ()>(Method::POST, "/v5/order/amend", None, Some(body), true)
2890 .await?;
2891
2892 let order_id = response
2893 .result
2894 .order_id
2895 .ok_or(BybitModifyOrderError::MissingOrderId)?;
2896
2897 let order = self
2898 .query_order_by_id(
2899 product_type,
2900 order_id.as_str(),
2901 BYBIT_ORDER_REALTIME,
2902 "after amendment",
2903 )
2904 .await
2905 .map_err(|source| BybitModifyOrderError::PostModifyLookup { source })?;
2906
2907 let ts_init = self.generate_ts_init();
2908
2909 parse_order_status_report(&order, &instrument, account_id, ts_init)
2910 }
2911
2912 pub async fn query_order(
2921 &self,
2922 account_id: AccountId,
2923 product_type: BybitProductType,
2924 instrument_id: InstrumentId,
2925 client_order_id: Option<ClientOrderId>,
2926 venue_order_id: Option<VenueOrderId>,
2927 ) -> anyhow::Result<Option<OrderStatusReport>> {
2928 log::debug!(
2929 "query_order: instrument_id={instrument_id}, client_order_id={client_order_id:?}, venue_order_id={venue_order_id:?}"
2930 );
2931
2932 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2933
2934 let mut params = BybitOpenOrdersParamsBuilder::default();
2935 params.category(product_type);
2936 params.symbol(bybit_symbol.raw_symbol().to_string());
2938
2939 if let Some(venue_order_id) = venue_order_id {
2940 params.order_id(venue_order_id.to_string());
2941 } else if let Some(client_order_id) = client_order_id {
2942 params.order_link_id(client_order_id.to_string());
2943 } else {
2944 anyhow::bail!("Either client_order_id or venue_order_id must be provided");
2945 }
2946
2947 let params = params.build().build_anyhow()?;
2948 let mut response: BybitOpenOrdersResponse = self
2949 .inner
2950 .send_request(Method::GET, BYBIT_ORDER_REALTIME, Some(¶ms), None, true)
2951 .await?;
2952
2953 if response.result.list.is_empty() && product_type != BybitProductType::Option {
2955 log::debug!("Order not found in open orders, trying with StopOrder filter");
2956
2957 let mut stop_params = BybitOpenOrdersParamsBuilder::default();
2958 stop_params.category(product_type);
2959 stop_params.symbol(bybit_symbol.raw_symbol().to_string());
2960 stop_params.order_filter(BybitOrderFilter::StopOrder);
2961
2962 if let Some(venue_order_id) = venue_order_id {
2963 stop_params.order_id(venue_order_id.to_string());
2964 } else if let Some(client_order_id) = client_order_id {
2965 stop_params.order_link_id(client_order_id.to_string());
2966 }
2967
2968 let stop_params = stop_params.build().build_anyhow()?;
2969 response = self
2970 .inner
2971 .send_request(
2972 Method::GET,
2973 BYBIT_ORDER_REALTIME,
2974 Some(&stop_params),
2975 None,
2976 true,
2977 )
2978 .await?;
2979 }
2980
2981 if response.result.list.is_empty() {
2983 log::debug!("Order not found in open orders, checking order history");
2984
2985 let mut history_params = BybitOrderHistoryParamsBuilder::default();
2986 history_params.category(product_type);
2987 history_params.symbol(bybit_symbol.raw_symbol().to_string());
2988
2989 if let Some(venue_order_id) = venue_order_id {
2990 history_params.order_id(venue_order_id.to_string());
2991 } else if let Some(client_order_id) = client_order_id {
2992 history_params.order_link_id(client_order_id.to_string());
2993 }
2994
2995 let history_params = history_params.build().build_anyhow()?;
2996
2997 let mut history_response: BybitOrderHistoryResponse = self
2998 .inner
2999 .send_request(
3000 Method::GET,
3001 BYBIT_ORDER_HISTORY,
3002 Some(&history_params),
3003 None,
3004 true,
3005 )
3006 .await?;
3007
3008 if history_response.result.list.is_empty() && product_type == BybitProductType::Option {
3009 log::debug!("Option order not found in order history");
3010 return Ok(None);
3011 }
3012
3013 if history_response.result.list.is_empty() && product_type != BybitProductType::Option {
3015 log::debug!("Order not found in order history, trying with StopOrder filter");
3016
3017 let mut stop_history_params = BybitOrderHistoryParamsBuilder::default();
3018 stop_history_params.category(product_type);
3019 stop_history_params.symbol(bybit_symbol.raw_symbol().to_string());
3020 stop_history_params.order_filter(BybitOrderFilter::StopOrder);
3021
3022 if let Some(venue_order_id) = venue_order_id {
3023 stop_history_params.order_id(venue_order_id.to_string());
3024 } else if let Some(client_order_id) = client_order_id {
3025 stop_history_params.order_link_id(client_order_id.to_string());
3026 }
3027
3028 let stop_history_params = stop_history_params
3029 .build()
3030 .map_err(|e| anyhow::anyhow!(e))?;
3031
3032 history_response = self
3033 .inner
3034 .send_request(
3035 Method::GET,
3036 BYBIT_ORDER_HISTORY,
3037 Some(&stop_history_params),
3038 None,
3039 true,
3040 )
3041 .await?;
3042
3043 if history_response.result.list.is_empty() {
3044 log::debug!("Order not found in order history with StopOrder filter either");
3045 return Ok(None);
3046 }
3047 }
3048
3049 response.result.list = history_response.result.list;
3051 }
3052
3053 let order = &response.result.list[0];
3054 let ts_init = self.generate_ts_init();
3055
3056 log::debug!(
3057 "Query order response: symbol={}, order_id={}, order_link_id={}",
3058 order.symbol.as_str(),
3059 order.order_id.as_str(),
3060 order.order_link_id.as_str()
3061 );
3062
3063 let instrument = self
3064 .instrument_from_cache(&instrument_id.symbol)
3065 .map_err(|e| {
3066 log::error!(
3067 "Instrument cache miss for symbol '{}': {}",
3068 instrument_id.symbol.as_str(),
3069 e
3070 );
3071 anyhow::anyhow!(
3072 "Failed to query order {}: {}",
3073 client_order_id
3074 .as_ref()
3075 .map(|id| id.to_string())
3076 .or_else(|| venue_order_id.as_ref().map(|id| id.to_string()))
3077 .unwrap_or_else(|| "unknown".to_string()),
3078 e
3079 )
3080 })?;
3081
3082 log::debug!("Retrieved instrument from cache: id={}", instrument.id());
3083
3084 let report =
3085 parse_order_status_report(order, &instrument, account_id, ts_init).map_err(|e| {
3086 log::error!(
3087 "Failed to parse order status report for {}: {}",
3088 order.order_link_id.as_str(),
3089 e
3090 );
3091 e
3092 })?;
3093
3094 log::debug!(
3095 "Successfully created OrderStatusReport for {}",
3096 order.order_link_id.as_str()
3097 );
3098
3099 Ok(Some(report))
3100 }
3101
3102 async fn fetch_fee_map(
3103 &self,
3104 product_type: BybitProductType,
3105 base_coin: Option<Ustr>,
3106 ) -> anyhow::Result<AHashMap<Ustr, BybitFeeRate>> {
3107 let mut fee_params = BybitFeeRateParamsBuilder::default();
3108 fee_params.category(product_type);
3109 if let Some(bc) = base_coin {
3110 fee_params.base_coin(bc.to_string());
3111 }
3112 let Ok(params) = fee_params.build() else {
3113 return Ok(AHashMap::new());
3114 };
3115
3116 match self.inner.get_fee_rate(¶ms).await {
3117 Ok(response) => Ok(response
3118 .result
3119 .list
3120 .into_iter()
3121 .map(|f| (f.symbol, f))
3122 .collect()),
3123 Err(BybitHttpError::MissingCredentials) => {
3124 log::warn!("Missing credentials for fee rates, using defaults");
3125 Ok(AHashMap::new())
3126 }
3127 Err(BybitHttpError::BybitError {
3128 error_code,
3129 ref message,
3130 }) => {
3131 log::warn!(
3132 "{}",
3133 self.fee_rate_rejection_warning(product_type, error_code, message)
3134 );
3135 Ok(AHashMap::new())
3136 }
3137 Err(e) => Err(e.into()),
3138 }
3139 }
3140
3141 async fn fetch_option_fee_map(
3142 &self,
3143 base_coin: Option<Ustr>,
3144 ) -> anyhow::Result<AHashMap<Ustr, BybitFeeRate>> {
3145 let mut fee_params = BybitFeeRateParamsBuilder::default();
3146 fee_params.category(BybitProductType::Option);
3147 if let Some(bc) = base_coin {
3148 fee_params.base_coin(bc.to_string());
3149 }
3150 let Ok(params) = fee_params.build() else {
3151 return Ok(AHashMap::new());
3152 };
3153
3154 match self.inner.get_fee_rate(¶ms).await {
3155 Ok(response) => Ok(response
3156 .result
3157 .list
3158 .into_iter()
3159 .filter_map(|f| f.base_coin.map(|bc| (bc, f)))
3160 .collect()),
3161 Err(BybitHttpError::MissingCredentials) => {
3162 log::warn!("Missing credentials for option fee rates, using defaults");
3163 Ok(AHashMap::new())
3164 }
3165 Err(BybitHttpError::BybitError {
3166 error_code,
3167 ref message,
3168 }) => {
3169 let error_detail = Self::format_bybit_error_detail(error_code, message);
3170 log::warn!(
3171 "Option fee rate request rejected via /v5/account/fee-rate ({error_detail}), using defaults"
3172 );
3173 Ok(AHashMap::new())
3174 }
3175 Err(e) => {
3176 log::warn!("Option fee rate request failed ({e}), using defaults");
3177 Ok(AHashMap::new())
3178 }
3179 }
3180 }
3181
3182 fn fee_rate_rejection_warning(
3183 &self,
3184 product_type: BybitProductType,
3185 error_code: i32,
3186 message: &str,
3187 ) -> String {
3188 let product_type = product_type.as_ref().to_ascii_lowercase();
3189 let error_detail = Self::format_bybit_error_detail(error_code, message);
3190
3191 if self
3192 .base_url()
3193 .starts_with(bybit_http_base_url(BybitEnvironment::Demo))
3194 && matches!(product_type.as_str(), "linear" | "inverse")
3195 && error_code == 10001
3196 {
3197 format!(
3198 "Bybit demo rejected the {product_type} fee rate request via \
3199 /v5/account/fee-rate ({error_detail}); demo derivatives fee rates appear \
3200 unsupported, using defaults"
3201 )
3202 } else {
3203 format!(
3204 "Fee rate request rejected for {product_type} instruments via \
3205 /v5/account/fee-rate ({error_detail}), using defaults"
3206 )
3207 }
3208 }
3209
3210 fn format_bybit_error_detail(error_code: i32, message: &str) -> String {
3211 let message = message.trim();
3212 if message.is_empty() {
3213 format!("error {error_code}, no message")
3214 } else {
3215 format!("error {error_code}: {message}")
3216 }
3217 }
3218
3219 async fn paginate_instruments<D, F>(
3220 &self,
3221 product_type: BybitProductType,
3222 symbol: &Option<String>,
3223 base_coin: Option<Ustr>,
3224 mut parse: F,
3225 ) -> anyhow::Result<Vec<InstrumentAny>>
3226 where
3227 D: DeserializeOwned,
3228 BybitCursorListResponse<D>: BybitResponseCheck,
3229 F: FnMut(&D) -> Option<InstrumentAny>,
3230 {
3231 let mut instruments = Vec::new();
3232 let mut cursor: Option<String> = None;
3233 let mut prev_cursor: Option<String> = None;
3234
3235 loop {
3236 let params = BybitInstrumentsInfoParams {
3237 category: product_type,
3238 symbol: symbol.clone(),
3239 status: None,
3240 base_coin: base_coin.map(|u| u.to_string()),
3241 limit: Some(1000),
3242 cursor: cursor.clone(),
3243 };
3244
3245 let response: BybitCursorListResponse<D> = self.inner.get_instruments(¶ms).await?;
3246
3247 for definition in &response.result.list {
3248 if let Some(instrument) = parse(definition) {
3249 instruments.push(instrument);
3250 }
3251 }
3252
3253 cursor = response.result.next_page_cursor;
3254 if cursor.as_ref().is_none_or(|c| c.is_empty()) || cursor == prev_cursor {
3255 break;
3256 }
3257 prev_cursor = cursor.clone();
3258 }
3259
3260 Ok(instruments)
3261 }
3262
3263 pub async fn request_instrument_statuses(
3273 &self,
3274 product_type: BybitProductType,
3275 ) -> anyhow::Result<AHashMap<InstrumentId, MarketStatusAction>> {
3276 let mut statuses = AHashMap::new();
3277 let mut cursor: Option<String> = None;
3278
3279 loop {
3280 let params = BybitInstrumentsInfoParams {
3281 category: product_type,
3282 symbol: None,
3283 status: None,
3284 base_coin: None,
3285 limit: Some(1000),
3286 cursor: cursor.clone(),
3287 };
3288
3289 match product_type {
3290 BybitProductType::Spot => {
3291 let response: BybitCursorListResponse<BybitInstrumentSpot> =
3292 self.inner.get_instruments(¶ms).await?;
3293
3294 for def in &response.result.list {
3295 let symbol = make_bybit_symbol(def.symbol, product_type);
3296 let id = InstrumentId::new(Symbol::from(symbol), *BYBIT_VENUE);
3297 statuses.insert(id, MarketStatusAction::from(def.status));
3298 }
3299 cursor = response.result.next_page_cursor;
3300 }
3301 BybitProductType::Linear => {
3302 let response: BybitCursorListResponse<BybitInstrumentLinear> =
3303 self.inner.get_instruments(¶ms).await?;
3304
3305 for def in &response.result.list {
3306 let symbol = make_bybit_symbol(def.symbol, product_type);
3307 let id = InstrumentId::new(Symbol::from(symbol), *BYBIT_VENUE);
3308 let status = MarketStatusAction::from(def.status);
3309 if status == MarketStatusAction::Trading
3310 && def.contract_type == BybitContractType::LinearPerpetual
3311 && def.delivery_time != "0"
3312 {
3313 statuses.insert(id, MarketStatusAction::PreClose);
3314 } else {
3315 statuses.insert(id, status);
3316 }
3317 }
3318 cursor = response.result.next_page_cursor;
3319 }
3320 BybitProductType::Inverse => {
3321 let response: BybitCursorListResponse<BybitInstrumentInverse> =
3322 self.inner.get_instruments(¶ms).await?;
3323
3324 for def in &response.result.list {
3325 let symbol = make_bybit_symbol(def.symbol, product_type);
3326 let id = InstrumentId::new(Symbol::from(symbol), *BYBIT_VENUE);
3327 let status = MarketStatusAction::from(def.status);
3328 if status == MarketStatusAction::Trading
3329 && def.contract_type == BybitContractType::InversePerpetual
3330 && def.delivery_time != "0"
3331 {
3332 statuses.insert(id, MarketStatusAction::PreClose);
3333 } else {
3334 statuses.insert(id, status);
3335 }
3336 }
3337 cursor = response.result.next_page_cursor;
3338 }
3339 BybitProductType::Option => {
3340 let response: BybitCursorListResponse<BybitInstrumentOption> =
3341 self.inner.get_instruments(¶ms).await?;
3342
3343 for def in &response.result.list {
3344 let symbol = make_bybit_symbol(def.symbol, product_type);
3345 let id = InstrumentId::new(Symbol::from(symbol), *BYBIT_VENUE);
3346 statuses.insert(id, MarketStatusAction::from(def.status));
3347 }
3348 cursor = response.result.next_page_cursor;
3349 }
3350 }
3351
3352 if cursor.as_ref().is_none_or(|c| c.is_empty()) {
3353 break;
3354 }
3355 }
3356
3357 Ok(statuses)
3358 }
3359
3360 pub async fn request_instruments(
3370 &self,
3371 product_type: BybitProductType,
3372 symbol: Option<String>,
3373 base_coin: Option<Ustr>,
3374 ) -> anyhow::Result<Vec<InstrumentAny>> {
3375 let ts_init = self.generate_ts_init();
3376
3377 let default_fee_rate = |symbol: Ustr| BybitFeeRate {
3378 symbol,
3379 taker_fee_rate: "0.001".to_string(),
3380 maker_fee_rate: "0.001".to_string(),
3381 base_coin: None,
3382 };
3383
3384 let instruments = match product_type {
3385 BybitProductType::Spot => {
3386 let fee_map = self.fetch_fee_map(product_type, base_coin).await?;
3387 self.paginate_instruments::<BybitInstrumentSpot, _>(
3388 product_type,
3389 &symbol,
3390 base_coin,
3391 |def| {
3392 let fee = fee_map
3393 .get(&def.symbol)
3394 .cloned()
3395 .unwrap_or_else(|| default_fee_rate(def.symbol));
3396 parse_spot_instrument(def, &fee, ts_init, ts_init).ok()
3397 },
3398 )
3399 .await?
3400 }
3401 BybitProductType::Linear => {
3402 let fee_map = self.fetch_fee_map(product_type, base_coin).await?;
3403 self.paginate_instruments::<BybitInstrumentLinear, _>(
3404 product_type,
3405 &symbol,
3406 base_coin,
3407 |def| {
3408 let fee = fee_map
3409 .get(&def.symbol)
3410 .cloned()
3411 .unwrap_or_else(|| default_fee_rate(def.symbol));
3412 parse_linear_instrument(def, &fee, ts_init, ts_init).ok()
3413 },
3414 )
3415 .await?
3416 }
3417 BybitProductType::Inverse => {
3418 let fee_map = self.fetch_fee_map(product_type, base_coin).await?;
3419 self.paginate_instruments::<BybitInstrumentInverse, _>(
3420 product_type,
3421 &symbol,
3422 base_coin,
3423 |def| {
3424 let fee = fee_map
3425 .get(&def.symbol)
3426 .cloned()
3427 .unwrap_or_else(|| default_fee_rate(def.symbol));
3428 parse_inverse_instrument(def, &fee, ts_init, ts_init).ok()
3429 },
3430 )
3431 .await?
3432 }
3433 BybitProductType::Option => {
3434 let fee_map = self.fetch_option_fee_map(base_coin).await?;
3435 self.paginate_instruments::<BybitInstrumentOption, _>(
3436 product_type,
3437 &symbol,
3438 base_coin,
3439 |def| {
3440 let fee = fee_map.get(&def.base_coin);
3441 parse_option_instrument(def, fee, ts_init, ts_init).ok()
3442 },
3443 )
3444 .await?
3445 }
3446 };
3447
3448 self.cache_instruments(&instruments);
3449
3450 Ok(instruments)
3451 }
3452
3453 pub async fn request_instruments_with_statuses(
3463 &self,
3464 product_type: BybitProductType,
3465 ) -> anyhow::Result<(
3466 Vec<InstrumentAny>,
3467 AHashMap<InstrumentId, MarketStatusAction>,
3468 )> {
3469 let ts_init = self.generate_ts_init();
3470 let mut statuses = AHashMap::new();
3471
3472 let default_fee_rate = |symbol: Ustr| BybitFeeRate {
3473 symbol,
3474 taker_fee_rate: "0.001".to_string(),
3475 maker_fee_rate: "0.001".to_string(),
3476 base_coin: None,
3477 };
3478
3479 let perp_status = |status: MarketStatusAction, is_scheduled_perp: bool| {
3481 if status == MarketStatusAction::Trading && is_scheduled_perp {
3482 MarketStatusAction::PreClose
3483 } else {
3484 status
3485 }
3486 };
3487
3488 let instruments = match product_type {
3489 BybitProductType::Spot => {
3490 let fee_map = self.fetch_fee_map(product_type, None).await?;
3491 self.paginate_instruments::<BybitInstrumentSpot, _>(
3492 product_type,
3493 &None::<String>,
3494 None,
3495 |def| {
3496 let id = InstrumentId::new(
3497 Symbol::from(make_bybit_symbol(def.symbol, product_type)),
3498 *BYBIT_VENUE,
3499 );
3500 statuses.insert(id, MarketStatusAction::from(def.status));
3501 let fee = fee_map
3502 .get(&def.symbol)
3503 .cloned()
3504 .unwrap_or_else(|| default_fee_rate(def.symbol));
3505 parse_spot_instrument(def, &fee, ts_init, ts_init).ok()
3506 },
3507 )
3508 .await?
3509 }
3510 BybitProductType::Linear => {
3511 let fee_map = self.fetch_fee_map(product_type, None).await?;
3512 self.paginate_instruments::<BybitInstrumentLinear, _>(
3513 product_type,
3514 &None::<String>,
3515 None,
3516 |def| {
3517 let id = InstrumentId::new(
3518 Symbol::from(make_bybit_symbol(def.symbol, product_type)),
3519 *BYBIT_VENUE,
3520 );
3521 let scheduled = def.contract_type == BybitContractType::LinearPerpetual
3522 && def.delivery_time != "0";
3523 statuses.insert(id, perp_status(def.status.into(), scheduled));
3524 let fee = fee_map
3525 .get(&def.symbol)
3526 .cloned()
3527 .unwrap_or_else(|| default_fee_rate(def.symbol));
3528 parse_linear_instrument(def, &fee, ts_init, ts_init).ok()
3529 },
3530 )
3531 .await?
3532 }
3533 BybitProductType::Inverse => {
3534 let fee_map = self.fetch_fee_map(product_type, None).await?;
3535 self.paginate_instruments::<BybitInstrumentInverse, _>(
3536 product_type,
3537 &None::<String>,
3538 None,
3539 |def| {
3540 let id = InstrumentId::new(
3541 Symbol::from(make_bybit_symbol(def.symbol, product_type)),
3542 *BYBIT_VENUE,
3543 );
3544 let scheduled = def.contract_type == BybitContractType::InversePerpetual
3545 && def.delivery_time != "0";
3546 statuses.insert(id, perp_status(def.status.into(), scheduled));
3547 let fee = fee_map
3548 .get(&def.symbol)
3549 .cloned()
3550 .unwrap_or_else(|| default_fee_rate(def.symbol));
3551 parse_inverse_instrument(def, &fee, ts_init, ts_init).ok()
3552 },
3553 )
3554 .await?
3555 }
3556 BybitProductType::Option => {
3557 let fee_map = self.fetch_option_fee_map(None).await?;
3558 self.paginate_instruments::<BybitInstrumentOption, _>(
3559 product_type,
3560 &None::<String>,
3561 None,
3562 |def| {
3563 let id = InstrumentId::new(
3564 Symbol::from(make_bybit_symbol(def.symbol, product_type)),
3565 *BYBIT_VENUE,
3566 );
3567 statuses.insert(id, MarketStatusAction::from(def.status));
3568 let fee = fee_map.get(&def.base_coin);
3569 parse_option_instrument(def, fee, ts_init, ts_init).ok()
3570 },
3571 )
3572 .await?
3573 }
3574 };
3575
3576 self.cache_instruments(&instruments);
3577
3578 Ok((instruments, statuses))
3579 }
3580
3581 pub async fn request_tickers(
3594 &self,
3595 params: &BybitTickersParams,
3596 ) -> anyhow::Result<Vec<BybitTickerData>> {
3597 use super::models::{
3598 BybitTickersLinearResponse, BybitTickersOptionResponse, BybitTickersSpotResponse,
3599 };
3600
3601 match params.category {
3602 BybitProductType::Spot => {
3603 let response: BybitTickersSpotResponse = self.inner.get_tickers(params).await?;
3604 Ok(response.result.list.into_iter().map(Into::into).collect())
3605 }
3606 BybitProductType::Linear | BybitProductType::Inverse => {
3607 let response: BybitTickersLinearResponse = self.inner.get_tickers(params).await?;
3608 Ok(response.result.list.into_iter().map(Into::into).collect())
3609 }
3610 BybitProductType::Option => {
3611 let response: BybitTickersOptionResponse = self.inner.get_tickers(params).await?;
3612 Ok(response.result.list.into_iter().map(Into::into).collect())
3613 }
3614 }
3615 }
3616
3617 pub async fn request_option_tickers_raw(
3626 &self,
3627 base_coin: &str,
3628 ) -> anyhow::Result<Vec<BybitTickerOption>> {
3629 let params = BybitTickersParams {
3630 category: BybitProductType::Option,
3631 symbol: None,
3632 base_coin: Some(base_coin.to_string()),
3633 exp_date: None,
3634 };
3635 let response: BybitTickersOptionResponse = self.inner.get_tickers(¶ms).await?;
3636 Ok(response.result.list)
3637 }
3638
3639 pub async fn request_option_tickers_raw_with_params(
3648 &self,
3649 params: &BybitTickersParams,
3650 ) -> anyhow::Result<Vec<BybitTickerOption>> {
3651 let response: BybitTickersOptionResponse = self.inner.get_tickers(params).await?;
3652 Ok(response.result.list)
3653 }
3654
3655 pub async fn request_trades(
3675 &self,
3676 product_type: BybitProductType,
3677 instrument_id: InstrumentId,
3678 limit: Option<u32>,
3679 ) -> anyhow::Result<Vec<TradeTick>> {
3680 let instrument = self.instrument_from_cache_by_id(instrument_id)?;
3681 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
3682
3683 let mut params_builder = BybitTradesParamsBuilder::default();
3684 params_builder.category(product_type);
3685 params_builder.symbol(bybit_symbol.raw_symbol().to_string());
3686
3687 if let Some(limit_val) = limit {
3688 params_builder.limit(limit_val);
3689 }
3690
3691 let params = params_builder.build().build_anyhow()?;
3692 let response = self.inner.get_recent_trades(¶ms).await?;
3693
3694 let mut trades = Vec::new();
3695
3696 for trade in response.result.list {
3697 if let Ok(trade_tick) = parse_trade_tick(&trade, &instrument, None) {
3698 trades.push(trade_tick);
3699 }
3700 }
3701
3702 Ok(trades)
3703 }
3704
3705 pub async fn request_funding_rates(
3718 &self,
3719 product_type: BybitProductType,
3720 instrument_id: InstrumentId,
3721 start: Option<DateTime<Utc>>,
3722 end: Option<DateTime<Utc>>,
3723 limit: Option<u32>,
3724 ) -> anyhow::Result<Vec<FundingRateUpdate>> {
3725 let instrument = self.instrument_from_cache_by_id(instrument_id)?;
3726 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
3727
3728 let start_ms = start.map(|dt| dt.timestamp_millis());
3729 let mut seen_timestamps: AHashSet<i64> = AHashSet::new();
3730
3731 let mut raw_funding_rates = Vec::new();
3732
3733 let mut current_end_ms = match (start, end) {
3735 (Some(_), None) => Some(Utc::now().timestamp_millis()),
3736 _ => end.map(|dt| dt.timestamp_millis()),
3737 };
3738
3739 loop {
3740 let mut params_builder = BybitFundingParamsBuilder::default();
3741 params_builder.category(product_type);
3742 params_builder.symbol(bybit_symbol.raw_symbol().to_string());
3743 params_builder.limit(limit.unwrap_or(200).clamp(0, 200)); if let Some(start_val) = start_ms {
3746 params_builder.start_time(start_val);
3747 }
3748
3749 if let Some(end_val) = current_end_ms {
3750 params_builder.end_time(end_val);
3751 }
3752
3753 let params = params_builder.build().build_anyhow()?;
3754 let response = self.inner.get_funding_history(¶ms).await?;
3755
3756 let funding_rates = response.result.list;
3757
3758 let mut new_funding_rates_with_ts: Vec<(i64, _)> = funding_rates
3759 .into_iter()
3760 .filter_map(|f| {
3761 let Ok(ts) = f.funding_rate_timestamp.parse::<i64>() else {
3762 return None;
3763 };
3764
3765 seen_timestamps.insert(ts).then_some((ts, f))
3766 })
3767 .collect();
3768
3769 new_funding_rates_with_ts.sort_by_key(|(ts, _)| Reverse(*ts));
3770
3771 let earliest_funding_time = match new_funding_rates_with_ts.last() {
3772 Some((last_ts, _)) => *last_ts,
3773 None => break,
3774 };
3775
3776 let new_funding_rates = new_funding_rates_with_ts.into_iter().map(|(_, f)| f);
3777 raw_funding_rates.extend(new_funding_rates);
3778
3779 if let Some(limit_val) = limit
3781 && raw_funding_rates.len() >= limit_val as usize
3782 {
3783 break;
3784 }
3785
3786 if let Some(start_val) = start_ms
3787 && earliest_funding_time <= start_val
3788 {
3789 break;
3790 }
3791
3792 current_end_ms = Some(earliest_funding_time - 1);
3794 }
3795
3796 if let Some(limit_val) = limit {
3797 raw_funding_rates.truncate(limit_val as usize);
3798 }
3799 let mut rates: Vec<FundingRateUpdate> = Vec::with_capacity(raw_funding_rates.len());
3800
3801 for window in raw_funding_rates.windows(2) {
3802 let raw = &window[0];
3803 let timestamp = raw
3804 .funding_rate_timestamp
3805 .parse::<i64>()
3806 .map_err(|_| anyhow::anyhow!("invalid funding_rate_timestamp"))?;
3807 let older_timestamp = window[1]
3808 .funding_rate_timestamp
3809 .parse::<i64>()
3810 .map_err(|_| anyhow::anyhow!("invalid funding_rate_timestamp"))?;
3811
3812 let interval_millis = timestamp - older_timestamp;
3813 let rate = parse_funding_rate(raw, &instrument, Some(interval_millis))?;
3814
3815 rates.push(rate);
3816 }
3817
3818 if let Some(last_raw) = raw_funding_rates.last() {
3819 let rate = parse_funding_rate(last_raw, &instrument, None)?;
3820 rates.push(rate);
3821 }
3822
3823 rates.reverse();
3824
3825 Ok(rates)
3826 }
3827
3828 pub async fn request_orderbook_snapshot(
3846 &self,
3847 product_type: BybitProductType,
3848 instrument_id: InstrumentId,
3849 limit: Option<u32>,
3850 ) -> anyhow::Result<OrderBookDeltas> {
3851 let instrument = self.instrument_from_cache_by_id(instrument_id)?;
3852 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
3853
3854 let mut params_builder = BybitOrderbookParamsBuilder::default();
3855 params_builder.category(product_type);
3856 params_builder.symbol(bybit_symbol.raw_symbol().to_string());
3857
3858 if let Some(limit) = limit {
3859 let max_limit = match product_type {
3860 BybitProductType::Spot => 200,
3861 BybitProductType::Option => 25,
3862 BybitProductType::Linear | BybitProductType::Inverse => 500,
3863 };
3864 let clamped_limit = limit.min(max_limit);
3865 if limit > max_limit {
3866 log::warn!(
3867 "Bybit orderbook snapshot request depth limit exceeds venue maximum; clamping: limit={limit}, clamped_limit={clamped_limit}",
3868 );
3869 }
3870 params_builder.limit(clamped_limit);
3871 }
3872
3873 let params = params_builder.build().build_anyhow()?;
3874 let response = self.inner.get_orderbook(¶ms).await?;
3875
3876 let deltas = parse_orderbook(&response.result, &instrument, None)?;
3877
3878 Ok(deltas)
3879 }
3880
3881 pub async fn request_bars(
3894 &self,
3895 product_type: BybitProductType,
3896 bar_type: BarType,
3897 start: Option<DateTime<Utc>>,
3898 end: Option<DateTime<Utc>>,
3899 limit: Option<u32>,
3900 timestamp_on_close: bool,
3901 ) -> anyhow::Result<Vec<Bar>> {
3902 let instrument_id = bar_type.instrument_id();
3903 let instrument = self.instrument_from_cache_by_id(instrument_id)?;
3904 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
3905
3906 let interval = bar_spec_to_bybit_interval(
3908 bar_type.spec().aggregation,
3909 bar_type.spec().step.get() as u64,
3910 )?;
3911
3912 let start_ms = start.map(|dt| dt.timestamp_millis());
3913 let mut seen_timestamps: AHashSet<i64> = AHashSet::new();
3914 let current_time_ms = get_atomic_clock_realtime().get_time_ms() as i64;
3915
3916 let mut pages: Vec<Vec<Bar>> = Vec::new();
3926 let mut total_bars = 0usize;
3927 let mut current_end = end.map(|dt| dt.timestamp_millis());
3928 let mut page_count = 0;
3929
3930 loop {
3931 page_count += 1;
3932
3933 let mut params_builder = BybitKlinesParamsBuilder::default();
3934 params_builder.category(product_type);
3935 params_builder.symbol(bybit_symbol.raw_symbol().to_string());
3936 params_builder.interval(interval);
3937 params_builder.limit(1000u32); if let Some(start_val) = start_ms {
3940 params_builder.start(start_val);
3941 }
3942
3943 if let Some(end_val) = current_end {
3944 params_builder.end(end_val);
3945 }
3946
3947 let params = params_builder.build().build_anyhow()?;
3948 let response = self.inner.get_klines(¶ms).await?;
3949
3950 let klines = response.result.list;
3951 if klines.is_empty() {
3952 break;
3953 }
3954
3955 let mut klines_with_ts: Vec<(i64, _)> = klines
3957 .into_iter()
3958 .filter_map(|k| k.start.parse::<i64>().ok().map(|ts| (ts, k)))
3959 .collect();
3960
3961 klines_with_ts.sort_by_key(|(ts, _)| *ts);
3962
3963 let has_new = klines_with_ts
3965 .iter()
3966 .any(|(ts, _)| !seen_timestamps.contains(ts));
3967
3968 if !has_new {
3969 break;
3970 }
3971
3972 let mut page_bars = Vec::with_capacity(klines_with_ts.len());
3973
3974 let mut earliest_ts: Option<i64> = None;
3975
3976 for (start_time, kline) in &klines_with_ts {
3977 if earliest_ts.is_none_or(|ts| *start_time < ts) {
3979 earliest_ts = Some(*start_time);
3980 }
3981
3982 let bar_end_time = interval.bar_end_time_ms(*start_time);
3983 if bar_end_time > current_time_ms {
3984 continue;
3985 }
3986
3987 if !seen_timestamps.contains(start_time)
3988 && let Ok(bar) =
3989 parse_kline_bar(kline, &instrument, bar_type, timestamp_on_close, None)
3990 {
3991 page_bars.push(bar);
3992 seen_timestamps.insert(*start_time);
3993 }
3994 }
3995
3996 total_bars += page_bars.len();
3999 pages.push(page_bars);
4000
4001 if let Some(limit_val) = limit
4003 && total_bars >= limit_val as usize
4004 {
4005 break;
4006 }
4007
4008 let Some(earliest_bar_time) = earliest_ts else {
4011 break;
4012 };
4013
4014 if let Some(start_val) = start_ms
4015 && earliest_bar_time <= start_val
4016 {
4017 break;
4018 }
4019
4020 current_end = Some(earliest_bar_time - 1);
4021
4022 if page_count > 100 {
4024 break;
4025 }
4026 }
4027
4028 let mut all_bars: Vec<Bar> = Vec::with_capacity(total_bars);
4030 for page in pages.into_iter().rev() {
4031 all_bars.extend(page);
4032 }
4033
4034 if let Some(limit_val) = limit {
4036 let limit_usize = limit_val as usize;
4037 if all_bars.len() > limit_usize {
4038 let start_idx = all_bars.len() - limit_usize;
4039 return Ok(all_bars[start_idx..].to_vec());
4040 }
4041 }
4042
4043 Ok(all_bars)
4044 }
4045
4046 fn instrument_from_cache_by_id(
4047 &self,
4048 instrument_id: InstrumentId,
4049 ) -> anyhow::Result<InstrumentAny> {
4050 self.get_instrument(&instrument_id.symbol.inner())
4051 .ok_or_else(|| InstrumentLookupError::not_found(instrument_id).into())
4052 }
4053
4054 pub async fn request_fee_rates(
4066 &self,
4067 product_type: BybitProductType,
4068 symbol: Option<String>,
4069 base_coin: Option<String>,
4070 ) -> anyhow::Result<Vec<BybitFeeRate>> {
4071 let params = BybitFeeRateParams {
4072 category: product_type,
4073 symbol,
4074 base_coin,
4075 };
4076
4077 let response = self.inner.get_fee_rate(¶ms).await?;
4078 Ok(response.result.list)
4079 }
4080
4081 pub async fn request_account_state(
4093 &self,
4094 account_type: BybitAccountType,
4095 account_id: AccountId,
4096 ) -> anyhow::Result<AccountState> {
4097 let params = BybitWalletBalanceParams {
4098 account_type,
4099 coin: None,
4100 };
4101
4102 let response = self.inner.get_wallet_balance(¶ms).await?;
4103 let ts_init = self.generate_ts_init();
4104
4105 let wallet_balance = response
4107 .result
4108 .list
4109 .first()
4110 .ok_or_else(|| anyhow::anyhow!("No wallet balance found in response"))?;
4111
4112 parse_account_state(wallet_balance, account_id, ts_init)
4113 }
4114
4115 #[expect(clippy::too_many_arguments)]
4126 pub async fn request_order_status_reports(
4127 &self,
4128 account_id: AccountId,
4129 product_type: BybitProductType,
4130 instrument_id: Option<InstrumentId>,
4131 open_only: bool,
4132 start: Option<DateTime<Utc>>,
4133 end: Option<DateTime<Utc>>,
4134 limit: Option<u32>,
4135 ) -> anyhow::Result<Vec<OrderStatusReport>> {
4136 let symbol_param = if let Some(id) = instrument_id.as_ref() {
4138 let symbol_str = id.symbol.as_str();
4139 if symbol_str.is_empty() {
4140 None
4141 } else {
4142 Some(BybitSymbol::new(symbol_str)?.raw_symbol().to_string())
4143 }
4144 } else {
4145 None
4146 };
4147
4148 let settle_coins_to_query: Vec<Option<String>> =
4151 if product_type == BybitProductType::Linear && symbol_param.is_none() {
4152 vec![Some("USDT".to_string()), Some("USDC".to_string())]
4153 } else {
4154 match product_type {
4155 BybitProductType::Inverse => vec![None],
4156 _ => vec![None],
4157 }
4158 };
4159
4160 let mut all_collected_orders = Vec::new();
4161 let mut total_collected_across_coins = 0;
4162
4163 for settle_coin in settle_coins_to_query {
4164 let remaining_limit = if let Some(limit) = limit {
4165 let remaining = (limit as usize).saturating_sub(total_collected_across_coins);
4166 if remaining == 0 {
4167 break;
4168 }
4169 Some(remaining as u32)
4170 } else {
4171 None
4172 };
4173
4174 let orders_for_coin = if open_only {
4175 let mut all_orders = Vec::new();
4176 let mut seen_ids: AHashSet<Ustr> = AHashSet::new();
4177
4178 let order_filters: Vec<Option<BybitOrderFilter>> =
4181 if product_type == BybitProductType::Option {
4182 vec![None]
4183 } else {
4184 vec![None, Some(BybitOrderFilter::StopOrder)]
4185 };
4186
4187 for order_filter in order_filters {
4188 let mut cursor: Option<String> = None;
4189
4190 loop {
4191 let remaining = if let Some(limit) = remaining_limit {
4192 (limit as usize).saturating_sub(all_orders.len())
4193 } else {
4194 usize::MAX
4195 };
4196
4197 if remaining == 0 {
4198 break;
4199 }
4200
4201 let page_limit = std::cmp::min(remaining, 50);
4203
4204 let mut p = BybitOpenOrdersParamsBuilder::default();
4205 p.category(product_type);
4206
4207 if let Some(symbol) = symbol_param.clone() {
4208 p.symbol(symbol);
4209 }
4210
4211 if let Some(coin) = settle_coin.clone() {
4212 p.settle_coin(coin);
4213 }
4214
4215 if let Some(of) = order_filter {
4216 p.order_filter(of);
4217 }
4218 p.limit(page_limit as u32);
4219
4220 if let Some(c) = cursor {
4221 p.cursor(c);
4222 }
4223 let params = p.build().build_anyhow()?;
4224 let response: BybitOpenOrdersResponse = self
4225 .inner
4226 .send_request(
4227 Method::GET,
4228 BYBIT_ORDER_REALTIME,
4229 Some(¶ms),
4230 None,
4231 true,
4232 )
4233 .await?;
4234
4235 for order in response.result.list {
4236 if seen_ids.insert(order.order_id) {
4237 all_orders.push(order);
4238 }
4239 }
4240
4241 cursor = response.result.next_page_cursor;
4242 if cursor.as_ref().is_none_or(|c| c.is_empty()) {
4243 break;
4244 }
4245 }
4246 }
4247
4248 all_orders
4249 } else {
4250 let mut all_orders = Vec::new();
4253 let mut open_orders = Vec::new();
4254 let mut seen_open_ids: AHashSet<Ustr> = AHashSet::new();
4255
4256 let order_filters: Vec<Option<BybitOrderFilter>> =
4259 if product_type == BybitProductType::Option {
4260 vec![None]
4261 } else {
4262 vec![None, Some(BybitOrderFilter::StopOrder)]
4263 };
4264
4265 for order_filter in &order_filters {
4266 let mut cursor: Option<String> = None;
4267
4268 loop {
4269 let remaining = if let Some(limit) = remaining_limit {
4270 (limit as usize).saturating_sub(open_orders.len())
4271 } else {
4272 usize::MAX
4273 };
4274
4275 if remaining == 0 {
4276 break;
4277 }
4278
4279 let page_limit = std::cmp::min(remaining, 50);
4281
4282 let mut open_params = BybitOpenOrdersParamsBuilder::default();
4283 open_params.category(product_type);
4284
4285 if let Some(symbol) = symbol_param.clone() {
4286 open_params.symbol(symbol);
4287 }
4288
4289 if let Some(coin) = settle_coin.clone() {
4290 open_params.settle_coin(coin);
4291 }
4292
4293 if let Some(of) = order_filter {
4294 open_params.order_filter(*of);
4295 }
4296 open_params.limit(page_limit as u32);
4297
4298 if let Some(c) = cursor {
4299 open_params.cursor(c);
4300 }
4301 let open_params = open_params.build().build_anyhow()?;
4302 let open_response: BybitOpenOrdersResponse = self
4303 .inner
4304 .send_request(
4305 Method::GET,
4306 BYBIT_ORDER_REALTIME,
4307 Some(&open_params),
4308 None,
4309 true,
4310 )
4311 .await?;
4312
4313 for order in open_response.result.list {
4314 if !seen_open_ids.contains(&order.order_id) {
4315 seen_open_ids.insert(order.order_id);
4316 open_orders.push(order);
4317 }
4318 }
4319
4320 cursor = open_response.result.next_page_cursor;
4321 if cursor.as_ref().is_none_or(|c| c.is_empty()) {
4322 break;
4323 }
4324 }
4325 }
4326
4327 let seen_order_ids: AHashSet<Ustr> = seen_open_ids;
4328 let total_open_orders = open_orders.len();
4329
4330 all_orders.extend(open_orders);
4331
4332 let mut total_history_orders = 0;
4333
4334 for order_filter in &order_filters {
4335 let mut cursor: Option<String> = None;
4336
4337 loop {
4338 let total_orders = total_open_orders + total_history_orders;
4339 let remaining = if let Some(limit) = remaining_limit {
4340 (limit as usize).saturating_sub(total_orders)
4341 } else {
4342 usize::MAX
4343 };
4344
4345 if remaining == 0 {
4346 break;
4347 }
4348
4349 let page_limit = std::cmp::min(remaining, 50);
4351
4352 let mut history_params = BybitOrderHistoryParamsBuilder::default();
4353 history_params.category(product_type);
4354
4355 if let Some(symbol) = symbol_param.clone() {
4356 history_params.symbol(symbol);
4357 }
4358
4359 if let Some(coin) = settle_coin.clone() {
4360 history_params.settle_coin(coin);
4361 }
4362
4363 if let Some(of) = order_filter {
4364 history_params.order_filter(*of);
4365 }
4366
4367 if let Some(start) = start {
4368 history_params.start_time(start.timestamp_millis());
4369 }
4370
4371 if let Some(end) = end {
4372 history_params.end_time(end.timestamp_millis());
4373 }
4374 history_params.limit(page_limit as u32);
4375
4376 if let Some(c) = cursor {
4377 history_params.cursor(c);
4378 }
4379 let history_params = history_params.build().build_anyhow()?;
4380 let history_response: BybitOrderHistoryResponse = self
4381 .inner
4382 .send_request(
4383 Method::GET,
4384 BYBIT_ORDER_HISTORY,
4385 Some(&history_params),
4386 None,
4387 true,
4388 )
4389 .await?;
4390
4391 for order in history_response.result.list {
4393 if !seen_order_ids.contains(&order.order_id) {
4394 all_orders.push(order);
4395 total_history_orders += 1;
4396 }
4397 }
4398
4399 cursor = history_response.result.next_page_cursor;
4400 if cursor.as_ref().is_none_or(|c| c.is_empty()) {
4401 break;
4402 }
4403 }
4404 }
4405
4406 all_orders
4407 };
4408
4409 total_collected_across_coins += orders_for_coin.len();
4410 all_collected_orders.extend(orders_for_coin);
4411 }
4412
4413 let ts_init = self.generate_ts_init();
4414
4415 let mut reports = Vec::new();
4416
4417 for order in all_collected_orders {
4418 if let Some(ref instrument_id) = instrument_id {
4419 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
4420
4421 if let Ok(report) =
4422 parse_order_status_report(&order, &instrument, account_id, ts_init)
4423 {
4424 reports.push(report);
4425 }
4426 } else {
4427 if !order.symbol.is_empty() {
4430 let symbol_with_product =
4431 Symbol::from_ustr_unchecked(make_bybit_symbol(order.symbol, product_type));
4432
4433 let Ok(instrument) = self.instrument_from_cache(&symbol_with_product) else {
4434 log::debug!(
4435 "Skipping order report for instrument not in cache: symbol={}, full_symbol={}",
4436 order.symbol,
4437 symbol_with_product
4438 );
4439 continue;
4440 };
4441
4442 match parse_order_status_report(&order, &instrument, account_id, ts_init) {
4443 Ok(report) => reports.push(report),
4444 Err(e) => {
4445 log::error!("Failed to parse order status report: {e}");
4446 }
4447 }
4448 }
4449 }
4450 }
4451
4452 Ok(reports)
4453 }
4454
4455 pub async fn request_fill_reports(
4467 &self,
4468 account_id: AccountId,
4469 product_type: BybitProductType,
4470 instrument_id: Option<InstrumentId>,
4471 start: Option<i64>,
4472 end: Option<i64>,
4473 limit: Option<u32>,
4474 ) -> anyhow::Result<Vec<FillReport>> {
4475 let symbol = if let Some(id) = instrument_id {
4477 let bybit_symbol = BybitSymbol::new(id.symbol.as_str())?;
4478 Some(bybit_symbol.raw_symbol().to_string())
4479 } else {
4480 None
4481 };
4482
4483 let mut all_executions = Vec::new();
4485 let mut cursor: Option<String> = None;
4486 let mut total_executions = 0;
4487
4488 loop {
4489 let remaining = if let Some(limit) = limit {
4491 (limit as usize).saturating_sub(total_executions)
4492 } else {
4493 usize::MAX
4494 };
4495
4496 if remaining == 0 {
4498 break;
4499 }
4500
4501 let page_limit = std::cmp::min(remaining, 100);
4503
4504 let params = BybitTradeHistoryParams {
4505 category: product_type,
4506 symbol: symbol.clone(),
4507 base_coin: None,
4508 order_id: None,
4509 order_link_id: None,
4510 start_time: start,
4511 end_time: end,
4512 exec_type: None,
4513 limit: Some(page_limit as u32),
4514 cursor: cursor.clone(),
4515 };
4516
4517 let response = self.inner.get_trade_history(¶ms).await?;
4518 let list_len = response.result.list.len();
4519 all_executions.extend(response.result.list);
4520 total_executions += list_len;
4521
4522 cursor = response.result.next_page_cursor;
4523 if cursor.is_none() || cursor.as_ref().is_none_or(|c| c.is_empty()) {
4524 break;
4525 }
4526 }
4527
4528 let ts_init = self.generate_ts_init();
4529 let mut reports = Vec::new();
4530
4531 for execution in all_executions {
4532 let symbol_with_product =
4535 Symbol::from_ustr_unchecked(make_bybit_symbol(execution.symbol, product_type));
4536
4537 let Ok(instrument) = self.instrument_from_cache(&symbol_with_product) else {
4538 log::debug!(
4539 "Skipping fill report for instrument not in cache: symbol={}, full_symbol={}",
4540 execution.symbol,
4541 symbol_with_product
4542 );
4543 continue;
4544 };
4545
4546 match parse_fill_report(&execution, account_id, &instrument, ts_init) {
4547 Ok(report) => reports.push(report),
4548 Err(e) => {
4549 log::error!("Failed to parse fill report: {e}");
4550 }
4551 }
4552 }
4553
4554 Ok(reports)
4555 }
4556
4557 pub async fn request_position_status_reports(
4569 &self,
4570 account_id: AccountId,
4571 product_type: BybitProductType,
4572 instrument_id: Option<InstrumentId>,
4573 ) -> anyhow::Result<Vec<PositionStatusReport>> {
4574 if product_type == BybitProductType::Spot {
4576 if self.use_spot_position_reports.load(Ordering::Relaxed) {
4577 return self
4578 .generate_spot_position_reports_from_wallet(account_id, instrument_id)
4579 .await;
4580 } else {
4581 return Ok(Vec::new());
4583 }
4584 }
4585
4586 let ts_init = self.generate_ts_init();
4587 let mut reports = Vec::new();
4588
4589 let symbol = if let Some(id) = instrument_id {
4591 let symbol_str = id.symbol.as_str();
4592 if symbol_str.is_empty() {
4593 anyhow::bail!("InstrumentId symbol is empty");
4594 }
4595 let bybit_symbol = BybitSymbol::new(symbol_str)?;
4596 Some(bybit_symbol.raw_symbol().to_string())
4597 } else {
4598 None
4599 };
4600
4601 if product_type == BybitProductType::Linear && symbol.is_none() {
4604 for settle_coin in ["USDT", "USDC"] {
4606 let mut cursor: Option<String> = None;
4607
4608 loop {
4609 let params = BybitPositionListParams {
4610 category: product_type,
4611 symbol: None,
4612 base_coin: None,
4613 settle_coin: Some(settle_coin.to_string()),
4614 limit: Some(200), cursor: cursor.clone(),
4616 };
4617
4618 let response = self.inner.get_positions(¶ms).await?;
4619
4620 for position in response.result.list {
4621 if position.symbol.is_empty() {
4622 continue;
4623 }
4624
4625 let symbol_with_product = Symbol::new(format!(
4626 "{}{}",
4627 position.symbol.as_str(),
4628 product_type.suffix()
4629 ));
4630
4631 let Ok(instrument) = self.instrument_from_cache(&symbol_with_product)
4632 else {
4633 log::debug!(
4634 "Skipping position report for instrument not in cache: symbol={}, full_symbol={}",
4635 position.symbol,
4636 symbol_with_product
4637 );
4638 continue;
4639 };
4640
4641 match parse_position_status_report(
4642 &position,
4643 account_id,
4644 &instrument,
4645 ts_init,
4646 ) {
4647 Ok(report) => reports.push(report),
4648 Err(e) => {
4649 log::error!("Failed to parse position status report: {e}");
4650 }
4651 }
4652 }
4653
4654 cursor = response.result.next_page_cursor;
4655 if cursor.as_ref().is_none_or(|c| c.is_empty()) {
4656 break;
4657 }
4658 }
4659 }
4660 } else {
4661 let mut cursor: Option<String> = None;
4663
4664 loop {
4665 let params = BybitPositionListParams {
4666 category: product_type,
4667 symbol: symbol.clone(),
4668 base_coin: None,
4669 settle_coin: None,
4670 limit: Some(200), cursor: cursor.clone(),
4672 };
4673
4674 let response = self.inner.get_positions(¶ms).await?;
4675
4676 for position in response.result.list {
4677 if position.symbol.is_empty() {
4678 continue;
4679 }
4680
4681 let symbol_with_product = Symbol::new(format!(
4682 "{}{}",
4683 position.symbol.as_str(),
4684 product_type.suffix()
4685 ));
4686
4687 let Ok(instrument) = self.instrument_from_cache(&symbol_with_product) else {
4688 log::debug!(
4689 "Skipping position report for instrument not in cache: symbol={}, full_symbol={}",
4690 position.symbol,
4691 symbol_with_product
4692 );
4693 continue;
4694 };
4695
4696 match parse_position_status_report(&position, account_id, &instrument, ts_init)
4697 {
4698 Ok(report) => reports.push(report),
4699 Err(e) => {
4700 log::error!("Failed to parse position status report: {e}");
4701 }
4702 }
4703 }
4704
4705 cursor = response.result.next_page_cursor;
4706 if cursor.is_none() || cursor.as_ref().is_none_or(|c| c.is_empty()) {
4707 break;
4708 }
4709 }
4710 }
4711
4712 Ok(reports)
4713 }
4714
4715 async fn query_order_by_id(
4716 &self,
4717 product_type: BybitProductType,
4718 order_id: &str,
4719 endpoint: &str,
4720 context: &str,
4721 ) -> anyhow::Result<BybitOrder> {
4722 let mut query_params = BybitOpenOrdersParamsBuilder::default();
4723 query_params.category(product_type);
4724 query_params.order_id(order_id.to_string());
4725
4726 let query_params = query_params.build().build_anyhow()?;
4727 let order_response: BybitOpenOrdersResponse = self
4728 .inner
4729 .send_request(Method::GET, endpoint, Some(&query_params), None, true)
4730 .await?;
4731
4732 order_response
4733 .result
4734 .list
4735 .into_iter()
4736 .next()
4737 .ok_or_else(|| anyhow::anyhow!("No order returned {context}"))
4738 }
4739}
4740
4741#[cfg(test)]
4742mod tests {
4743 use rstest::rstest;
4744
4745 use super::*;
4746
4747 #[rstest]
4748 fn test_client_creation() {
4749 let client = BybitHttpClient::new(None, 60, 3, 1000, 10_000, 5_000, None);
4750 assert!(client.is_ok());
4751
4752 let client = client.unwrap();
4753 assert!(client.base_url().contains("bybit.com"));
4754 assert!(client.credential().is_none());
4755 }
4756
4757 #[rstest]
4758 fn test_client_with_credentials() {
4759 let client = BybitHttpClient::with_credentials(
4760 "test_key".to_string(),
4761 "test_secret".to_string(),
4762 Some("https://api-testnet.bybit.com".to_string()),
4763 60,
4764 3,
4765 1000,
4766 10_000,
4767 5_000,
4768 None,
4769 );
4770 assert!(client.is_ok());
4771
4772 let client = client.unwrap();
4773 assert!(client.credential().is_some());
4774 }
4775
4776 #[rstest]
4777 fn test_build_path_with_params() {
4778 #[derive(Serialize)]
4779 struct TestParams {
4780 category: String,
4781 symbol: String,
4782 }
4783
4784 let params = TestParams {
4785 category: "linear".to_string(),
4786 symbol: "BTCUSDT".to_string(),
4787 };
4788
4789 let path = BybitRawHttpClient::build_path("/v5/market/test", ¶ms);
4790 assert!(path.is_ok());
4791 assert!(path.unwrap().contains("category=linear"));
4792 }
4793
4794 #[rstest]
4795 fn test_build_path_without_params() {
4796 let params = ();
4797 let path = BybitRawHttpClient::build_path("/v5/market/time", ¶ms);
4798 assert!(path.is_ok());
4799 assert_eq!(path.unwrap(), "/v5/market/time");
4800 }
4801
4802 #[rstest]
4803 fn test_params_serialization_matches_build_path() {
4804 #[derive(Serialize)]
4806 struct TestParams {
4807 category: String,
4808 limit: u32,
4809 }
4810
4811 let params = TestParams {
4812 category: "spot".to_string(),
4813 limit: 50,
4814 };
4815
4816 let old_path = BybitRawHttpClient::build_path(BYBIT_ORDER_REALTIME, ¶ms).unwrap();
4818 let old_query = old_path.split('?').nth(1).unwrap_or("");
4819
4820 let new_query = serde_urlencoded::to_string(¶ms).unwrap();
4822
4823 assert_eq!(old_query, new_query);
4825 }
4826
4827 #[rstest]
4828 fn test_params_serialization_order() {
4829 #[derive(Serialize)]
4831 struct OrderParams {
4832 category: String,
4833 symbol: String,
4834 limit: u32,
4835 }
4836
4837 let params = OrderParams {
4838 category: "spot".to_string(),
4839 symbol: "BTCUSDT".to_string(),
4840 limit: 50,
4841 };
4842
4843 let query1 = serde_urlencoded::to_string(¶ms).unwrap();
4845 let query2 = serde_urlencoded::to_string(¶ms).unwrap();
4846 let query3 = serde_urlencoded::to_string(¶ms).unwrap();
4847
4848 assert_eq!(query1, query2);
4849 assert_eq!(query2, query3);
4850
4851 assert!(query1.contains("category=spot"));
4853 assert!(query1.contains("symbol=BTCUSDT"));
4854 assert!(query1.contains("limit=50"));
4855 }
4856
4857 #[rstest]
4858 #[case(
4859 "https://api-demo.bybit.com",
4860 BybitProductType::Linear,
4861 10001,
4862 "",
4863 "Bybit demo rejected the linear fee rate request via /v5/account/fee-rate \
4864 (error 10001, no message); demo derivatives fee rates appear unsupported, using defaults"
4865 )]
4866 #[case(
4867 "https://api-demo.bybit.com",
4868 BybitProductType::Inverse,
4869 10001,
4870 "",
4871 "Bybit demo rejected the inverse fee rate request via /v5/account/fee-rate \
4872 (error 10001, no message); demo derivatives fee rates appear unsupported, using defaults"
4873 )]
4874 #[case(
4875 "https://api.bybit.com",
4876 BybitProductType::Spot,
4877 10001,
4878 "Parameter error",
4879 "Fee rate request rejected for spot instruments via /v5/account/fee-rate \
4880 (error 10001: Parameter error), using defaults"
4881 )]
4882 #[case(
4883 "https://api-demo.bybit.com",
4884 BybitProductType::Spot,
4885 10001,
4886 "Parameter error",
4887 "Fee rate request rejected for spot instruments via /v5/account/fee-rate \
4888 (error 10001: Parameter error), using defaults"
4889 )]
4890 #[case(
4891 "https://api.bybit.com",
4892 BybitProductType::Linear,
4893 10001,
4894 "Parameter error",
4895 "Fee rate request rejected for linear instruments via /v5/account/fee-rate \
4896 (error 10001: Parameter error), using defaults"
4897 )]
4898 fn test_fee_rate_rejection_warning(
4899 #[case] base_url: &str,
4900 #[case] product_type: BybitProductType,
4901 #[case] error_code: i32,
4902 #[case] message: &str,
4903 #[case] expected: &str,
4904 ) {
4905 let client =
4906 BybitHttpClient::new(Some(base_url.to_string()), 60, 3, 1000, 10_000, 5_000, None)
4907 .unwrap();
4908
4909 let warning = client.fee_rate_rejection_warning(product_type, error_code, message);
4910
4911 assert_eq!(warning, expected);
4912 }
4913
4914 #[rstest]
4915 #[case(10001, "", "error 10001, no message")]
4916 #[case(10001, "Parameter error", "error 10001: Parameter error")]
4917 fn test_format_bybit_error_detail(
4918 #[case] error_code: i32,
4919 #[case] message: &str,
4920 #[case] expected: &str,
4921 ) {
4922 let detail = BybitHttpClient::format_bybit_error_detail(error_code, message);
4923
4924 assert_eq!(detail, expected);
4925 }
4926
4927 #[rstest]
4928 #[case(
4929 10001,
4930 "",
4931 "Option fee rate request rejected via /v5/account/fee-rate \
4932 (error 10001, no message), using defaults"
4933 )]
4934 #[case(
4935 10001,
4936 "Parameter error",
4937 "Option fee rate request rejected via /v5/account/fee-rate \
4938 (error 10001: Parameter error), using defaults"
4939 )]
4940 fn test_option_fee_rate_warning_message(
4941 #[case] error_code: i32,
4942 #[case] message: &str,
4943 #[case] expected: &str,
4944 ) {
4945 let error_detail = BybitHttpClient::format_bybit_error_detail(error_code, message);
4946 let warning = format!(
4947 "Option fee rate request rejected via /v5/account/fee-rate ({error_detail}), using defaults"
4948 );
4949
4950 assert_eq!(warning, expected);
4951 }
4952}