1use std::{
21 cmp::Reverse,
22 collections::HashMap,
23 fmt::{Debug, Display},
24 num::NonZeroU32,
25 sync::{
26 Arc, LazyLock,
27 atomic::{AtomicBool, AtomicU64, Ordering},
28 },
29};
30
31use ahash::{AHashMap, AHashSet};
32use arc_swap::ArcSwap;
33use jiff::Timestamp;
34use nautilus_common::cache::InstrumentLookupError;
35use nautilus_core::{
36 AtomicMap, AtomicTime, env::get_or_env_var_opt, nanos::UnixNanos,
37 time::get_atomic_clock_realtime,
38};
39use nautilus_model::{
40 data::{Bar, BarType, FundingRateUpdate, OrderBookDeltas, TradeTick},
41 enums::{MarketStatusAction, OrderSide, OrderType, PositionSide, TimeInForce},
42 events::account::state::AccountState,
43 identifiers::{AccountId, ClientOrderId, InstrumentId, Symbol, VenueOrderId},
44 instruments::{Instrument, InstrumentAny},
45 reports::{FillReport, OrderStatusReport, PositionStatusReport},
46 types::{Price, Quantity},
47};
48use nautilus_network::{
49 http::{HttpClient, HttpRedirectPolicy, Method, create_standard_nautilus_headers},
50 ratelimiter::quota::Quota,
51 retry::{RetryConfig, RetryError, RetryManager},
52};
53use rust_decimal::Decimal;
54use serde::{Serialize, de::DeserializeOwned};
55use tokio_util::sync::CancellationToken;
56use ustr::Ustr;
57
58use super::{
59 error::{BybitCancelOrderError, BybitHttpError, BybitModifyOrderError, BybitSubmitOrderError},
60 models::{
61 BybitAccountDetailsResponse, BybitAccountInfoResponse, BybitBorrowResponse,
62 BybitEscrowSubMembersResponse, BybitFeeRate, BybitFeeRateResponse, BybitFundingResponse,
63 BybitInstrumentInverse, BybitInstrumentInverseResponse, BybitInstrumentLinear,
64 BybitInstrumentLinearResponse, BybitInstrumentOption, BybitInstrumentOptionResponse,
65 BybitInstrumentSpot, BybitInstrumentSpotResponse, BybitKlinesResponse,
66 BybitNoConvertRepayResponse, BybitOpenOrdersResponse, BybitOrder,
67 BybitOrderHistoryResponse, BybitOrderbookResponse, BybitPlaceOrderResponse,
68 BybitPositionListResponse, BybitRepayResponse, BybitServerTimeResponse,
69 BybitSetLeverageResponse, BybitSetMarginModeResponse, BybitSetTradingStopResponse,
70 BybitSubApiKeyInfo, BybitSubApiKeysResponse, BybitSubMember, BybitSubMembersPagedResponse,
71 BybitSubMembersResponse, BybitSwitchModeResponse, BybitTickerData, BybitTickerOption,
72 BybitTickersOptionResponse, BybitTradeHistoryResponse, BybitTradesResponse,
73 BybitUpdateMasterApiResponse, BybitUpdateSubApiResponse, BybitWalletBalanceResponse,
74 },
75 query::{
76 BybitAmendOrderParamsBuilder, BybitBatchAmendOrderEntryBuilder,
77 BybitBatchCancelOrderEntryBuilder, BybitBatchCancelOrderParamsBuilder,
78 BybitBatchPlaceOrderEntryBuilder, BybitBorrowParamsBuilder,
79 BybitCancelAllOrdersParamsBuilder, BybitCancelOrderParamsBuilder, BybitFeeRateParams,
80 BybitFeeRateParamsBuilder, BybitFundingParams, BybitFundingParamsBuilder,
81 BybitInstrumentsInfoParams, BybitKlinesParams, BybitKlinesParamsBuilder,
82 BybitNativeTpSlParams, BybitNoConvertRepayParamsBuilder, BybitOpenOrdersParamsBuilder,
83 BybitOrderHistoryParamsBuilder, BybitOrderbookParams, BybitOrderbookParamsBuilder,
84 BybitPlaceOrderParamsBuilder, BybitPositionListParams, BybitRepayParamsBuilder,
85 BybitSetLeverageParamsBuilder, BybitSetMarginModeParamsBuilder, BybitSetTradingStopParams,
86 BybitSubApiKeysParams, BybitSubMembersPageParams, BybitSwitchModeParamsBuilder,
87 BybitTickersParams, BybitTradeHistoryParams, BybitTradesParams, BybitTradesParamsBuilder,
88 BybitUpdateMasterApiParams, BybitUpdateSubApiParams, BybitWalletBalanceParams,
89 },
90};
91use crate::common::{
92 consts::{BYBIT_NAUTILUS_BROKER_ID, BYBIT_VENUE},
93 credential::{Credential, credential_env_vars},
94 enums::{
95 BybitAccountType, BybitBboSideType, BybitContractType, BybitEnvironment, BybitExecType,
96 BybitMarginMode, BybitOpenOnly, BybitOrderFilter, BybitOrderSide, BybitOrderSmpType,
97 BybitOrderType, BybitPositionIdx, BybitPositionMode, BybitProductType, BybitRepayStatus,
98 BybitTpSlMode,
99 },
100 models::{BybitCursorListResponse, BybitErrorCheck, BybitResponseCheck},
101 parse::{
102 bar_spec_to_bybit_interval, bybit_rejection_due_post_only, make_bybit_symbol,
103 map_time_in_force, parse_account_state, parse_fill_report, parse_funding_rate,
104 parse_inverse_instrument, parse_kline_bar, parse_linear_instrument,
105 parse_option_instrument, parse_order_status_report, parse_orderbook,
106 parse_position_status_report, parse_spot_instrument, parse_trade_tick, spot_leverage,
107 spot_market_unit, trigger_direction,
108 },
109 rate_limit::{
110 BYBIT_RATE_LIMIT_HEADER, BYBIT_RATE_LIMIT_RESET_HEADER, BYBIT_RATE_LIMIT_STATUS_HEADER,
111 BybitRateLimiter, batch_call_limit, batch_endpoint_limit, batch_send_limit, batch_weight,
112 category_from_payload,
113 },
114 retry::should_retry_http,
115 symbol::BybitSymbol,
116 urls::bybit_http_base_url,
117};
118
119const DEFAULT_RECV_WINDOW_MS: u64 = 5_000;
120
121trait BuilderResultExt<T> {
122 fn build_anyhow(self) -> anyhow::Result<T>;
123}
124
125impl<T, E: Display> BuilderResultExt<T> for Result<T, E> {
126 fn build_anyhow(self) -> anyhow::Result<T> {
127 self.map_err(|e| anyhow::anyhow!("{e}"))
128 }
129}
130
131const BYBIT_INSTRUMENTS_INFO: &str = "/v5/market/instruments-info";
132const BYBIT_ORDER_REALTIME: &str = "/v5/order/realtime";
133const BYBIT_ORDER_HISTORY: &str = "/v5/order/history";
134const BYBIT_EXECUTION_LIST: &str = "/v5/execution/list";
135const BYBIT_POSITION_LIST: &str = "/v5/position/list";
136
137#[derive(Debug, Default)]
149struct CursorWalk {
150 followed: AHashSet<String>,
151 last: Option<String>,
152}
153
154impl CursorWalk {
155 fn advance(
163 &mut self,
164 endpoint: &str,
165 cursor: Option<String>,
166 ) -> anyhow::Result<Option<String>> {
167 let Some(cursor) = cursor.filter(|cursor| !cursor.is_empty()) else {
168 return Ok(None);
169 };
170
171 anyhow::ensure!(
172 self.last.as_deref() != Some(cursor.as_str()),
173 "{endpoint} pagination cursor did not advance from {cursor:?}",
174 );
175 anyhow::ensure!(
176 self.followed.insert(cursor.clone()),
177 "{endpoint} pagination repeated cursor {cursor:?}",
178 );
179
180 self.last = Some(cursor.clone());
181
182 Ok(Some(cursor))
183 }
184}
185
186pub static BYBIT_REST_QUOTA: LazyLock<Quota> = LazyLock::new(|| {
188 Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant")
189});
190
191pub static BYBIT_REPAY_QUOTA: LazyLock<Quota> = LazyLock::new(|| {
193 Quota::per_second(NonZeroU32::new(1).expect("non-zero")).expect("valid constant")
194});
195
196#[cfg_attr(
201 feature = "python",
202 pyo3::pyclass(module = "nautilus_trader.adapters.bybit", from_py_object)
203)]
204#[cfg_attr(
205 feature = "python",
206 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.bybit")
207)]
208#[derive(Clone)]
209pub struct BybitRawHttpClient {
210 base_url: String,
211 client: Arc<ArcSwap<HttpClient>>,
212 rate_limiter: BybitRateLimiter,
213 credential: Option<Credential>,
214 recv_window_ms: u64,
215 timeout_secs: u64,
216 proxy_url: Option<String>,
217 session_generation: Arc<AtomicU64>,
218 retry_manager: RetryManager<BybitHttpError>,
219 cancellation_token: Arc<parking_lot::Mutex<CancellationToken>>,
220}
221
222impl Default for BybitRawHttpClient {
223 fn default() -> Self {
224 Self::new(None, 60, 3, 1000, 10_000, DEFAULT_RECV_WINDOW_MS, None)
225 .expect("Failed to create default BybitRawHttpClient")
226 }
227}
228
229impl Debug for BybitRawHttpClient {
230 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
231 f.debug_struct(stringify!(BybitRawHttpClient))
232 .field("base_url", &self.base_url)
233 .field("has_credentials", &self.credential.is_some())
234 .field("recv_window_ms", &self.recv_window_ms)
235 .finish()
236 }
237}
238
239impl BybitRawHttpClient {
240 pub fn cancel_all_requests(&self) {
242 self.cancellation_token.lock().cancel();
243 }
244
245 pub fn reset_cancellation_token(&self) {
248 let mut guard = self.cancellation_token.lock();
249 *guard = CancellationToken::new();
250 }
251
252 pub fn cancellation_token(&self) -> CancellationToken {
254 self.cancellation_token.lock().clone()
255 }
256
257 pub fn new(
263 base_url: Option<String>,
264 timeout_secs: u64,
265 max_retries: u32,
266 retry_delay_ms: u64,
267 retry_delay_max_ms: u64,
268 recv_window_ms: u64,
269 proxy_url: Option<String>,
270 ) -> Result<Self, BybitHttpError> {
271 let retry_config = RetryConfig {
272 max_retries,
273 initial_delay_ms: retry_delay_ms,
274 max_delay_ms: retry_delay_max_ms,
275 backoff_factor: 2.0,
276 jitter_ms: 1000,
277 operation_timeout_ms: Some(60_000),
278 immediate_first: false,
279 max_elapsed_ms: Some(180_000),
280 };
281
282 let retry_manager = RetryManager::new(retry_config);
283 let base_url =
284 base_url.unwrap_or_else(|| bybit_http_base_url(BybitEnvironment::Mainnet).to_string());
285 let rate_limiter = BybitRateLimiter::for_http(&base_url, None, proxy_url.as_deref());
286 let client = Self::build_http_client(timeout_secs, proxy_url.clone())?;
287 let session_generation = rate_limiter.http_session_generation();
288
289 Ok(Self {
290 base_url,
291 client: Arc::new(ArcSwap::from_pointee(client)),
292 rate_limiter,
293 credential: None,
294 recv_window_ms,
295 timeout_secs,
296 proxy_url,
297 session_generation: Arc::new(AtomicU64::new(session_generation)),
298 retry_manager,
299 cancellation_token: Arc::new(parking_lot::Mutex::new(CancellationToken::new())),
300 })
301 }
302
303 #[expect(clippy::too_many_arguments)]
309 pub fn with_credentials(
310 api_key: String,
311 api_secret: String,
312 base_url: Option<String>,
313 timeout_secs: u64,
314 max_retries: u32,
315 retry_delay_ms: u64,
316 retry_delay_max_ms: u64,
317 recv_window_ms: u64,
318 proxy_url: Option<String>,
319 ) -> Result<Self, BybitHttpError> {
320 let retry_config = RetryConfig {
321 max_retries,
322 initial_delay_ms: retry_delay_ms,
323 max_delay_ms: retry_delay_max_ms,
324 backoff_factor: 2.0,
325 jitter_ms: 1000,
326 operation_timeout_ms: Some(60_000),
327 immediate_first: false,
328 max_elapsed_ms: Some(180_000),
329 };
330
331 let retry_manager = RetryManager::new(retry_config);
332 let base_url =
333 base_url.unwrap_or_else(|| bybit_http_base_url(BybitEnvironment::Mainnet).to_string());
334 let credential = Credential::new(api_key, api_secret);
335 let rate_limiter =
336 BybitRateLimiter::for_http(&base_url, Some(credential.api_key()), proxy_url.as_deref());
337 let client = Self::build_http_client(timeout_secs, proxy_url.clone())?;
338 let session_generation = rate_limiter.http_session_generation();
339
340 Ok(Self {
341 base_url,
342 client: Arc::new(ArcSwap::from_pointee(client)),
343 rate_limiter,
344 credential: Some(credential),
345 recv_window_ms,
346 timeout_secs,
347 proxy_url,
348 session_generation: Arc::new(AtomicU64::new(session_generation)),
349 retry_manager,
350 cancellation_token: Arc::new(parking_lot::Mutex::new(CancellationToken::new())),
351 })
352 }
353
354 #[expect(clippy::too_many_arguments)]
366 pub fn new_with_env(
367 api_key: Option<String>,
368 api_secret: Option<String>,
369 base_url: Option<String>,
370 demo: bool,
371 testnet: bool,
372 timeout_secs: u64,
373 max_retries: u32,
374 retry_delay_ms: u64,
375 retry_delay_max_ms: u64,
376 recv_window_ms: u64,
377 proxy_url: Option<String>,
378 ) -> Result<Self, BybitHttpError> {
379 let environment = if demo {
380 BybitEnvironment::Demo
381 } else if testnet {
382 BybitEnvironment::Testnet
383 } else {
384 BybitEnvironment::Mainnet
385 };
386 let base_url =
387 Some(base_url.unwrap_or_else(|| bybit_http_base_url(environment).to_string()));
388 let (key_var, secret_var) = credential_env_vars(environment);
389 let key = get_or_env_var_opt(api_key, key_var);
390 let secret = get_or_env_var_opt(api_secret, secret_var);
391
392 if let (Some(k), Some(s)) = (key, secret) {
393 Self::with_credentials(
394 k,
395 s,
396 base_url,
397 timeout_secs,
398 max_retries,
399 retry_delay_ms,
400 retry_delay_max_ms,
401 recv_window_ms,
402 proxy_url,
403 )
404 } else {
405 Self::new(
406 base_url,
407 timeout_secs,
408 max_retries,
409 retry_delay_ms,
410 retry_delay_max_ms,
411 recv_window_ms,
412 proxy_url,
413 )
414 }
415 }
416
417 fn default_headers() -> HashMap<String, String> {
418 let mut headers: HashMap<String, String> =
419 create_standard_nautilus_headers().into_iter().collect();
420 headers.insert(
421 "X-Referer".to_string(),
422 BYBIT_NAUTILUS_BROKER_ID.to_string(),
423 );
424 headers
425 }
426
427 fn build_http_client(
428 timeout_secs: u64,
429 proxy_url: Option<String>,
430 ) -> Result<HttpClient, BybitHttpError> {
431 HttpClient::builder()
432 .redirect_policy(HttpRedirectPolicy::Reject)
433 .headers(Self::default_headers())
434 .header_keys(vec![
435 BYBIT_RATE_LIMIT_HEADER.to_string(),
436 BYBIT_RATE_LIMIT_STATUS_HEADER.to_string(),
437 BYBIT_RATE_LIMIT_RESET_HEADER.to_string(),
438 ])
439 .timeout_secs(timeout_secs)
440 .maybe_proxy_url(proxy_url)
441 .rate_limiters(Vec::new())
442 .build()
443 .map_err(|e| BybitHttpError::NetworkError(format!("Failed to create HTTP client: {e}")))
444 }
445
446 fn refresh_http_session(&self, generation: u64) -> Result<(), BybitHttpError> {
447 let current = self.session_generation.load(Ordering::Acquire);
448 if current == generation {
449 return Ok(());
450 }
451
452 let client = Self::build_http_client(self.timeout_secs, self.proxy_url.clone())?;
453 self.client.store(Arc::new(client));
454 self.session_generation.store(generation, Ordering::Release);
455 Ok(())
456 }
457
458 fn request_rate_limit(
459 endpoint: &str,
460 payload: Option<&str>,
461 ) -> (Option<BybitProductType>, u32) {
462 let category = category_from_payload(payload);
463 let weight = if endpoint.ends_with("-batch") {
464 let order_count = payload
465 .and_then(|payload| serde_json::from_str::<serde_json::Value>(payload).ok())
466 .and_then(|value| value.get("request")?.as_array().map(Vec::len))
467 .unwrap_or(1);
468 category.map_or(1, |category| batch_weight(category, order_count))
469 } else {
470 1
471 };
472 (category, weight)
473 }
474
475 fn observe_rate_limit_headers(
476 &self,
477 endpoint: &str,
478 category: Option<BybitProductType>,
479 headers: &HashMap<String, String>,
480 ) {
481 let Some(limit) = headers
482 .get(BYBIT_RATE_LIMIT_HEADER)
483 .and_then(|value| value.parse::<u32>().ok())
484 else {
485 return;
486 };
487 let Some(remaining) = headers
488 .get(BYBIT_RATE_LIMIT_STATUS_HEADER)
489 .and_then(|value| value.parse::<u32>().ok())
490 else {
491 return;
492 };
493 let reset_timestamp_ms = headers
494 .get(BYBIT_RATE_LIMIT_RESET_HEADER)
495 .and_then(|value| value.parse::<i64>().ok());
496 self.rate_limiter
497 .observe_account(endpoint, category, limit, remaining, reset_timestamp_ms);
498 }
499
500 fn is_rate_limit_403(status: u16, body: &str) -> bool {
501 status == 403 && body.to_ascii_lowercase().contains("access too frequent")
502 }
503
504 fn sign_request(
505 &self,
506 timestamp: &str,
507 params: Option<&str>,
508 ) -> Result<HashMap<String, String>, BybitHttpError> {
509 let credential = self
510 .credential
511 .as_ref()
512 .ok_or(BybitHttpError::MissingCredentials)?;
513
514 let signature = credential.sign_with_payload(timestamp, self.recv_window_ms, params);
515
516 let mut headers = HashMap::new();
517 headers.insert(
518 "X-BAPI-API-KEY".to_string(),
519 credential.api_key().to_string(),
520 );
521 headers.insert("X-BAPI-TIMESTAMP".to_string(), timestamp.to_string());
522 headers.insert("X-BAPI-SIGN".to_string(), signature);
523 headers.insert(
524 "X-BAPI-RECV-WINDOW".to_string(),
525 self.recv_window_ms.to_string(),
526 );
527
528 Ok(headers)
529 }
530
531 async fn send_request<T: DeserializeOwned + BybitResponseCheck, P: Serialize>(
532 &self,
533 method: Method,
534 endpoint: &str,
535 params: Option<&P>,
536 body: Option<Vec<u8>>,
537 authenticate: bool,
538 ) -> Result<T, BybitHttpError> {
539 let endpoint = endpoint.to_string();
540 let url = format!("{}{endpoint}", self.base_url);
541 let method_clone = method.clone();
542 let body_clone = body.clone();
543
544 let params_str = if method == Method::GET {
546 params
547 .map(serde_urlencoded::to_string)
548 .transpose()
549 .map_err(|e| {
550 BybitHttpError::JsonError(format!("Failed to serialize params: {e}"))
551 })?
552 } else {
553 None
554 };
555
556 let operation = || {
557 let url = url.clone();
558 let method = method_clone.clone();
559 let body = body_clone.clone();
560 let endpoint = endpoint.clone();
561 let params_str = params_str.clone();
562
563 async move {
564 let full_url = if let Some(ref query) = params_str {
565 if query.is_empty() {
566 url
567 } else {
568 format!("{url}?{query}")
569 }
570 } else {
571 url
572 };
573
574 let sign_payload = if method == Method::GET {
575 params_str.as_deref()
576 } else {
577 body.as_ref()
578 .and_then(|body| std::str::from_utf8(body).ok())
579 };
580 let (category, weight) = Self::request_rate_limit(&endpoint, sign_payload);
581
582 let generation = self.rate_limiter.http_session_generation();
583 self.refresh_http_session(generation)?;
584 self.rate_limiter
585 .acquire_http(&endpoint, category, weight, authenticate)
586 .await
587 .map_err(BybitHttpError::ValidationError)?;
588 let generation = self.rate_limiter.http_session_generation();
589 self.refresh_http_session(generation)?;
590
591 let mut headers = Self::default_headers();
592
593 if authenticate {
594 let timestamp = get_atomic_clock_realtime().get_time_ms().to_string();
595 let auth_headers = self.sign_request(×tamp, sign_payload)?;
596 headers.extend(auth_headers);
597 }
598
599 if method == Method::POST || method == Method::PUT {
600 headers.insert("Content-Type".to_string(), "application/json".to_string());
601 }
602
603 let response = self
604 .client
605 .load()
606 .request(method, full_url, None, Some(headers), body, None, None)
607 .await?;
608
609 self.observe_rate_limit_headers(&endpoint, category, &response.headers);
610
611 if response.status.as_u16() >= 400 {
612 let body = String::from_utf8_lossy(&response.body).to_string();
613 if Self::is_rate_limit_403(response.status.as_u16(), &body) {
614 let generation = self.rate_limiter.reset_http_sessions();
615 self.refresh_http_session(generation)?;
616 }
617 return Err(BybitHttpError::UnexpectedStatus {
618 status: response.status.as_u16(),
619 body,
620 });
621 }
622
623 match serde_json::from_slice::<T>(&response.body) {
625 Ok(result) => {
626 if result.ret_code() != 0 {
628 return Err(BybitHttpError::BybitError {
629 error_code: result.ret_code() as i32,
630 message: result.ret_msg().to_string(),
631 });
632 }
633 Ok(result)
634 }
635 Err(json_err) => {
636 if let Ok(error_check) =
639 serde_json::from_slice::<BybitErrorCheck>(&response.body)
640 && error_check.ret_code != 0
641 {
642 return Err(BybitHttpError::BybitError {
643 error_code: error_check.ret_code as i32,
644 message: error_check.ret_msg,
645 });
646 }
647 Err(json_err.into())
649 }
650 }
651 }
652 };
653
654 let create_error = |error: RetryError| -> BybitHttpError {
655 match error {
656 RetryError::Canceled => {
657 BybitHttpError::Canceled("Adapter disconnecting or shutting down".to_string())
658 }
659 error => BybitHttpError::NetworkError(error.to_string()),
660 }
661 };
662
663 let token = self.cancellation_token();
664
665 self.retry_manager
666 .invocation(
667 endpoint.as_str(),
668 operation,
669 should_retry_http,
670 create_error,
671 )
672 .cancellation_token(&token)
673 .execute()
674 .await
675 }
676
677 #[cfg(test)]
678 fn build_path<S: Serialize>(base: &str, params: &S) -> Result<String, BybitHttpError> {
679 let query = serde_urlencoded::to_string(params)
680 .map_err(|e| BybitHttpError::JsonError(e.to_string()))?;
681
682 if query.is_empty() {
683 Ok(base.to_owned())
684 } else {
685 Ok(format!("{base}?{query}"))
686 }
687 }
688
689 pub async fn get_server_time(&self) -> Result<BybitServerTimeResponse, BybitHttpError> {
699 self.send_request::<_, ()>(Method::GET, "/v5/market/time", None, None, false)
700 .await
701 }
702
703 pub async fn get_instruments<T: DeserializeOwned + BybitResponseCheck>(
713 &self,
714 params: &BybitInstrumentsInfoParams,
715 ) -> Result<T, BybitHttpError> {
716 self.send_request(
717 Method::GET,
718 BYBIT_INSTRUMENTS_INFO,
719 Some(params),
720 None,
721 false,
722 )
723 .await
724 }
725
726 pub async fn get_instruments_spot(
736 &self,
737 params: &BybitInstrumentsInfoParams,
738 ) -> Result<BybitInstrumentSpotResponse, BybitHttpError> {
739 self.get_instruments(params).await
740 }
741
742 pub async fn get_instruments_linear(
752 &self,
753 params: &BybitInstrumentsInfoParams,
754 ) -> Result<BybitInstrumentLinearResponse, BybitHttpError> {
755 self.get_instruments(params).await
756 }
757
758 pub async fn get_instruments_inverse(
768 &self,
769 params: &BybitInstrumentsInfoParams,
770 ) -> Result<BybitInstrumentInverseResponse, BybitHttpError> {
771 self.get_instruments(params).await
772 }
773
774 pub async fn get_instruments_option(
784 &self,
785 params: &BybitInstrumentsInfoParams,
786 ) -> Result<BybitInstrumentOptionResponse, BybitHttpError> {
787 self.get_instruments(params).await
788 }
789
790 pub async fn get_klines(
800 &self,
801 params: &BybitKlinesParams,
802 ) -> Result<BybitKlinesResponse, BybitHttpError> {
803 self.send_request(Method::GET, "/v5/market/kline", Some(params), None, false)
804 .await
805 }
806
807 pub async fn get_recent_trades(
817 &self,
818 params: &BybitTradesParams,
819 ) -> Result<BybitTradesResponse, BybitHttpError> {
820 self.send_request(
821 Method::GET,
822 "/v5/market/recent-trade",
823 Some(params),
824 None,
825 false,
826 )
827 .await
828 }
829
830 pub async fn get_funding_history(
840 &self,
841 params: &BybitFundingParams,
842 ) -> Result<BybitFundingResponse, BybitHttpError> {
843 self.send_request(
844 Method::GET,
845 "/v5/market/funding/history",
846 Some(params),
847 None,
848 false,
849 )
850 .await
851 }
852
853 pub async fn get_orderbook(
863 &self,
864 params: &BybitOrderbookParams,
865 ) -> Result<BybitOrderbookResponse, BybitHttpError> {
866 self.send_request(
867 Method::GET,
868 "/v5/market/orderbook",
869 Some(params),
870 None,
871 false,
872 )
873 .await
874 }
875
876 #[expect(clippy::too_many_arguments)]
890 pub async fn get_open_orders(
891 &self,
892 category: BybitProductType,
893 symbol: Option<String>,
894 base_coin: Option<String>,
895 settle_coin: Option<String>,
896 order_id: Option<String>,
897 order_link_id: Option<String>,
898 open_only: Option<BybitOpenOnly>,
899 order_filter: Option<BybitOrderFilter>,
900 limit: Option<u32>,
901 cursor: Option<String>,
902 ) -> Result<BybitOpenOrdersResponse, BybitHttpError> {
903 let mut builder = BybitOpenOrdersParamsBuilder::default();
904 builder.category(category);
905
906 if let Some(s) = symbol {
907 builder.symbol(s);
908 }
909
910 if let Some(bc) = base_coin {
911 builder.base_coin(bc);
912 }
913
914 if let Some(sc) = settle_coin {
915 builder.settle_coin(sc);
916 }
917
918 if let Some(oi) = order_id {
919 builder.order_id(oi);
920 }
921
922 if let Some(ol) = order_link_id {
923 builder.order_link_id(ol);
924 }
925
926 if let Some(oo) = open_only {
927 builder.open_only(oo);
928 }
929
930 if let Some(of) = order_filter {
931 builder.order_filter(of);
932 }
933
934 if let Some(l) = limit {
935 builder.limit(l);
936 }
937
938 if let Some(c) = cursor {
939 builder.cursor(c);
940 }
941
942 let params = builder
943 .build()
944 .expect("Failed to build BybitOpenOrdersParams");
945
946 self.send_request(Method::GET, BYBIT_ORDER_REALTIME, Some(¶ms), None, true)
947 .await
948 }
949
950 pub async fn place_order(
960 &self,
961 request: &serde_json::Value,
962 ) -> Result<BybitPlaceOrderResponse, BybitHttpError> {
963 let body = serde_json::to_vec(request)?;
964 self.send_request::<_, ()>(Method::POST, "/v5/order/create", None, Some(body), true)
965 .await
966 }
967
968 pub async fn get_wallet_balance(
978 &self,
979 params: &BybitWalletBalanceParams,
980 ) -> Result<BybitWalletBalanceResponse, BybitHttpError> {
981 self.send_request(
982 Method::GET,
983 "/v5/account/wallet-balance",
984 Some(params),
985 None,
986 true,
987 )
988 .await
989 }
990
991 pub async fn get_account_info(&self) -> Result<BybitAccountInfoResponse, BybitHttpError> {
1001 self.send_request::<_, ()>(Method::GET, "/v5/account/info", None, None, true)
1002 .await
1003 }
1004
1005 pub async fn get_account_details(&self) -> Result<BybitAccountDetailsResponse, BybitHttpError> {
1015 self.send_request::<_, ()>(Method::GET, "/v5/user/query-api", None, None, true)
1016 .await
1017 }
1018
1019 pub async fn update_sub_api_key(
1029 &self,
1030 params: &BybitUpdateSubApiParams,
1031 ) -> Result<BybitUpdateSubApiResponse, BybitHttpError> {
1032 let body = serde_json::to_vec(params)?;
1033 self.send_request::<_, ()>(
1034 Method::POST,
1035 "/v5/user/update-sub-api",
1036 None,
1037 Some(body),
1038 true,
1039 )
1040 .await
1041 }
1042
1043 pub async fn update_master_api_key(
1053 &self,
1054 params: &BybitUpdateMasterApiParams,
1055 ) -> Result<BybitUpdateMasterApiResponse, BybitHttpError> {
1056 let body = serde_json::to_vec(params)?;
1057 self.send_request::<_, ()>(Method::POST, "/v5/user/update-api", None, Some(body), true)
1058 .await
1059 }
1060
1061 pub async fn get_sub_members(&self) -> Result<BybitSubMembersResponse, BybitHttpError> {
1071 self.send_request::<_, ()>(Method::GET, "/v5/user/query-sub-members", None, None, true)
1072 .await
1073 }
1074
1075 pub async fn get_sub_members_paged(
1085 &self,
1086 params: &BybitSubMembersPageParams,
1087 ) -> Result<BybitSubMembersPagedResponse, BybitHttpError> {
1088 self.send_request(Method::GET, "/v5/user/submembers", Some(params), None, true)
1089 .await
1090 }
1091
1092 pub async fn get_escrow_sub_members(
1102 &self,
1103 params: &BybitSubMembersPageParams,
1104 ) -> Result<BybitEscrowSubMembersResponse, BybitHttpError> {
1105 self.send_request(
1106 Method::GET,
1107 "/v5/user/escrow_sub_members",
1108 Some(params),
1109 None,
1110 true,
1111 )
1112 .await
1113 }
1114
1115 pub async fn get_sub_api_keys(
1125 &self,
1126 params: &BybitSubApiKeysParams,
1127 ) -> Result<BybitSubApiKeysResponse, BybitHttpError> {
1128 self.send_request(
1129 Method::GET,
1130 "/v5/user/sub-apikeys",
1131 Some(params),
1132 None,
1133 true,
1134 )
1135 .await
1136 }
1137
1138 pub async fn fetch_all_sub_members_paged(
1145 &self,
1146 page_size: Option<u32>,
1147 ) -> Result<Vec<BybitSubMember>, BybitHttpError> {
1148 let mut members = Vec::new();
1149 let mut cursor: Option<String> = None;
1150
1151 loop {
1152 let params = BybitSubMembersPageParams {
1153 page_size,
1154 next_cursor: cursor.take(),
1155 };
1156 let mut page = self.get_sub_members_paged(¶ms).await?;
1157 let next = page.result.continuation_cursor().map(str::to_owned);
1158 members.append(&mut page.result.sub_members);
1159
1160 match next {
1161 Some(c) => cursor = Some(c),
1162 None => break,
1163 }
1164 }
1165
1166 Ok(members)
1167 }
1168
1169 pub async fn fetch_all_escrow_sub_members(
1177 &self,
1178 page_size: Option<u32>,
1179 ) -> Result<Vec<BybitSubMember>, BybitHttpError> {
1180 let mut members = Vec::new();
1181 let mut cursor: Option<String> = None;
1182
1183 loop {
1184 let params = BybitSubMembersPageParams {
1185 page_size,
1186 next_cursor: cursor.take(),
1187 };
1188 let mut page = self.get_escrow_sub_members(¶ms).await?;
1189 let next = page.result.continuation_cursor().map(str::to_owned);
1190 members.append(&mut page.result.sub_members);
1191
1192 match next {
1193 Some(c) => cursor = Some(c),
1194 None => break,
1195 }
1196 }
1197
1198 Ok(members)
1199 }
1200
1201 pub async fn fetch_all_sub_api_keys(
1209 &self,
1210 sub_member_id: impl Into<String>,
1211 limit: Option<u32>,
1212 ) -> Result<Vec<BybitSubApiKeyInfo>, BybitHttpError> {
1213 let sub_member_id = sub_member_id.into();
1214 let mut keys = Vec::new();
1215 let mut cursor: Option<String> = None;
1216
1217 loop {
1218 let params = BybitSubApiKeysParams {
1219 sub_member_id: sub_member_id.clone(),
1220 limit,
1221 cursor: cursor.take(),
1222 };
1223 let mut page = self.get_sub_api_keys(¶ms).await?;
1224 let next = page.result.continuation_cursor().map(str::to_owned);
1225 keys.append(&mut page.result.keys);
1226
1227 match next {
1228 Some(c) => cursor = Some(c),
1229 None => break,
1230 }
1231 }
1232
1233 Ok(keys)
1234 }
1235
1236 pub async fn get_fee_rate(
1246 &self,
1247 params: &BybitFeeRateParams,
1248 ) -> Result<BybitFeeRateResponse, BybitHttpError> {
1249 self.send_request(
1250 Method::GET,
1251 "/v5/account/fee-rate",
1252 Some(params),
1253 None,
1254 true,
1255 )
1256 .await
1257 }
1258
1259 pub async fn set_margin_mode(
1276 &self,
1277 margin_mode: BybitMarginMode,
1278 ) -> Result<BybitSetMarginModeResponse, BybitHttpError> {
1279 let params = BybitSetMarginModeParamsBuilder::default()
1280 .set_margin_mode(margin_mode)
1281 .build()
1282 .expect("Failed to build BybitSetMarginModeParams");
1283
1284 let body = serde_json::to_vec(¶ms)?;
1285 self.send_request::<_, ()>(
1286 Method::POST,
1287 "/v5/account/set-margin-mode",
1288 None,
1289 Some(body),
1290 true,
1291 )
1292 .await
1293 }
1294
1295 pub async fn set_leverage(
1312 &self,
1313 product_type: BybitProductType,
1314 symbol: &str,
1315 buy_leverage: &str,
1316 sell_leverage: &str,
1317 ) -> Result<BybitSetLeverageResponse, BybitHttpError> {
1318 let params = BybitSetLeverageParamsBuilder::default()
1319 .category(product_type)
1320 .symbol(symbol.to_string())
1321 .buy_leverage(buy_leverage.to_string())
1322 .sell_leverage(sell_leverage.to_string())
1323 .build()
1324 .expect("Failed to build BybitSetLeverageParams");
1325
1326 let body = serde_json::to_vec(¶ms)?;
1327 self.send_request::<_, ()>(
1328 Method::POST,
1329 "/v5/position/set-leverage",
1330 None,
1331 Some(body),
1332 true,
1333 )
1334 .await
1335 }
1336
1337 pub async fn switch_mode(
1354 &self,
1355 product_type: BybitProductType,
1356 mode: BybitPositionMode,
1357 symbol: Option<String>,
1358 coin: Option<String>,
1359 ) -> Result<BybitSwitchModeResponse, BybitHttpError> {
1360 let mut builder = BybitSwitchModeParamsBuilder::default();
1361 builder.category(product_type);
1362 builder.mode(mode);
1363
1364 if let Some(s) = symbol {
1365 builder.symbol(s);
1366 }
1367
1368 if let Some(c) = coin {
1369 builder.coin(c);
1370 }
1371
1372 let params = builder
1373 .build()
1374 .expect("Failed to build BybitSwitchModeParams");
1375
1376 let body = serde_json::to_vec(¶ms)?;
1377 self.send_request::<_, ()>(
1378 Method::POST,
1379 "/v5/position/switch-mode",
1380 None,
1381 Some(body),
1382 true,
1383 )
1384 .await
1385 }
1386
1387 pub async fn set_trading_stop(
1400 &self,
1401 params: &BybitSetTradingStopParams,
1402 ) -> Result<BybitSetTradingStopResponse, BybitHttpError> {
1403 let body = serde_json::to_vec(params)?;
1404 self.send_request::<_, ()>(
1405 Method::POST,
1406 "/v5/position/trading-stop",
1407 None,
1408 Some(body),
1409 true,
1410 )
1411 .await
1412 }
1413
1414 pub async fn borrow(
1431 &self,
1432 coin: &str,
1433 amount: &str,
1434 ) -> Result<BybitBorrowResponse, BybitHttpError> {
1435 let params = BybitBorrowParamsBuilder::default()
1436 .coin(coin.to_string())
1437 .amount(amount.to_string())
1438 .build()
1439 .expect("Failed to build BybitBorrowParams");
1440
1441 let body = serde_json::to_vec(¶ms)?;
1442 self.send_request::<_, ()>(Method::POST, "/v5/account/borrow", None, Some(body), true)
1443 .await
1444 }
1445
1446 pub async fn no_convert_repay(
1464 &self,
1465 coin: &str,
1466 amount: Option<&str>,
1467 ) -> Result<BybitNoConvertRepayResponse, BybitHttpError> {
1468 let mut builder = BybitNoConvertRepayParamsBuilder::default();
1469 builder.coin(coin.to_string());
1470
1471 if let Some(amt) = amount {
1472 builder.amount(amt.to_string());
1473 }
1474
1475 let params = builder
1476 .build()
1477 .expect("Failed to build BybitNoConvertRepayParams");
1478
1479 if let Ok(params_json) = serde_json::to_string(¶ms) {
1480 log::debug!("Repay request params: {params_json}");
1481 }
1482
1483 let body = serde_json::to_vec(¶ms)?;
1484 let result = self
1485 .send_request::<_, ()>(
1486 Method::POST,
1487 "/v5/account/no-convert-repay",
1488 None,
1489 Some(body),
1490 true,
1491 )
1492 .await;
1493
1494 if let Err(ref e) = result
1495 && let Ok(params_json) = serde_json::to_string(¶ms)
1496 {
1497 log::error!("Repay request failed with params {params_json}: {e}");
1498 }
1499
1500 result
1501 }
1502
1503 pub async fn repay(
1521 &self,
1522 coin: Option<&str>,
1523 amount: Option<&str>,
1524 ) -> Result<BybitRepayResponse, BybitHttpError> {
1525 let mut builder = BybitRepayParamsBuilder::default();
1526
1527 if let Some(coin) = coin {
1528 builder.coin(coin.to_string());
1529 }
1530
1531 if let Some(amt) = amount {
1532 builder.amount(amt.to_string());
1533 }
1534
1535 let params = builder.build().expect("Failed to build BybitRepayParams");
1536
1537 if let Ok(params_json) = serde_json::to_string(¶ms) {
1538 log::debug!("Repay request params: {params_json}");
1539 }
1540
1541 let body = serde_json::to_vec(¶ms)?;
1542 let result = self
1543 .send_request::<_, ()>(Method::POST, "/v5/account/repay", None, Some(body), true)
1544 .await;
1545
1546 if let Err(ref e) = result
1547 && let Ok(params_json) = serde_json::to_string(¶ms)
1548 {
1549 log::error!("Repay request failed with params {params_json}: {e}");
1550 }
1551
1552 result
1553 }
1554
1555 pub async fn get_tickers<T: DeserializeOwned + BybitResponseCheck>(
1565 &self,
1566 params: &BybitTickersParams,
1567 ) -> Result<T, BybitHttpError> {
1568 self.send_request(Method::GET, "/v5/market/tickers", Some(params), None, false)
1569 .await
1570 }
1571
1572 pub async fn get_trade_history(
1582 &self,
1583 params: &BybitTradeHistoryParams,
1584 ) -> Result<BybitTradeHistoryResponse, BybitHttpError> {
1585 self.send_request(Method::GET, BYBIT_EXECUTION_LIST, Some(params), None, true)
1586 .await
1587 }
1588
1589 pub async fn get_positions(
1602 &self,
1603 params: &BybitPositionListParams,
1604 ) -> Result<BybitPositionListResponse, BybitHttpError> {
1605 self.send_request(Method::GET, BYBIT_POSITION_LIST, Some(params), None, true)
1606 .await
1607 }
1608
1609 #[must_use]
1611 pub fn base_url(&self) -> &str {
1612 &self.base_url
1613 }
1614
1615 #[must_use]
1617 pub fn recv_window_ms(&self) -> u64 {
1618 self.recv_window_ms
1619 }
1620
1621 #[must_use]
1623 pub fn credential(&self) -> Option<&Credential> {
1624 self.credential.as_ref()
1625 }
1626}
1627
1628#[cfg_attr(
1630 feature = "python",
1631 pyo3::pyclass(module = "nautilus_trader.adapters.bybit", from_py_object)
1632)]
1633#[cfg_attr(
1634 feature = "python",
1635 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.bybit")
1636)]
1637pub struct BybitHttpClient {
1642 pub(crate) inner: Arc<BybitRawHttpClient>,
1643 pub(crate) instruments_cache: Arc<AtomicMap<Ustr, InstrumentAny>>,
1644 clock: &'static AtomicTime,
1645 cache_initialized: Arc<AtomicBool>,
1646 use_spot_position_reports: Arc<AtomicBool>,
1647}
1648
1649impl Clone for BybitHttpClient {
1650 fn clone(&self) -> Self {
1651 Self {
1652 inner: self.inner.clone(),
1653 instruments_cache: self.instruments_cache.clone(),
1654 cache_initialized: self.cache_initialized.clone(),
1655 use_spot_position_reports: self.use_spot_position_reports.clone(),
1656 clock: self.clock,
1657 }
1658 }
1659}
1660
1661impl Default for BybitHttpClient {
1662 fn default() -> Self {
1663 Self::new(None, 60, 3, 1000, 10_000, DEFAULT_RECV_WINDOW_MS, None)
1664 .expect("Failed to create default BybitHttpClient")
1665 }
1666}
1667
1668impl Debug for BybitHttpClient {
1669 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1670 f.debug_struct(stringify!(BybitHttpClient))
1671 .field("inner", &self.inner)
1672 .finish()
1673 }
1674}
1675
1676impl BybitHttpClient {
1677 pub fn new(
1683 base_url: Option<String>,
1684 timeout_secs: u64,
1685 max_retries: u32,
1686 retry_delay_ms: u64,
1687 retry_delay_max_ms: u64,
1688 recv_window_ms: u64,
1689 proxy_url: Option<String>,
1690 ) -> Result<Self, BybitHttpError> {
1691 Ok(Self {
1692 inner: Arc::new(BybitRawHttpClient::new(
1693 base_url,
1694 timeout_secs,
1695 max_retries,
1696 retry_delay_ms,
1697 retry_delay_max_ms,
1698 recv_window_ms,
1699 proxy_url,
1700 )?),
1701 instruments_cache: Arc::new(AtomicMap::new()),
1702 cache_initialized: Arc::new(AtomicBool::new(false)),
1703 use_spot_position_reports: Arc::new(AtomicBool::new(false)),
1704 clock: get_atomic_clock_realtime(),
1705 })
1706 }
1707
1708 #[expect(clippy::too_many_arguments)]
1714 pub fn with_credentials(
1715 api_key: String,
1716 api_secret: String,
1717 base_url: Option<String>,
1718 timeout_secs: u64,
1719 max_retries: u32,
1720 retry_delay_ms: u64,
1721 retry_delay_max_ms: u64,
1722 recv_window_ms: u64,
1723 proxy_url: Option<String>,
1724 ) -> Result<Self, BybitHttpError> {
1725 Ok(Self {
1726 inner: Arc::new(BybitRawHttpClient::with_credentials(
1727 api_key,
1728 api_secret,
1729 base_url,
1730 timeout_secs,
1731 max_retries,
1732 retry_delay_ms,
1733 retry_delay_max_ms,
1734 recv_window_ms,
1735 proxy_url,
1736 )?),
1737 instruments_cache: Arc::new(AtomicMap::new()),
1738 cache_initialized: Arc::new(AtomicBool::new(false)),
1739 use_spot_position_reports: Arc::new(AtomicBool::new(false)),
1740 clock: get_atomic_clock_realtime(),
1741 })
1742 }
1743
1744 #[expect(clippy::too_many_arguments)]
1757 pub fn new_with_env(
1758 api_key: Option<String>,
1759 api_secret: Option<String>,
1760 base_url: Option<String>,
1761 demo: bool,
1762 testnet: bool,
1763 timeout_secs: u64,
1764 max_retries: u32,
1765 retry_delay_ms: u64,
1766 retry_delay_max_ms: u64,
1767 recv_window_ms: u64,
1768 proxy_url: Option<String>,
1769 ) -> Result<Self, BybitHttpError> {
1770 Ok(Self {
1771 inner: Arc::new(BybitRawHttpClient::new_with_env(
1772 api_key,
1773 api_secret,
1774 base_url,
1775 demo,
1776 testnet,
1777 timeout_secs,
1778 max_retries,
1779 retry_delay_ms,
1780 retry_delay_max_ms,
1781 recv_window_ms,
1782 proxy_url,
1783 )?),
1784 instruments_cache: Arc::new(AtomicMap::new()),
1785 cache_initialized: Arc::new(AtomicBool::new(false)),
1786 use_spot_position_reports: Arc::new(AtomicBool::new(false)),
1787 clock: get_atomic_clock_realtime(),
1788 })
1789 }
1790
1791 #[must_use]
1792 pub fn base_url(&self) -> &str {
1793 self.inner.base_url()
1794 }
1795
1796 #[must_use]
1797 pub fn recv_window_ms(&self) -> u64 {
1798 self.inner.recv_window_ms()
1799 }
1800
1801 #[must_use]
1802 pub fn credential(&self) -> Option<&Credential> {
1803 self.inner.credential()
1804 }
1805
1806 pub fn set_use_spot_position_reports(&self, use_spot_position_reports: bool) {
1807 self.use_spot_position_reports
1808 .store(use_spot_position_reports, Ordering::Relaxed);
1809 }
1810
1811 pub fn cancel_all_requests(&self) {
1812 self.inner.cancel_all_requests();
1813 }
1814
1815 pub fn reset_cancellation_token(&self) {
1816 self.inner.reset_cancellation_token();
1817 }
1818
1819 pub fn cancellation_token(&self) -> CancellationToken {
1820 self.inner.cancellation_token()
1821 }
1822
1823 pub fn cache_instrument(&self, instrument: InstrumentAny) {
1825 self.instruments_cache
1826 .insert(instrument.symbol().inner(), instrument);
1827 self.cache_initialized.store(true, Ordering::Release);
1828 }
1829
1830 pub fn cache_instruments(&self, instruments: &[InstrumentAny]) {
1832 self.instruments_cache.rcu(|m| {
1833 for instrument in instruments {
1834 m.insert(instrument.symbol().inner(), instrument.clone());
1835 }
1836 });
1837 self.cache_initialized.store(true, Ordering::Release);
1838 }
1839
1840 pub fn get_instrument(&self, symbol: &Ustr) -> Option<InstrumentAny> {
1841 self.instruments_cache.get_cloned(symbol)
1842 }
1843
1844 fn instrument_from_cache(&self, symbol: &Symbol) -> anyhow::Result<InstrumentAny> {
1845 self.get_instrument(&symbol.inner()).ok_or_else(|| {
1846 anyhow::anyhow!(
1847 "Instrument {symbol} not found in cache, ensure instruments loaded first"
1848 )
1849 })
1850 }
1851
1852 #[must_use]
1853 fn generate_ts_init(&self) -> UnixNanos {
1854 self.clock.get_time_ns()
1855 }
1856
1857 pub async fn get_server_time(&self) -> Result<BybitServerTimeResponse, BybitHttpError> {
1869 self.inner.get_server_time().await
1870 }
1871
1872 pub async fn get_instruments<T: DeserializeOwned + BybitResponseCheck>(
1884 &self,
1885 params: &BybitInstrumentsInfoParams,
1886 ) -> Result<T, BybitHttpError> {
1887 self.inner.get_instruments(params).await
1888 }
1889
1890 pub async fn get_instruments_spot(
1902 &self,
1903 params: &BybitInstrumentsInfoParams,
1904 ) -> Result<BybitInstrumentSpotResponse, BybitHttpError> {
1905 self.inner.get_instruments_spot(params).await
1906 }
1907
1908 pub async fn get_instruments_linear(
1920 &self,
1921 params: &BybitInstrumentsInfoParams,
1922 ) -> Result<BybitInstrumentLinearResponse, BybitHttpError> {
1923 self.inner.get_instruments_linear(params).await
1924 }
1925
1926 pub async fn get_instruments_inverse(
1938 &self,
1939 params: &BybitInstrumentsInfoParams,
1940 ) -> Result<BybitInstrumentInverseResponse, BybitHttpError> {
1941 self.inner.get_instruments_inverse(params).await
1942 }
1943
1944 pub async fn get_instruments_option(
1956 &self,
1957 params: &BybitInstrumentsInfoParams,
1958 ) -> Result<BybitInstrumentOptionResponse, BybitHttpError> {
1959 self.inner.get_instruments_option(params).await
1960 }
1961
1962 pub async fn get_klines(
1974 &self,
1975 params: &BybitKlinesParams,
1976 ) -> Result<BybitKlinesResponse, BybitHttpError> {
1977 self.inner.get_klines(params).await
1978 }
1979
1980 pub async fn get_recent_trades(
1992 &self,
1993 params: &BybitTradesParams,
1994 ) -> Result<BybitTradesResponse, BybitHttpError> {
1995 self.inner.get_recent_trades(params).await
1996 }
1997
1998 #[expect(clippy::too_many_arguments)]
2010 pub async fn get_open_orders(
2011 &self,
2012 category: BybitProductType,
2013 symbol: Option<String>,
2014 base_coin: Option<String>,
2015 settle_coin: Option<String>,
2016 order_id: Option<String>,
2017 order_link_id: Option<String>,
2018 open_only: Option<BybitOpenOnly>,
2019 order_filter: Option<BybitOrderFilter>,
2020 limit: Option<u32>,
2021 cursor: Option<String>,
2022 ) -> Result<BybitOpenOrdersResponse, BybitHttpError> {
2023 self.inner
2024 .get_open_orders(
2025 category,
2026 symbol,
2027 base_coin,
2028 settle_coin,
2029 order_id,
2030 order_link_id,
2031 open_only,
2032 order_filter,
2033 limit,
2034 cursor,
2035 )
2036 .await
2037 }
2038
2039 pub async fn place_order(
2051 &self,
2052 request: &serde_json::Value,
2053 ) -> Result<BybitPlaceOrderResponse, BybitHttpError> {
2054 self.inner.place_order(request).await
2055 }
2056
2057 pub async fn get_wallet_balance(
2069 &self,
2070 params: &BybitWalletBalanceParams,
2071 ) -> Result<BybitWalletBalanceResponse, BybitHttpError> {
2072 self.inner.get_wallet_balance(params).await
2073 }
2074
2075 pub async fn get_account_info(&self) -> Result<BybitAccountInfoResponse, BybitHttpError> {
2087 self.inner.get_account_info().await
2088 }
2089
2090 pub async fn get_account_details(&self) -> Result<BybitAccountDetailsResponse, BybitHttpError> {
2102 self.inner.get_account_details().await
2103 }
2104
2105 pub async fn update_sub_api_key(
2117 &self,
2118 params: &BybitUpdateSubApiParams,
2119 ) -> Result<BybitUpdateSubApiResponse, BybitHttpError> {
2120 self.inner.update_sub_api_key(params).await
2121 }
2122
2123 pub async fn update_master_api_key(
2135 &self,
2136 params: &BybitUpdateMasterApiParams,
2137 ) -> Result<BybitUpdateMasterApiResponse, BybitHttpError> {
2138 self.inner.update_master_api_key(params).await
2139 }
2140
2141 pub async fn get_sub_members(&self) -> Result<BybitSubMembersResponse, BybitHttpError> {
2153 self.inner.get_sub_members().await
2154 }
2155
2156 pub async fn get_sub_members_paged(
2168 &self,
2169 params: &BybitSubMembersPageParams,
2170 ) -> Result<BybitSubMembersPagedResponse, BybitHttpError> {
2171 self.inner.get_sub_members_paged(params).await
2172 }
2173
2174 pub async fn get_escrow_sub_members(
2186 &self,
2187 params: &BybitSubMembersPageParams,
2188 ) -> Result<BybitEscrowSubMembersResponse, BybitHttpError> {
2189 self.inner.get_escrow_sub_members(params).await
2190 }
2191
2192 pub async fn get_sub_api_keys(
2204 &self,
2205 params: &BybitSubApiKeysParams,
2206 ) -> Result<BybitSubApiKeysResponse, BybitHttpError> {
2207 self.inner.get_sub_api_keys(params).await
2208 }
2209
2210 pub async fn get_positions(
2223 &self,
2224 params: &BybitPositionListParams,
2225 ) -> Result<BybitPositionListResponse, BybitHttpError> {
2226 self.inner.get_positions(params).await
2227 }
2228
2229 pub async fn get_fee_rate(
2242 &self,
2243 params: &BybitFeeRateParams,
2244 ) -> Result<BybitFeeRateResponse, BybitHttpError> {
2245 self.inner.get_fee_rate(params).await
2246 }
2247
2248 pub async fn set_margin_mode(
2261 &self,
2262 margin_mode: BybitMarginMode,
2263 ) -> Result<BybitSetMarginModeResponse, BybitHttpError> {
2264 self.inner.set_margin_mode(margin_mode).await
2265 }
2266
2267 pub async fn set_leverage(
2280 &self,
2281 product_type: BybitProductType,
2282 symbol: &str,
2283 buy_leverage: &str,
2284 sell_leverage: &str,
2285 ) -> Result<BybitSetLeverageResponse, BybitHttpError> {
2286 self.inner
2287 .set_leverage(product_type, symbol, buy_leverage, sell_leverage)
2288 .await
2289 }
2290
2291 pub async fn switch_mode(
2304 &self,
2305 product_type: BybitProductType,
2306 mode: BybitPositionMode,
2307 symbol: Option<String>,
2308 coin: Option<String>,
2309 ) -> Result<BybitSwitchModeResponse, BybitHttpError> {
2310 self.inner
2311 .switch_mode(product_type, mode, symbol, coin)
2312 .await
2313 }
2314
2315 pub async fn set_trading_stop(
2328 &self,
2329 params: &BybitSetTradingStopParams,
2330 ) -> Result<BybitSetTradingStopResponse, BybitHttpError> {
2331 self.inner.set_trading_stop(params).await
2332 }
2333
2334 pub async fn get_spot_borrow_amount(&self, coin: &str) -> anyhow::Result<Decimal> {
2349 let params = BybitWalletBalanceParams {
2350 account_type: BybitAccountType::Unified,
2351 coin: Some(coin.to_string()),
2352 };
2353
2354 let response = self.inner.get_wallet_balance(¶ms).await?;
2355
2356 let borrow_amount = response
2357 .result
2358 .list
2359 .first()
2360 .and_then(|wallet| wallet.coin.iter().find(|c| c.coin == coin))
2361 .map_or(Decimal::ZERO, |balance| balance.spot_borrow);
2362
2363 Ok(borrow_amount)
2364 }
2365
2366 pub async fn borrow_spot(
2382 &self,
2383 coin: &str,
2384 amount: Quantity,
2385 ) -> anyhow::Result<BybitBorrowResponse> {
2386 let amount_str = amount.to_string();
2387 self.inner
2388 .borrow(coin, &amount_str)
2389 .await
2390 .map_err(|e| anyhow::anyhow!("Failed to borrow {amount} {coin}: {e}"))
2391 }
2392
2393 pub async fn repay_spot_borrow(
2410 &self,
2411 coin: &str,
2412 amount: Option<Quantity>,
2413 ) -> anyhow::Result<BybitNoConvertRepayResponse> {
2414 let amount_str = amount.as_ref().map(|q| q.to_string());
2415 let response = self
2416 .inner
2417 .no_convert_repay(coin, amount_str.as_deref())
2418 .await
2419 .map_err(|e| anyhow::anyhow!("Failed to repay spot borrow for {coin}: {e}"))?;
2420 Self::ensure_repay_accepted(coin, response.result.result_status)?;
2421 Ok(response)
2422 }
2423
2424 pub async fn repay_spot_borrow_with_conversion(
2442 &self,
2443 coin: &str,
2444 amount: Option<Quantity>,
2445 ) -> anyhow::Result<BybitRepayResponse> {
2446 let amount_str = amount.as_ref().map(|q| q.to_string());
2447 let response = self
2448 .inner
2449 .repay(Some(coin), amount_str.as_deref())
2450 .await
2451 .map_err(|e| {
2452 anyhow::anyhow!("Failed to repay spot borrow (with conversion) for {coin}: {e}")
2453 })?;
2454 Self::ensure_repay_accepted(coin, response.result.result_status)?;
2455 Ok(response)
2456 }
2457
2458 fn ensure_repay_accepted(coin: &str, status: BybitRepayStatus) -> anyhow::Result<()> {
2459 anyhow::ensure!(
2460 status != BybitRepayStatus::Failed,
2461 "Bybit repay for {coin} returned result status {status}"
2462 );
2463 Ok(())
2464 }
2465
2466 async fn generate_spot_position_reports_from_wallet(
2474 &self,
2475 account_id: AccountId,
2476 instrument_id: InstrumentId,
2477 ) -> anyhow::Result<Vec<PositionStatusReport>> {
2478 let params = BybitWalletBalanceParams {
2479 account_type: BybitAccountType::Unified,
2480 coin: None,
2481 };
2482
2483 let response = self.inner.get_wallet_balance(¶ms).await?;
2484 let ts_init = self.generate_ts_init();
2485
2486 let mut wallet_by_coin: HashMap<Ustr, Decimal> = HashMap::new();
2487
2488 for wallet in &response.result.list {
2489 for coin_balance in &wallet.coin {
2490 let balance = coin_balance.wallet_balance - coin_balance.spot_borrow;
2491 *wallet_by_coin
2492 .entry(coin_balance.coin)
2493 .or_insert(Decimal::ZERO) += balance;
2494 }
2495 }
2496
2497 let mut reports = Vec::new();
2498
2499 if let Some(instrument) = self
2500 .instruments_cache
2501 .get_cloned(&instrument_id.symbol.inner())
2502 {
2503 let base_currency = instrument
2504 .base_currency()
2505 .expect("SPOT instrument should have base currency");
2506 let coin = base_currency.code;
2507 let wallet_balance = wallet_by_coin.get(&coin).copied().unwrap_or(Decimal::ZERO);
2508
2509 let side = if wallet_balance > Decimal::ZERO {
2510 PositionSide::Long
2511 } else if wallet_balance < Decimal::ZERO {
2512 PositionSide::Short
2513 } else {
2514 PositionSide::Flat
2515 };
2516
2517 let abs_balance = wallet_balance.abs();
2518 let quantity = Quantity::from_decimal_dp(abs_balance, instrument.size_precision())?;
2519
2520 let report = PositionStatusReport::new(
2521 account_id,
2522 instrument_id,
2523 side,
2524 quantity,
2525 ts_init,
2526 ts_init,
2527 None,
2528 None,
2529 None,
2530 );
2531
2532 reports.push(report);
2533 }
2534
2535 Ok(reports)
2536 }
2537
2538 #[expect(clippy::too_many_arguments)]
2549 pub async fn submit_order(
2550 &self,
2551 account_id: AccountId,
2552 product_type: BybitProductType,
2553 instrument_id: InstrumentId,
2554 client_order_id: ClientOrderId,
2555 order_side: OrderSide,
2556 order_type: OrderType,
2557 quantity: Quantity,
2558 time_in_force: Option<TimeInForce>,
2559 price: Option<Price>,
2560 trigger_price: Option<Price>,
2561 post_only: Option<bool>,
2562 reduce_only: bool,
2563 is_quote_quantity: bool,
2564 is_leverage: bool,
2565 position_idx: Option<BybitPositionIdx>,
2566 bbo_side_type: Option<BybitBboSideType>,
2567 bbo_level: Option<String>,
2568 smp_type: Option<BybitOrderSmpType>,
2569 native_tp_sl: Option<&BybitNativeTpSlParams>,
2570 ) -> anyhow::Result<OrderStatusReport> {
2571 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
2572 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2573
2574 let bybit_side = match order_side {
2575 OrderSide::Buy => BybitOrderSide::Buy,
2576 OrderSide::Sell => BybitOrderSide::Sell,
2577 };
2578
2579 let (bybit_order_type, is_stop_order) = match order_type {
2581 OrderType::Market => (BybitOrderType::Market, false),
2582 OrderType::Limit => (BybitOrderType::Limit, false),
2583 OrderType::StopMarket | OrderType::MarketIfTouched => (BybitOrderType::Market, true),
2584 OrderType::StopLimit | OrderType::LimitIfTouched => (BybitOrderType::Limit, true),
2585 _ => anyhow::bail!("Unsupported order type: {order_type:?}"),
2586 };
2587
2588 let bybit_tif = map_time_in_force(bybit_order_type, time_in_force, post_only)
2589 .map_err(|tif| anyhow::anyhow!("Unsupported time in force: {tif:?}"))?;
2590 let market_unit = spot_market_unit(product_type, bybit_order_type, is_quote_quantity);
2591 let trigger_dir = trigger_direction(order_type, order_side, is_stop_order);
2592
2593 let mut order_entry = BybitBatchPlaceOrderEntryBuilder::default();
2594 order_entry.symbol(bybit_symbol.raw_symbol().to_string());
2595 order_entry.side(bybit_side);
2596 order_entry.order_type(bybit_order_type);
2597 order_entry.qty(quantity.to_string());
2598 order_entry.time_in_force(bybit_tif);
2599 order_entry.order_link_id(client_order_id.to_string());
2600 order_entry.market_unit(market_unit);
2601 order_entry.trigger_direction(trigger_dir);
2602
2603 if bbo_side_type.is_none()
2604 && let Some(price) = price
2605 {
2606 order_entry.price(Some(price.to_string()));
2607 }
2608
2609 if let Some(trigger_price) = trigger_price {
2610 order_entry.trigger_price(Some(trigger_price.to_string()));
2611 }
2612
2613 if reduce_only {
2614 order_entry.reduce_only(Some(true));
2615 }
2616
2617 order_entry.is_leverage(spot_leverage(product_type, is_leverage));
2618
2619 if let Some(idx) = position_idx {
2620 order_entry.position_idx(Some(idx));
2621 }
2622
2623 order_entry.bbo_side_type(bbo_side_type);
2624 order_entry.bbo_level(bbo_level);
2625 order_entry.smp_type(smp_type);
2626
2627 if let Some(tp_sl) = native_tp_sl {
2628 if let Some(ref tp) = tp_sl.take_profit {
2629 order_entry.take_profit(Some(tp.clone()));
2630 }
2631
2632 if let Some(ref sl) = tp_sl.stop_loss {
2633 order_entry.stop_loss(Some(sl.clone()));
2634 }
2635
2636 if let Some(tp_trigger) = tp_sl.tp_trigger_by {
2637 order_entry.tp_trigger_by(Some(tp_trigger));
2638 }
2639
2640 if let Some(sl_trigger) = tp_sl.sl_trigger_by {
2641 order_entry.sl_trigger_by(Some(sl_trigger));
2642 }
2643
2644 if let Some(tp_ot) = tp_sl.tp_order_type {
2645 order_entry.tp_order_type(Some(tp_ot));
2646 }
2647
2648 if let Some(sl_ot) = tp_sl.sl_order_type {
2649 order_entry.sl_order_type(Some(sl_ot));
2650 }
2651
2652 if let Some(ref tp_lp) = tp_sl.tp_limit_price {
2653 order_entry.tp_limit_price(Some(tp_lp.clone()));
2654 }
2655
2656 if let Some(ref sl_lp) = tp_sl.sl_limit_price {
2657 order_entry.sl_limit_price(Some(sl_lp.clone()));
2658 }
2659
2660 let mode = tp_sl.tpsl_mode.or_else(|| {
2663 (tp_sl.take_profit.is_some() || tp_sl.stop_loss.is_some())
2664 .then_some(BybitTpSlMode::Full)
2665 });
2666
2667 if let Some(m) = mode {
2668 order_entry.tpsl_mode(Some(m));
2669 }
2670
2671 if let Some(close) = tp_sl.close_on_trigger {
2672 order_entry.close_on_trigger(Some(close));
2673 }
2674
2675 if let Some(ref iv) = tp_sl.order_iv {
2676 order_entry.order_iv(Some(iv.clone()));
2677 }
2678
2679 if let Some(mmp) = tp_sl.mmp {
2680 order_entry.mmp(Some(mmp));
2681 }
2682 }
2683
2684 let order_entry = order_entry.build().build_anyhow()?;
2685
2686 let mut params = BybitPlaceOrderParamsBuilder::default();
2687 params.category(product_type);
2688 params.order(order_entry);
2689
2690 let params = params.build().build_anyhow()?;
2691
2692 let body = serde_json::to_value(¶ms)?;
2693 let response = self.inner.place_order(&body).await?;
2694
2695 let order_id = response
2696 .result
2697 .order_id
2698 .ok_or(BybitSubmitOrderError::MissingOrderId)?;
2699
2700 let order = self
2701 .query_order_by_id(
2702 product_type,
2703 order_id.as_str(),
2704 BYBIT_ORDER_REALTIME,
2705 "after submission",
2706 )
2707 .await
2708 .map_err(|source| BybitSubmitOrderError::PostSubmitLookup { source })?;
2709
2710 let is_rejection = order.order_status == crate::common::enums::BybitOrderStatus::Rejected
2715 || (order.order_status == crate::common::enums::BybitOrderStatus::Canceled
2716 && bybit_rejection_due_post_only(order.reject_reason.as_str()));
2717 if is_rejection && (order.cum_exec_qty.as_str() == "0" || order.cum_exec_qty.is_empty()) {
2718 return Err(BybitSubmitOrderError::Rejected {
2719 reason: order.reject_reason.to_string(),
2720 }
2721 .into());
2722 }
2723
2724 let ts_init = self.generate_ts_init();
2725
2726 parse_order_status_report(&order, &instrument, account_id, ts_init)
2727 }
2728
2729 pub async fn cancel_order(
2739 &self,
2740 account_id: AccountId,
2741 product_type: BybitProductType,
2742 instrument_id: InstrumentId,
2743 client_order_id: Option<ClientOrderId>,
2744 venue_order_id: Option<VenueOrderId>,
2745 ) -> anyhow::Result<OrderStatusReport> {
2746 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
2747 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2748
2749 let mut cancel_entry = BybitBatchCancelOrderEntryBuilder::default();
2750 cancel_entry.symbol(bybit_symbol.raw_symbol().to_string());
2751
2752 if let Some(venue_order_id) = venue_order_id {
2753 cancel_entry.order_id(venue_order_id.to_string());
2754 } else if let Some(client_order_id) = client_order_id {
2755 cancel_entry.order_link_id(client_order_id.to_string());
2756 } else {
2757 anyhow::bail!("Either client_order_id or venue_order_id must be provided");
2758 }
2759
2760 let cancel_entry = cancel_entry.build().build_anyhow()?;
2761
2762 let mut params = BybitCancelOrderParamsBuilder::default();
2763 params.category(product_type);
2764 params.order(cancel_entry);
2765
2766 let params = params.build().build_anyhow()?;
2767 let body = serde_json::to_vec(¶ms)?;
2768
2769 let response: BybitPlaceOrderResponse = self
2770 .inner
2771 .send_request::<_, ()>(Method::POST, "/v5/order/cancel", None, Some(body), true)
2772 .await?;
2773
2774 let order_id = response
2775 .result
2776 .order_id
2777 .ok_or(BybitCancelOrderError::MissingOrderId)?;
2778
2779 let order = self
2780 .query_order_by_id(
2781 product_type,
2782 order_id.as_str(),
2783 BYBIT_ORDER_HISTORY,
2784 "after cancellation",
2785 )
2786 .await
2787 .map_err(|source| BybitCancelOrderError::PostCancelLookup { source })?;
2788
2789 let ts_init = self.generate_ts_init();
2790
2791 parse_order_status_report(&order, &instrument, account_id, ts_init)
2792 }
2793
2794 pub async fn batch_cancel_orders(
2804 &self,
2805 account_id: AccountId,
2806 product_type: BybitProductType,
2807 instrument_ids: Vec<InstrumentId>,
2808 client_order_ids: Vec<Option<ClientOrderId>>,
2809 venue_order_ids: Vec<Option<VenueOrderId>>,
2810 ) -> anyhow::Result<Vec<OrderStatusReport>> {
2811 if instrument_ids.len() != client_order_ids.len()
2812 || instrument_ids.len() != venue_order_ids.len()
2813 {
2814 anyhow::bail!(
2815 "instrument_ids, client_order_ids, and venue_order_ids must have the same length"
2816 );
2817 }
2818
2819 if instrument_ids.is_empty() {
2820 return Ok(Vec::new());
2821 }
2822
2823 let call_limit = batch_call_limit(product_type);
2824 if instrument_ids.len() > call_limit {
2825 anyhow::bail!(
2826 "Batch cancel limit is {call_limit} orders for {}",
2827 product_type.as_str()
2828 );
2829 }
2830
2831 let mut cancel_entries = Vec::new();
2832
2833 for ((instrument_id, client_order_id), venue_order_id) in instrument_ids
2834 .iter()
2835 .zip(client_order_ids.iter())
2836 .zip(venue_order_ids.iter())
2837 {
2838 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2839 let mut cancel_entry = BybitBatchCancelOrderEntryBuilder::default();
2840 cancel_entry.symbol(bybit_symbol.raw_symbol().to_string());
2841
2842 if let Some(venue_order_id) = venue_order_id {
2843 cancel_entry.order_id(venue_order_id.to_string());
2844 } else if let Some(client_order_id) = client_order_id {
2845 cancel_entry.order_link_id(client_order_id.to_string());
2846 } else {
2847 anyhow::bail!(
2848 "Either client_order_id or venue_order_id must be provided for each order"
2849 );
2850 }
2851
2852 cancel_entries.push(cancel_entry.build().build_anyhow()?);
2853 }
2854
2855 let chunk_limit = batch_endpoint_limit(product_type).min(batch_send_limit(product_type));
2856 for chunk in cancel_entries.chunks(chunk_limit) {
2857 let mut params = BybitBatchCancelOrderParamsBuilder::default();
2858 params.category(product_type);
2859 params.request(chunk.to_vec());
2860
2861 let params = params.build().build_anyhow()?;
2862 let body = serde_json::to_vec(¶ms)?;
2863
2864 let _response: BybitPlaceOrderResponse = self
2865 .inner
2866 .send_request::<_, ()>(
2867 Method::POST,
2868 "/v5/order/cancel-batch",
2869 None,
2870 Some(body),
2871 true,
2872 )
2873 .await?;
2874 }
2875
2876 let mut reports = Vec::new();
2878
2879 for (instrument_id, (client_order_id, venue_order_id)) in instrument_ids
2880 .iter()
2881 .zip(client_order_ids.iter().zip(venue_order_ids.iter()))
2882 {
2883 let Ok(instrument) = self.instrument_from_cache(&instrument_id.symbol) else {
2884 log::debug!(
2885 "Skipping cancelled order report for instrument not in cache: symbol={}",
2886 instrument_id.symbol
2887 );
2888 continue;
2889 };
2890
2891 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2892
2893 let mut query_params = BybitOpenOrdersParamsBuilder::default();
2894 query_params.category(product_type);
2895 query_params.symbol(bybit_symbol.raw_symbol().to_string());
2896
2897 if let Some(venue_order_id) = venue_order_id {
2898 query_params.order_id(venue_order_id.to_string());
2899 } else if let Some(client_order_id) = client_order_id {
2900 query_params.order_link_id(client_order_id.to_string());
2901 }
2902
2903 let query_params = query_params.build().build_anyhow()?;
2904 let order_response: BybitOrderHistoryResponse = self
2905 .inner
2906 .send_request(
2907 Method::GET,
2908 BYBIT_ORDER_HISTORY,
2909 Some(&query_params),
2910 None,
2911 true,
2912 )
2913 .await?;
2914
2915 if let Some(order) = order_response.result.list.into_iter().next() {
2916 let ts_init = self.generate_ts_init();
2917 let report = parse_order_status_report(&order, &instrument, account_id, ts_init)?;
2918 reports.push(report);
2919 }
2920 }
2921
2922 Ok(reports)
2923 }
2924
2925 pub async fn cancel_all_orders(
2934 &self,
2935 account_id: AccountId,
2936 product_type: BybitProductType,
2937 instrument_id: InstrumentId,
2938 ) -> anyhow::Result<Vec<OrderStatusReport>> {
2939 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
2940 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2941
2942 let mut params = BybitCancelAllOrdersParamsBuilder::default();
2943 params.category(product_type);
2944 params.symbol(bybit_symbol.raw_symbol().to_string());
2945
2946 let params = params.build().build_anyhow()?;
2947 let body = serde_json::to_vec(¶ms)?;
2948
2949 let _response: crate::common::models::BybitListResponse<serde_json::Value> = self
2950 .inner
2951 .send_request::<_, ()>(Method::POST, "/v5/order/cancel-all", None, Some(body), true)
2952 .await?;
2953
2954 let mut query_params = BybitOrderHistoryParamsBuilder::default();
2956 query_params.category(product_type);
2957 query_params.symbol(bybit_symbol.raw_symbol().to_string());
2958 query_params.limit(50u32);
2959
2960 let query_params = query_params.build().build_anyhow()?;
2961 let order_response: BybitOrderHistoryResponse = self
2962 .inner
2963 .send_request(
2964 Method::GET,
2965 BYBIT_ORDER_HISTORY,
2966 Some(&query_params),
2967 None,
2968 true,
2969 )
2970 .await?;
2971
2972 let ts_init = self.generate_ts_init();
2973
2974 let mut reports = Vec::new();
2975
2976 for order in order_response.result.list {
2977 if let Ok(report) = parse_order_status_report(&order, &instrument, account_id, ts_init)
2978 {
2979 reports.push(report);
2980 }
2981 }
2982
2983 Ok(reports)
2984 }
2985
2986 #[expect(clippy::too_many_arguments)]
2997 pub async fn modify_order(
2998 &self,
2999 account_id: AccountId,
3000 product_type: BybitProductType,
3001 instrument_id: InstrumentId,
3002 client_order_id: Option<ClientOrderId>,
3003 venue_order_id: Option<VenueOrderId>,
3004 quantity: Option<Quantity>,
3005 price: Option<Price>,
3006 ) -> anyhow::Result<OrderStatusReport> {
3007 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
3008 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
3009
3010 let mut amend_entry = BybitBatchAmendOrderEntryBuilder::default();
3011 amend_entry.symbol(bybit_symbol.raw_symbol().to_string());
3012
3013 if let Some(venue_order_id) = venue_order_id {
3014 amend_entry.order_id(venue_order_id.to_string());
3015 } else if let Some(client_order_id) = client_order_id {
3016 amend_entry.order_link_id(client_order_id.to_string());
3017 } else {
3018 anyhow::bail!("Either client_order_id or venue_order_id must be provided");
3019 }
3020
3021 if let Some(quantity) = quantity {
3022 amend_entry.qty(Some(quantity.to_string()));
3023 }
3024
3025 if let Some(price) = price {
3026 amend_entry.price(Some(price.to_string()));
3027 }
3028
3029 let amend_entry = amend_entry.build().build_anyhow()?;
3030
3031 let mut params = BybitAmendOrderParamsBuilder::default();
3032 params.category(product_type);
3033 params.order(amend_entry);
3034
3035 let params = params.build().build_anyhow()?;
3036 let body = serde_json::to_vec(¶ms)?;
3037
3038 let response: BybitPlaceOrderResponse = self
3039 .inner
3040 .send_request::<_, ()>(Method::POST, "/v5/order/amend", None, Some(body), true)
3041 .await?;
3042
3043 let order_id = response
3044 .result
3045 .order_id
3046 .ok_or(BybitModifyOrderError::MissingOrderId)?;
3047
3048 let order = self
3049 .query_order_by_id(
3050 product_type,
3051 order_id.as_str(),
3052 BYBIT_ORDER_REALTIME,
3053 "after amendment",
3054 )
3055 .await
3056 .map_err(|source| BybitModifyOrderError::PostModifyLookup { source })?;
3057
3058 let ts_init = self.generate_ts_init();
3059
3060 parse_order_status_report(&order, &instrument, account_id, ts_init)
3061 }
3062
3063 pub async fn query_order(
3072 &self,
3073 account_id: AccountId,
3074 product_type: BybitProductType,
3075 instrument_id: InstrumentId,
3076 client_order_id: Option<ClientOrderId>,
3077 venue_order_id: Option<VenueOrderId>,
3078 ) -> anyhow::Result<Option<OrderStatusReport>> {
3079 log::debug!(
3080 "query_order: instrument_id={instrument_id}, client_order_id={client_order_id:?}, venue_order_id={venue_order_id:?}"
3081 );
3082
3083 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
3084
3085 let mut params = BybitOpenOrdersParamsBuilder::default();
3086 params.category(product_type);
3087 params.symbol(bybit_symbol.raw_symbol().to_string());
3089
3090 if let Some(venue_order_id) = venue_order_id {
3091 params.order_id(venue_order_id.to_string());
3092 } else if let Some(client_order_id) = client_order_id {
3093 params.order_link_id(client_order_id.to_string());
3094 } else {
3095 anyhow::bail!("Either client_order_id or venue_order_id must be provided");
3096 }
3097
3098 let params = params.build().build_anyhow()?;
3099 let mut response: BybitOpenOrdersResponse = self
3100 .inner
3101 .send_request(Method::GET, BYBIT_ORDER_REALTIME, Some(¶ms), None, true)
3102 .await?;
3103
3104 if response.result.list.is_empty() && product_type != BybitProductType::Option {
3106 log::debug!("Order not found in open orders, trying with StopOrder filter");
3107
3108 let mut stop_params = BybitOpenOrdersParamsBuilder::default();
3109 stop_params.category(product_type);
3110 stop_params.symbol(bybit_symbol.raw_symbol().to_string());
3111 stop_params.order_filter(BybitOrderFilter::StopOrder);
3112
3113 if let Some(venue_order_id) = venue_order_id {
3114 stop_params.order_id(venue_order_id.to_string());
3115 } else if let Some(client_order_id) = client_order_id {
3116 stop_params.order_link_id(client_order_id.to_string());
3117 }
3118
3119 let stop_params = stop_params.build().build_anyhow()?;
3120 response = self
3121 .inner
3122 .send_request(
3123 Method::GET,
3124 BYBIT_ORDER_REALTIME,
3125 Some(&stop_params),
3126 None,
3127 true,
3128 )
3129 .await?;
3130 }
3131
3132 if response.result.list.is_empty() {
3134 log::debug!("Order not found in open orders, checking order history");
3135
3136 let mut history_params = BybitOrderHistoryParamsBuilder::default();
3137 history_params.category(product_type);
3138 history_params.symbol(bybit_symbol.raw_symbol().to_string());
3139
3140 if let Some(venue_order_id) = venue_order_id {
3141 history_params.order_id(venue_order_id.to_string());
3142 } else if let Some(client_order_id) = client_order_id {
3143 history_params.order_link_id(client_order_id.to_string());
3144 }
3145
3146 let history_params = history_params.build().build_anyhow()?;
3147
3148 let mut history_response: BybitOrderHistoryResponse = self
3149 .inner
3150 .send_request(
3151 Method::GET,
3152 BYBIT_ORDER_HISTORY,
3153 Some(&history_params),
3154 None,
3155 true,
3156 )
3157 .await?;
3158
3159 if history_response.result.list.is_empty() && product_type == BybitProductType::Option {
3160 log::debug!("Option order not found in order history");
3161 return Ok(None);
3162 }
3163
3164 if history_response.result.list.is_empty() && product_type != BybitProductType::Option {
3166 log::debug!("Order not found in order history, trying with StopOrder filter");
3167
3168 let mut stop_history_params = BybitOrderHistoryParamsBuilder::default();
3169 stop_history_params.category(product_type);
3170 stop_history_params.symbol(bybit_symbol.raw_symbol().to_string());
3171 stop_history_params.order_filter(BybitOrderFilter::StopOrder);
3172
3173 if let Some(venue_order_id) = venue_order_id {
3174 stop_history_params.order_id(venue_order_id.to_string());
3175 } else if let Some(client_order_id) = client_order_id {
3176 stop_history_params.order_link_id(client_order_id.to_string());
3177 }
3178
3179 let stop_history_params = stop_history_params
3180 .build()
3181 .map_err(|e| anyhow::anyhow!(e))?;
3182
3183 history_response = self
3184 .inner
3185 .send_request(
3186 Method::GET,
3187 BYBIT_ORDER_HISTORY,
3188 Some(&stop_history_params),
3189 None,
3190 true,
3191 )
3192 .await?;
3193
3194 if history_response.result.list.is_empty() {
3195 log::debug!("Order not found in order history with StopOrder filter either");
3196 return Ok(None);
3197 }
3198 }
3199
3200 response.result.list = history_response.result.list;
3202 }
3203
3204 let order = &response.result.list[0];
3205 let ts_init = self.generate_ts_init();
3206
3207 log::debug!(
3208 "Query order response: symbol={}, order_id={}, order_link_id={}",
3209 order.symbol.as_str(),
3210 order.order_id.as_str(),
3211 order.order_link_id.as_str()
3212 );
3213
3214 let instrument = self
3215 .instrument_from_cache(&instrument_id.symbol)
3216 .map_err(|e| {
3217 log::error!(
3218 "Instrument cache miss for symbol '{}': {}",
3219 instrument_id.symbol.as_str(),
3220 e
3221 );
3222 anyhow::anyhow!(
3223 "Failed to query order {}: {}",
3224 client_order_id
3225 .as_ref()
3226 .map(|id| id.to_string())
3227 .or_else(|| venue_order_id.as_ref().map(|id| id.to_string()))
3228 .unwrap_or_else(|| "unknown".to_string()),
3229 e
3230 )
3231 })?;
3232
3233 log::debug!("Retrieved instrument from cache: id={}", instrument.id());
3234
3235 let report =
3236 parse_order_status_report(order, &instrument, account_id, ts_init).map_err(|e| {
3237 log::error!(
3238 "Failed to parse order status report for {}: {}",
3239 order.order_link_id.as_str(),
3240 e
3241 );
3242 e
3243 })?;
3244
3245 log::debug!(
3246 "Successfully created OrderStatusReport for {}",
3247 order.order_link_id.as_str()
3248 );
3249
3250 Ok(Some(report))
3251 }
3252
3253 async fn fetch_fee_map(
3254 &self,
3255 product_type: BybitProductType,
3256 base_coin: Option<Ustr>,
3257 ) -> anyhow::Result<AHashMap<Ustr, BybitFeeRate>> {
3258 let mut fee_params = BybitFeeRateParamsBuilder::default();
3259 fee_params.category(product_type);
3260 if let Some(bc) = base_coin {
3261 fee_params.base_coin(bc.to_string());
3262 }
3263 let Ok(params) = fee_params.build() else {
3264 return Ok(AHashMap::new());
3265 };
3266
3267 match self.inner.get_fee_rate(¶ms).await {
3268 Ok(response) => Ok(response
3269 .result
3270 .list
3271 .into_iter()
3272 .map(|f| (f.symbol, f))
3273 .collect()),
3274 Err(BybitHttpError::MissingCredentials) => {
3275 log::warn!("Missing credentials for fee rates, using defaults");
3276 Ok(AHashMap::new())
3277 }
3278 Err(BybitHttpError::BybitError {
3279 error_code,
3280 ref message,
3281 }) => {
3282 log::warn!(
3283 "{}",
3284 self.fee_rate_rejection_warning(product_type, error_code, message)
3285 );
3286 Ok(AHashMap::new())
3287 }
3288 Err(e) => Err(e.into()),
3289 }
3290 }
3291
3292 async fn fetch_option_fee_map(
3293 &self,
3294 base_coin: Option<Ustr>,
3295 ) -> anyhow::Result<AHashMap<Ustr, BybitFeeRate>> {
3296 let mut fee_params = BybitFeeRateParamsBuilder::default();
3297 fee_params.category(BybitProductType::Option);
3298 if let Some(bc) = base_coin {
3299 fee_params.base_coin(bc.to_string());
3300 }
3301 let Ok(params) = fee_params.build() else {
3302 return Ok(AHashMap::new());
3303 };
3304
3305 match self.inner.get_fee_rate(¶ms).await {
3306 Ok(response) => Ok(response
3307 .result
3308 .list
3309 .into_iter()
3310 .filter_map(|f| f.base_coin.map(|bc| (bc, f)))
3311 .collect()),
3312 Err(BybitHttpError::MissingCredentials) => {
3313 log::warn!("Missing credentials for option fee rates, using defaults");
3314 Ok(AHashMap::new())
3315 }
3316 Err(BybitHttpError::BybitError {
3317 error_code,
3318 ref message,
3319 }) => {
3320 let error_detail = Self::format_bybit_error_detail(error_code, message);
3321 log::warn!(
3322 "Option fee rate request rejected via /v5/account/fee-rate ({error_detail}), using defaults"
3323 );
3324 Ok(AHashMap::new())
3325 }
3326 Err(e) => {
3327 log::warn!("Option fee rate request failed ({e}), using defaults");
3328 Ok(AHashMap::new())
3329 }
3330 }
3331 }
3332
3333 fn fee_rate_rejection_warning(
3334 &self,
3335 product_type: BybitProductType,
3336 error_code: i32,
3337 message: &str,
3338 ) -> String {
3339 let product_type = product_type.as_ref().to_ascii_lowercase();
3340 let error_detail = Self::format_bybit_error_detail(error_code, message);
3341
3342 if self
3343 .base_url()
3344 .starts_with(bybit_http_base_url(BybitEnvironment::Demo))
3345 && matches!(product_type.as_str(), "linear" | "inverse")
3346 && error_code == 10001
3347 {
3348 format!(
3349 "Bybit demo rejected the {product_type} fee rate request via \
3350 /v5/account/fee-rate ({error_detail}); demo derivatives fee rates appear \
3351 unsupported, using defaults"
3352 )
3353 } else {
3354 format!(
3355 "Fee rate request rejected for {product_type} instruments via \
3356 /v5/account/fee-rate ({error_detail}), using defaults"
3357 )
3358 }
3359 }
3360
3361 fn format_bybit_error_detail(error_code: i32, message: &str) -> String {
3362 let message = message.trim();
3363 if message.is_empty() {
3364 format!("error {error_code}, no message")
3365 } else {
3366 format!("error {error_code}: {message}")
3367 }
3368 }
3369
3370 async fn paginate_instruments<D, F>(
3371 &self,
3372 product_type: BybitProductType,
3373 symbol: &Option<String>,
3374 base_coin: Option<Ustr>,
3375 mut parse: F,
3376 ) -> anyhow::Result<Vec<InstrumentAny>>
3377 where
3378 D: DeserializeOwned,
3379 BybitCursorListResponse<D>: BybitResponseCheck,
3380 F: FnMut(&D) -> Option<InstrumentAny>,
3381 {
3382 let mut instruments = Vec::new();
3383 let mut cursor: Option<String> = None;
3384 let mut cursor_walk = CursorWalk::default();
3385
3386 loop {
3387 let params = BybitInstrumentsInfoParams {
3388 category: product_type,
3389 symbol: symbol.clone(),
3390 status: None,
3391 base_coin: base_coin.map(|u| u.to_string()),
3392 limit: Some(1000),
3393 cursor: cursor.clone(),
3394 };
3395
3396 let response: BybitCursorListResponse<D> = self.inner.get_instruments(¶ms).await?;
3397
3398 for definition in &response.result.list {
3399 if let Some(instrument) = parse(definition) {
3400 instruments.push(instrument);
3401 }
3402 }
3403
3404 cursor = match cursor_walk
3408 .advance(BYBIT_INSTRUMENTS_INFO, response.result.next_page_cursor)
3409 {
3410 Ok(cursor) => cursor,
3411 Err(e) => {
3412 log::warn!(
3413 "{e}, keeping the {} instrument(s) already read",
3414 instruments.len()
3415 );
3416 break;
3417 }
3418 };
3419
3420 if cursor.is_none() {
3421 break;
3422 }
3423 }
3424
3425 Ok(instruments)
3426 }
3427
3428 pub async fn request_instrument_statuses(
3438 &self,
3439 product_type: BybitProductType,
3440 ) -> anyhow::Result<AHashMap<InstrumentId, MarketStatusAction>> {
3441 let mut statuses = AHashMap::new();
3442 let mut cursor: Option<String> = None;
3443 let mut cursor_walk = CursorWalk::default();
3444
3445 loop {
3446 let params = BybitInstrumentsInfoParams {
3447 category: product_type,
3448 symbol: None,
3449 status: None,
3450 base_coin: None,
3451 limit: Some(1000),
3452 cursor: cursor.clone(),
3453 };
3454
3455 match product_type {
3456 BybitProductType::Spot => {
3457 let response: BybitCursorListResponse<BybitInstrumentSpot> =
3458 self.inner.get_instruments(¶ms).await?;
3459
3460 for def in &response.result.list {
3461 let symbol = make_bybit_symbol(def.symbol, product_type);
3462 let id = InstrumentId::new(Symbol::from(symbol), *BYBIT_VENUE);
3463 statuses.insert(id, MarketStatusAction::from(def.status));
3464 }
3465 cursor = response.result.next_page_cursor;
3466 }
3467 BybitProductType::Linear => {
3468 let response: BybitCursorListResponse<BybitInstrumentLinear> =
3469 self.inner.get_instruments(¶ms).await?;
3470
3471 for def in &response.result.list {
3472 let symbol = make_bybit_symbol(def.symbol, product_type);
3473 let id = InstrumentId::new(Symbol::from(symbol), *BYBIT_VENUE);
3474 let status = MarketStatusAction::from(def.status);
3475 if status == MarketStatusAction::Trading
3476 && def.contract_type == BybitContractType::LinearPerpetual
3477 && def.delivery_time != "0"
3478 {
3479 statuses.insert(id, MarketStatusAction::PreClose);
3480 } else {
3481 statuses.insert(id, status);
3482 }
3483 }
3484 cursor = response.result.next_page_cursor;
3485 }
3486 BybitProductType::Inverse => {
3487 let response: BybitCursorListResponse<BybitInstrumentInverse> =
3488 self.inner.get_instruments(¶ms).await?;
3489
3490 for def in &response.result.list {
3491 let symbol = make_bybit_symbol(def.symbol, product_type);
3492 let id = InstrumentId::new(Symbol::from(symbol), *BYBIT_VENUE);
3493 let status = MarketStatusAction::from(def.status);
3494 if status == MarketStatusAction::Trading
3495 && def.contract_type == BybitContractType::InversePerpetual
3496 && def.delivery_time != "0"
3497 {
3498 statuses.insert(id, MarketStatusAction::PreClose);
3499 } else {
3500 statuses.insert(id, status);
3501 }
3502 }
3503 cursor = response.result.next_page_cursor;
3504 }
3505 BybitProductType::Option => {
3506 let response: BybitCursorListResponse<BybitInstrumentOption> =
3507 self.inner.get_instruments(¶ms).await?;
3508
3509 for def in &response.result.list {
3510 let symbol = make_bybit_symbol(def.symbol, product_type);
3511 let id = InstrumentId::new(Symbol::from(symbol), *BYBIT_VENUE);
3512 statuses.insert(id, MarketStatusAction::from(def.status));
3513 }
3514 cursor = response.result.next_page_cursor;
3515 }
3516 }
3517
3518 cursor = cursor_walk.advance(BYBIT_INSTRUMENTS_INFO, cursor)?;
3519 if cursor.is_none() {
3520 break;
3521 }
3522 }
3523
3524 Ok(statuses)
3525 }
3526
3527 pub async fn request_instruments(
3537 &self,
3538 product_type: BybitProductType,
3539 symbol: Option<String>,
3540 base_coin: Option<Ustr>,
3541 ) -> anyhow::Result<Vec<InstrumentAny>> {
3542 let ts_init = self.generate_ts_init();
3543
3544 let default_fee_rate = |symbol: Ustr| BybitFeeRate {
3545 symbol,
3546 taker_fee_rate: "0.001".to_string(),
3547 maker_fee_rate: "0.001".to_string(),
3548 base_coin: None,
3549 };
3550
3551 let instruments = match product_type {
3552 BybitProductType::Spot => {
3553 let fee_map = self.fetch_fee_map(product_type, base_coin).await?;
3554 self.paginate_instruments::<BybitInstrumentSpot, _>(
3555 product_type,
3556 &symbol,
3557 base_coin,
3558 |def| {
3559 let fee = fee_map
3560 .get(&def.symbol)
3561 .cloned()
3562 .unwrap_or_else(|| default_fee_rate(def.symbol));
3563 parse_spot_instrument(def, &fee, ts_init, ts_init).ok()
3564 },
3565 )
3566 .await?
3567 }
3568 BybitProductType::Linear => {
3569 let fee_map = self.fetch_fee_map(product_type, base_coin).await?;
3570 self.paginate_instruments::<BybitInstrumentLinear, _>(
3571 product_type,
3572 &symbol,
3573 base_coin,
3574 |def| {
3575 let fee = fee_map
3576 .get(&def.symbol)
3577 .cloned()
3578 .unwrap_or_else(|| default_fee_rate(def.symbol));
3579 parse_linear_instrument(def, &fee, ts_init, ts_init).ok()
3580 },
3581 )
3582 .await?
3583 }
3584 BybitProductType::Inverse => {
3585 let fee_map = self.fetch_fee_map(product_type, base_coin).await?;
3586 self.paginate_instruments::<BybitInstrumentInverse, _>(
3587 product_type,
3588 &symbol,
3589 base_coin,
3590 |def| {
3591 let fee = fee_map
3592 .get(&def.symbol)
3593 .cloned()
3594 .unwrap_or_else(|| default_fee_rate(def.symbol));
3595 parse_inverse_instrument(def, &fee, ts_init, ts_init).ok()
3596 },
3597 )
3598 .await?
3599 }
3600 BybitProductType::Option => {
3601 let fee_map = self.fetch_option_fee_map(base_coin).await?;
3602 self.paginate_instruments::<BybitInstrumentOption, _>(
3603 product_type,
3604 &symbol,
3605 base_coin,
3606 |def| {
3607 let fee = fee_map.get(&def.base_coin);
3608 parse_option_instrument(def, fee, ts_init, ts_init).ok()
3609 },
3610 )
3611 .await?
3612 }
3613 };
3614
3615 self.cache_instruments(&instruments);
3616
3617 Ok(instruments)
3618 }
3619
3620 pub async fn request_instruments_with_statuses(
3630 &self,
3631 product_type: BybitProductType,
3632 ) -> anyhow::Result<(
3633 Vec<InstrumentAny>,
3634 AHashMap<InstrumentId, MarketStatusAction>,
3635 )> {
3636 let ts_init = self.generate_ts_init();
3637 let mut statuses = AHashMap::new();
3638
3639 let default_fee_rate = |symbol: Ustr| BybitFeeRate {
3640 symbol,
3641 taker_fee_rate: "0.001".to_string(),
3642 maker_fee_rate: "0.001".to_string(),
3643 base_coin: None,
3644 };
3645
3646 let perp_status = |status: MarketStatusAction, is_scheduled_perp: bool| {
3648 if status == MarketStatusAction::Trading && is_scheduled_perp {
3649 MarketStatusAction::PreClose
3650 } else {
3651 status
3652 }
3653 };
3654
3655 let instruments = match product_type {
3656 BybitProductType::Spot => {
3657 let fee_map = self.fetch_fee_map(product_type, None).await?;
3658 self.paginate_instruments::<BybitInstrumentSpot, _>(
3659 product_type,
3660 &None::<String>,
3661 None,
3662 |def| {
3663 let id = InstrumentId::new(
3664 Symbol::from(make_bybit_symbol(def.symbol, product_type)),
3665 *BYBIT_VENUE,
3666 );
3667 statuses.insert(id, MarketStatusAction::from(def.status));
3668 let fee = fee_map
3669 .get(&def.symbol)
3670 .cloned()
3671 .unwrap_or_else(|| default_fee_rate(def.symbol));
3672 parse_spot_instrument(def, &fee, ts_init, ts_init).ok()
3673 },
3674 )
3675 .await?
3676 }
3677 BybitProductType::Linear => {
3678 let fee_map = self.fetch_fee_map(product_type, None).await?;
3679 self.paginate_instruments::<BybitInstrumentLinear, _>(
3680 product_type,
3681 &None::<String>,
3682 None,
3683 |def| {
3684 let id = InstrumentId::new(
3685 Symbol::from(make_bybit_symbol(def.symbol, product_type)),
3686 *BYBIT_VENUE,
3687 );
3688 let scheduled = def.contract_type == BybitContractType::LinearPerpetual
3689 && def.delivery_time != "0";
3690 statuses.insert(id, perp_status(def.status.into(), scheduled));
3691 let fee = fee_map
3692 .get(&def.symbol)
3693 .cloned()
3694 .unwrap_or_else(|| default_fee_rate(def.symbol));
3695 parse_linear_instrument(def, &fee, ts_init, ts_init).ok()
3696 },
3697 )
3698 .await?
3699 }
3700 BybitProductType::Inverse => {
3701 let fee_map = self.fetch_fee_map(product_type, None).await?;
3702 self.paginate_instruments::<BybitInstrumentInverse, _>(
3703 product_type,
3704 &None::<String>,
3705 None,
3706 |def| {
3707 let id = InstrumentId::new(
3708 Symbol::from(make_bybit_symbol(def.symbol, product_type)),
3709 *BYBIT_VENUE,
3710 );
3711 let scheduled = def.contract_type == BybitContractType::InversePerpetual
3712 && def.delivery_time != "0";
3713 statuses.insert(id, perp_status(def.status.into(), scheduled));
3714 let fee = fee_map
3715 .get(&def.symbol)
3716 .cloned()
3717 .unwrap_or_else(|| default_fee_rate(def.symbol));
3718 parse_inverse_instrument(def, &fee, ts_init, ts_init).ok()
3719 },
3720 )
3721 .await?
3722 }
3723 BybitProductType::Option => {
3724 let fee_map = self.fetch_option_fee_map(None).await?;
3725 self.paginate_instruments::<BybitInstrumentOption, _>(
3726 product_type,
3727 &None::<String>,
3728 None,
3729 |def| {
3730 let id = InstrumentId::new(
3731 Symbol::from(make_bybit_symbol(def.symbol, product_type)),
3732 *BYBIT_VENUE,
3733 );
3734 statuses.insert(id, MarketStatusAction::from(def.status));
3735 let fee = fee_map.get(&def.base_coin);
3736 parse_option_instrument(def, fee, ts_init, ts_init).ok()
3737 },
3738 )
3739 .await?
3740 }
3741 };
3742
3743 self.cache_instruments(&instruments);
3744
3745 Ok((instruments, statuses))
3746 }
3747
3748 pub async fn request_tickers(
3761 &self,
3762 params: &BybitTickersParams,
3763 ) -> anyhow::Result<Vec<BybitTickerData>> {
3764 use super::models::{
3765 BybitTickersLinearResponse, BybitTickersOptionResponse, BybitTickersSpotResponse,
3766 };
3767
3768 match params.category {
3769 BybitProductType::Spot => {
3770 let response: BybitTickersSpotResponse = self.inner.get_tickers(params).await?;
3771 Ok(response.result.list.into_iter().map(Into::into).collect())
3772 }
3773 BybitProductType::Linear | BybitProductType::Inverse => {
3774 let response: BybitTickersLinearResponse = self.inner.get_tickers(params).await?;
3775 Ok(response.result.list.into_iter().map(Into::into).collect())
3776 }
3777 BybitProductType::Option => {
3778 let response: BybitTickersOptionResponse = self.inner.get_tickers(params).await?;
3779 Ok(response.result.list.into_iter().map(Into::into).collect())
3780 }
3781 }
3782 }
3783
3784 pub async fn request_option_tickers_raw(
3792 &self,
3793 base_coin: &str,
3794 ) -> anyhow::Result<Vec<BybitTickerOption>> {
3795 let params = BybitTickersParams {
3796 category: BybitProductType::Option,
3797 symbol: None,
3798 base_coin: Some(base_coin.to_string()),
3799 exp_date: None,
3800 };
3801 let response: BybitTickersOptionResponse = self.inner.get_tickers(¶ms).await?;
3802 Ok(response.result.list)
3803 }
3804
3805 pub async fn request_option_tickers_raw_with_params(
3814 &self,
3815 params: &BybitTickersParams,
3816 ) -> anyhow::Result<Vec<BybitTickerOption>> {
3817 let response: BybitTickersOptionResponse = self.inner.get_tickers(params).await?;
3818 Ok(response.result.list)
3819 }
3820
3821 pub async fn request_trades(
3841 &self,
3842 product_type: BybitProductType,
3843 instrument_id: InstrumentId,
3844 limit: Option<u32>,
3845 ) -> anyhow::Result<Vec<TradeTick>> {
3846 let instrument = self.instrument_from_cache_by_id(instrument_id)?;
3847 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
3848
3849 let mut params_builder = BybitTradesParamsBuilder::default();
3850 params_builder.category(product_type);
3851 params_builder.symbol(bybit_symbol.raw_symbol().to_string());
3852
3853 if let Some(limit_val) = limit {
3854 params_builder.limit(limit_val);
3855 }
3856
3857 let params = params_builder.build().build_anyhow()?;
3858 let response = self.inner.get_recent_trades(¶ms).await?;
3859
3860 let mut trades = Vec::new();
3861
3862 for trade in response.result.list {
3863 if let Ok(trade_tick) = parse_trade_tick(&trade, &instrument, None) {
3864 trades.push(trade_tick);
3865 }
3866 }
3867
3868 Ok(trades)
3869 }
3870
3871 pub async fn request_funding_rates(
3884 &self,
3885 product_type: BybitProductType,
3886 instrument_id: InstrumentId,
3887 start: Option<Timestamp>,
3888 end: Option<Timestamp>,
3889 limit: Option<u32>,
3890 ) -> anyhow::Result<Vec<FundingRateUpdate>> {
3891 let instrument = self.instrument_from_cache_by_id(instrument_id)?;
3892 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
3893
3894 let start_ms = start.map(|dt| dt.as_millisecond());
3895 let mut seen_timestamps: AHashSet<i64> = AHashSet::new();
3896
3897 let mut raw_funding_rates = Vec::new();
3898
3899 let mut current_end_ms = match (start, end) {
3901 (Some(_), None) => Some(Timestamp::now().as_millisecond()),
3902 _ => end.map(|dt| dt.as_millisecond()),
3903 };
3904
3905 loop {
3906 let mut params_builder = BybitFundingParamsBuilder::default();
3907 params_builder.category(product_type);
3908 params_builder.symbol(bybit_symbol.raw_symbol().to_string());
3909 params_builder.limit(limit.unwrap_or(200).clamp(0, 200)); if let Some(start_val) = start_ms {
3912 params_builder.start_time(start_val);
3913 }
3914
3915 if let Some(end_val) = current_end_ms {
3916 params_builder.end_time(end_val);
3917 }
3918
3919 let params = params_builder.build().build_anyhow()?;
3920 let response = self.inner.get_funding_history(¶ms).await?;
3921
3922 let funding_rates = response.result.list;
3923
3924 let mut new_funding_rates_with_ts: Vec<(i64, _)> = funding_rates
3925 .into_iter()
3926 .filter_map(|f| {
3927 let Ok(ts) = f.funding_rate_timestamp.parse::<i64>() else {
3928 return None;
3929 };
3930
3931 seen_timestamps.insert(ts).then_some((ts, f))
3932 })
3933 .collect();
3934
3935 new_funding_rates_with_ts.sort_by_key(|(ts, _)| Reverse(*ts));
3936
3937 let earliest_funding_time = match new_funding_rates_with_ts.last() {
3938 Some((last_ts, _)) => *last_ts,
3939 None => break,
3940 };
3941
3942 let new_funding_rates = new_funding_rates_with_ts.into_iter().map(|(_, f)| f);
3943 raw_funding_rates.extend(new_funding_rates);
3944
3945 if let Some(limit_val) = limit
3947 && raw_funding_rates.len() >= limit_val as usize
3948 {
3949 break;
3950 }
3951
3952 if let Some(start_val) = start_ms
3953 && earliest_funding_time <= start_val
3954 {
3955 break;
3956 }
3957
3958 current_end_ms = Some(earliest_funding_time - 1);
3960 }
3961
3962 if let Some(limit_val) = limit {
3963 raw_funding_rates.truncate(limit_val as usize);
3964 }
3965 let mut rates: Vec<FundingRateUpdate> = Vec::with_capacity(raw_funding_rates.len());
3966
3967 for window in raw_funding_rates.windows(2) {
3968 let raw = &window[0];
3969 let timestamp = raw
3970 .funding_rate_timestamp
3971 .parse::<i64>()
3972 .map_err(|_| anyhow::anyhow!("invalid funding_rate_timestamp"))?;
3973 let older_timestamp = window[1]
3974 .funding_rate_timestamp
3975 .parse::<i64>()
3976 .map_err(|_| anyhow::anyhow!("invalid funding_rate_timestamp"))?;
3977
3978 let interval_millis = timestamp - older_timestamp;
3979 let rate = parse_funding_rate(raw, &instrument, Some(interval_millis))?;
3980
3981 rates.push(rate);
3982 }
3983
3984 if let Some(last_raw) = raw_funding_rates.last() {
3985 let rate = parse_funding_rate(last_raw, &instrument, None)?;
3986 rates.push(rate);
3987 }
3988
3989 rates.reverse();
3990
3991 Ok(rates)
3992 }
3993
3994 pub async fn request_orderbook_snapshot(
4012 &self,
4013 product_type: BybitProductType,
4014 instrument_id: InstrumentId,
4015 limit: Option<u32>,
4016 ) -> anyhow::Result<OrderBookDeltas> {
4017 let instrument = self.instrument_from_cache_by_id(instrument_id)?;
4018 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
4019
4020 let mut params_builder = BybitOrderbookParamsBuilder::default();
4021 params_builder.category(product_type);
4022 params_builder.symbol(bybit_symbol.raw_symbol().to_string());
4023
4024 if let Some(limit) = limit {
4025 let max_limit = match product_type {
4026 BybitProductType::Spot => 200,
4027 BybitProductType::Option => 25,
4028 BybitProductType::Linear | BybitProductType::Inverse => 500,
4029 };
4030 let clamped_limit = limit.min(max_limit);
4031 if limit > max_limit {
4032 log::warn!(
4033 "Bybit orderbook snapshot request depth limit exceeds venue maximum; clamping: limit={limit}, clamped_limit={clamped_limit}",
4034 );
4035 }
4036 params_builder.limit(clamped_limit);
4037 }
4038
4039 let params = params_builder.build().build_anyhow()?;
4040 let response = self.inner.get_orderbook(¶ms).await?;
4041
4042 let deltas = parse_orderbook(&response.result, &instrument, None)?;
4043
4044 Ok(deltas)
4045 }
4046
4047 pub async fn request_bars(
4060 &self,
4061 product_type: BybitProductType,
4062 bar_type: BarType,
4063 start: Option<Timestamp>,
4064 end: Option<Timestamp>,
4065 limit: Option<u32>,
4066 timestamp_on_close: bool,
4067 ) -> anyhow::Result<Vec<Bar>> {
4068 let instrument_id = bar_type.instrument_id();
4069 let instrument = self.instrument_from_cache_by_id(instrument_id)?;
4070 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
4071
4072 let interval = bar_spec_to_bybit_interval(
4074 bar_type.spec().aggregation,
4075 bar_type.spec().step.get() as u64,
4076 )?;
4077
4078 let start_ms = start.map(|dt| dt.as_millisecond());
4079 let mut seen_timestamps: AHashSet<i64> = AHashSet::new();
4080 let current_time_ms = get_atomic_clock_realtime().get_time_ms() as i64;
4081
4082 let mut pages: Vec<Vec<Bar>> = Vec::new();
4092 let mut total_bars = 0usize;
4093 let mut current_end = end.map(|dt| dt.as_millisecond());
4094 let mut page_count = 0;
4095
4096 loop {
4097 page_count += 1;
4098
4099 let mut params_builder = BybitKlinesParamsBuilder::default();
4100 params_builder.category(product_type);
4101 params_builder.symbol(bybit_symbol.raw_symbol().to_string());
4102 params_builder.interval(interval);
4103 params_builder.limit(1000u32); if let Some(start_val) = start_ms {
4106 params_builder.start(start_val);
4107 }
4108
4109 if let Some(end_val) = current_end {
4110 params_builder.end(end_val);
4111 }
4112
4113 let params = params_builder.build().build_anyhow()?;
4114 let response = self.inner.get_klines(¶ms).await?;
4115
4116 let klines = response.result.list;
4117 if klines.is_empty() {
4118 break;
4119 }
4120
4121 let mut klines_with_ts: Vec<(i64, _)> = klines
4123 .into_iter()
4124 .filter_map(|k| k.start.parse::<i64>().ok().map(|ts| (ts, k)))
4125 .collect();
4126
4127 klines_with_ts.sort_by_key(|(ts, _)| *ts);
4128
4129 let has_new = klines_with_ts
4131 .iter()
4132 .any(|(ts, _)| !seen_timestamps.contains(ts));
4133
4134 if !has_new {
4135 break;
4136 }
4137
4138 let mut page_bars = Vec::with_capacity(klines_with_ts.len());
4139
4140 let mut earliest_ts: Option<i64> = None;
4141
4142 for (start_time, kline) in &klines_with_ts {
4143 if earliest_ts.is_none_or(|ts| *start_time < ts) {
4145 earliest_ts = Some(*start_time);
4146 }
4147
4148 let bar_end_time = interval.bar_end_time_ms(*start_time);
4149 if bar_end_time > current_time_ms {
4150 continue;
4151 }
4152
4153 if !seen_timestamps.contains(start_time)
4154 && let Ok(bar) =
4155 parse_kline_bar(kline, &instrument, bar_type, timestamp_on_close, None)
4156 {
4157 page_bars.push(bar);
4158 seen_timestamps.insert(*start_time);
4159 }
4160 }
4161
4162 total_bars += page_bars.len();
4165 pages.push(page_bars);
4166
4167 if let Some(limit_val) = limit
4169 && total_bars >= limit_val as usize
4170 {
4171 break;
4172 }
4173
4174 let Some(earliest_bar_time) = earliest_ts else {
4177 break;
4178 };
4179
4180 if let Some(start_val) = start_ms
4181 && earliest_bar_time <= start_val
4182 {
4183 break;
4184 }
4185
4186 current_end = Some(earliest_bar_time - 1);
4187
4188 if page_count > 100 {
4190 break;
4191 }
4192 }
4193
4194 let mut all_bars: Vec<Bar> = Vec::with_capacity(total_bars);
4196 for page in pages.into_iter().rev() {
4197 all_bars.extend(page);
4198 }
4199
4200 if let Some(limit_val) = limit {
4202 let limit_usize = limit_val as usize;
4203 if all_bars.len() > limit_usize {
4204 let start_idx = all_bars.len() - limit_usize;
4205 return Ok(all_bars[start_idx..].to_vec());
4206 }
4207 }
4208
4209 Ok(all_bars)
4210 }
4211
4212 fn instrument_from_cache_by_id(
4213 &self,
4214 instrument_id: InstrumentId,
4215 ) -> anyhow::Result<InstrumentAny> {
4216 self.get_instrument(&instrument_id.symbol.inner())
4217 .ok_or_else(|| InstrumentLookupError::not_found(instrument_id).into())
4218 }
4219
4220 pub async fn request_fee_rates(
4232 &self,
4233 product_type: BybitProductType,
4234 symbol: Option<String>,
4235 base_coin: Option<String>,
4236 ) -> anyhow::Result<Vec<BybitFeeRate>> {
4237 let params = BybitFeeRateParams {
4238 category: product_type,
4239 symbol,
4240 base_coin,
4241 };
4242
4243 let response = self.inner.get_fee_rate(¶ms).await?;
4244 Ok(response.result.list)
4245 }
4246
4247 pub async fn request_account_state(
4259 &self,
4260 account_type: BybitAccountType,
4261 account_id: AccountId,
4262 ) -> anyhow::Result<AccountState> {
4263 let params = BybitWalletBalanceParams {
4264 account_type,
4265 coin: None,
4266 };
4267
4268 let response = self.inner.get_wallet_balance(¶ms).await?;
4269 let ts_init = self.generate_ts_init();
4270
4271 let wallet_balance = response
4273 .result
4274 .list
4275 .first()
4276 .ok_or_else(|| anyhow::anyhow!("No wallet balance found in response"))?;
4277
4278 parse_account_state(wallet_balance, account_id, ts_init)
4279 }
4280
4281 #[expect(clippy::too_many_arguments)]
4297 pub async fn request_order_status_reports(
4298 &self,
4299 account_id: AccountId,
4300 product_type: BybitProductType,
4301 instrument_id: Option<InstrumentId>,
4302 open_only: bool,
4303 start: Option<Timestamp>,
4304 end: Option<Timestamp>,
4305 limit: Option<u32>,
4306 ) -> anyhow::Result<Vec<OrderStatusReport>> {
4307 let symbol_param = if let Some(id) = instrument_id.as_ref() {
4309 let symbol_str = id.symbol.as_str();
4310 if symbol_str.is_empty() {
4311 None
4312 } else {
4313 Some(BybitSymbol::new(symbol_str)?.raw_symbol().to_string())
4314 }
4315 } else {
4316 None
4317 };
4318
4319 let settle_coins_to_query: Vec<Option<String>> =
4322 if product_type == BybitProductType::Linear && symbol_param.is_none() {
4323 vec![Some("USDT".to_string()), Some("USDC".to_string())]
4324 } else {
4325 match product_type {
4326 BybitProductType::Inverse => vec![None],
4327 _ => vec![None],
4328 }
4329 };
4330
4331 let mut all_collected_orders = Vec::new();
4332 let mut total_collected_across_coins = 0;
4333
4334 for settle_coin in settle_coins_to_query {
4335 let remaining_limit = if let Some(limit) = limit {
4336 let remaining = (limit as usize).saturating_sub(total_collected_across_coins);
4337 if remaining == 0 {
4338 break;
4339 }
4340 Some(remaining as u32)
4341 } else {
4342 None
4343 };
4344
4345 let orders_for_coin = if open_only {
4346 let mut all_orders = Vec::new();
4347 let mut seen_ids: AHashSet<Ustr> = AHashSet::new();
4348
4349 let order_filters: Vec<Option<BybitOrderFilter>> =
4352 if product_type == BybitProductType::Option {
4353 vec![None]
4354 } else {
4355 vec![None, Some(BybitOrderFilter::StopOrder)]
4356 };
4357
4358 let open_only_modes = [None, Some(BybitOpenOnly::ClosedRecent)];
4359
4360 for oo in open_only_modes {
4361 for order_filter in &order_filters {
4362 let mut cursor: Option<String> = None;
4363 let mut cursor_walk = CursorWalk::default();
4364
4365 loop {
4366 let remaining = if let Some(limit) = remaining_limit {
4367 (limit as usize).saturating_sub(all_orders.len())
4368 } else {
4369 usize::MAX
4370 };
4371
4372 if remaining == 0 {
4373 break;
4374 }
4375
4376 let page_limit = std::cmp::min(remaining, 50);
4378
4379 let mut p = BybitOpenOrdersParamsBuilder::default();
4380 p.category(product_type);
4381
4382 if let Some(symbol) = symbol_param.clone() {
4383 p.symbol(symbol);
4384 }
4385
4386 if let Some(coin) = settle_coin.clone() {
4387 p.settle_coin(coin);
4388 }
4389
4390 if let Some(of) = order_filter {
4391 p.order_filter(*of);
4392 }
4393
4394 if let Some(oo) = oo {
4395 p.open_only(oo);
4396 }
4397 p.limit(page_limit as u32);
4398
4399 if let Some(c) = cursor {
4400 p.cursor(c);
4401 }
4402 let params = p.build().build_anyhow()?;
4403 let response: BybitOpenOrdersResponse = self
4404 .inner
4405 .send_request(
4406 Method::GET,
4407 BYBIT_ORDER_REALTIME,
4408 Some(¶ms),
4409 None,
4410 true,
4411 )
4412 .await?;
4413
4414 for order in response.result.list {
4415 if seen_ids.insert(order.order_id) {
4416 all_orders.push(order);
4417 }
4418 }
4419
4420 if oo.is_some() {
4422 break;
4423 }
4424
4425 cursor = cursor_walk
4426 .advance(BYBIT_ORDER_REALTIME, response.result.next_page_cursor)?;
4427
4428 if cursor.is_none() {
4429 break;
4430 }
4431 }
4432 }
4433 }
4434
4435 all_orders
4436 } else {
4437 let mut all_orders = Vec::new();
4440 let mut open_orders = Vec::new();
4441 let mut seen_open_ids: AHashSet<Ustr> = AHashSet::new();
4442
4443 let order_filters: Vec<Option<BybitOrderFilter>> =
4446 if product_type == BybitProductType::Option {
4447 vec![None]
4448 } else {
4449 vec![None, Some(BybitOrderFilter::StopOrder)]
4450 };
4451
4452 for order_filter in &order_filters {
4453 let mut cursor: Option<String> = None;
4454 let mut cursor_walk = CursorWalk::default();
4455
4456 loop {
4457 let remaining = if let Some(limit) = remaining_limit {
4458 (limit as usize).saturating_sub(open_orders.len())
4459 } else {
4460 usize::MAX
4461 };
4462
4463 if remaining == 0 {
4464 break;
4465 }
4466
4467 let page_limit = std::cmp::min(remaining, 50);
4469
4470 let mut open_params = BybitOpenOrdersParamsBuilder::default();
4471 open_params.category(product_type);
4472
4473 if let Some(symbol) = symbol_param.clone() {
4474 open_params.symbol(symbol);
4475 }
4476
4477 if let Some(coin) = settle_coin.clone() {
4478 open_params.settle_coin(coin);
4479 }
4480
4481 if let Some(of) = order_filter {
4482 open_params.order_filter(*of);
4483 }
4484 open_params.limit(page_limit as u32);
4485
4486 if let Some(c) = cursor {
4487 open_params.cursor(c);
4488 }
4489 let open_params = open_params.build().build_anyhow()?;
4490 let open_response: BybitOpenOrdersResponse = self
4491 .inner
4492 .send_request(
4493 Method::GET,
4494 BYBIT_ORDER_REALTIME,
4495 Some(&open_params),
4496 None,
4497 true,
4498 )
4499 .await?;
4500
4501 for order in open_response.result.list {
4502 if !seen_open_ids.contains(&order.order_id) {
4503 seen_open_ids.insert(order.order_id);
4504 open_orders.push(order);
4505 }
4506 }
4507
4508 cursor = cursor_walk
4509 .advance(BYBIT_ORDER_REALTIME, open_response.result.next_page_cursor)?;
4510
4511 if cursor.is_none() {
4512 break;
4513 }
4514 }
4515 }
4516
4517 let seen_order_ids: AHashSet<Ustr> = seen_open_ids;
4518 let total_open_orders = open_orders.len();
4519
4520 all_orders.extend(open_orders);
4521
4522 let mut total_history_orders = 0;
4523
4524 for order_filter in &order_filters {
4525 let mut cursor: Option<String> = None;
4526 let mut cursor_walk = CursorWalk::default();
4527
4528 loop {
4529 let total_orders = total_open_orders + total_history_orders;
4530 let remaining = if let Some(limit) = remaining_limit {
4531 (limit as usize).saturating_sub(total_orders)
4532 } else {
4533 usize::MAX
4534 };
4535
4536 if remaining == 0 {
4537 break;
4538 }
4539
4540 let page_limit = std::cmp::min(remaining, 50);
4542
4543 let mut history_params = BybitOrderHistoryParamsBuilder::default();
4544 history_params.category(product_type);
4545
4546 if let Some(symbol) = symbol_param.clone() {
4547 history_params.symbol(symbol);
4548 }
4549
4550 if let Some(coin) = settle_coin.clone() {
4551 history_params.settle_coin(coin);
4552 }
4553
4554 if let Some(of) = order_filter {
4555 history_params.order_filter(*of);
4556 }
4557
4558 if let Some(start) = start {
4559 history_params.start_time(start.as_millisecond());
4560 }
4561
4562 if let Some(end) = end {
4563 history_params.end_time(end.as_millisecond());
4564 }
4565 history_params.limit(page_limit as u32);
4566
4567 if let Some(c) = cursor {
4568 history_params.cursor(c);
4569 }
4570 let history_params = history_params.build().build_anyhow()?;
4571 let history_response: BybitOrderHistoryResponse = self
4572 .inner
4573 .send_request(
4574 Method::GET,
4575 BYBIT_ORDER_HISTORY,
4576 Some(&history_params),
4577 None,
4578 true,
4579 )
4580 .await?;
4581
4582 for order in history_response.result.list {
4584 if !seen_order_ids.contains(&order.order_id) {
4585 all_orders.push(order);
4586 total_history_orders += 1;
4587 }
4588 }
4589
4590 cursor = cursor_walk.advance(
4591 BYBIT_ORDER_HISTORY,
4592 history_response.result.next_page_cursor,
4593 )?;
4594
4595 if cursor.is_none() {
4596 break;
4597 }
4598 }
4599 }
4600
4601 all_orders
4602 };
4603
4604 total_collected_across_coins += orders_for_coin.len();
4605 all_collected_orders.extend(orders_for_coin);
4606 }
4607
4608 let ts_init = self.generate_ts_init();
4609
4610 let mut reports = Vec::new();
4611
4612 for order in all_collected_orders {
4613 if let Some(ref instrument_id) = instrument_id {
4614 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
4615
4616 if let Ok(report) =
4617 parse_order_status_report(&order, &instrument, account_id, ts_init)
4618 {
4619 reports.push(report);
4620 }
4621 } else {
4622 if !order.symbol.is_empty() {
4625 let symbol_with_product =
4626 Symbol::from_ustr_unchecked(make_bybit_symbol(order.symbol, product_type));
4627
4628 let Ok(instrument) = self.instrument_from_cache(&symbol_with_product) else {
4629 log::debug!(
4630 "Skipping order report for instrument not in cache: symbol={}, full_symbol={}",
4631 order.symbol,
4632 symbol_with_product
4633 );
4634 continue;
4635 };
4636
4637 match parse_order_status_report(&order, &instrument, account_id, ts_init) {
4638 Ok(report) => reports.push(report),
4639 Err(e) => {
4640 log::error!("Failed to parse order status report: {e}");
4641 }
4642 }
4643 }
4644 }
4645 }
4646
4647 Ok(reports)
4648 }
4649
4650 pub async fn request_fill_reports(
4662 &self,
4663 account_id: AccountId,
4664 product_type: BybitProductType,
4665 instrument_id: Option<InstrumentId>,
4666 start: Option<i64>,
4667 end: Option<i64>,
4668 limit: Option<u32>,
4669 ) -> anyhow::Result<Vec<FillReport>> {
4670 let symbol = if let Some(id) = instrument_id {
4672 let bybit_symbol = BybitSymbol::new(id.symbol.as_str())?;
4673 Some(bybit_symbol.raw_symbol().to_string())
4674 } else {
4675 None
4676 };
4677
4678 let mut all_executions = Vec::new();
4680 let mut cursor: Option<String> = None;
4681 let mut cursor_walk = CursorWalk::default();
4682 let mut total_executions = 0;
4683
4684 loop {
4685 let remaining = if let Some(limit) = limit {
4687 (limit as usize).saturating_sub(total_executions)
4688 } else {
4689 usize::MAX
4690 };
4691
4692 if remaining == 0 {
4694 break;
4695 }
4696
4697 let page_limit = std::cmp::min(remaining, 100);
4699
4700 let params = BybitTradeHistoryParams {
4701 category: product_type,
4702 symbol: symbol.clone(),
4703 base_coin: None,
4704 order_id: None,
4705 order_link_id: None,
4706 start_time: start,
4707 end_time: end,
4708 exec_type: None,
4709 limit: Some(page_limit as u32),
4710 cursor: cursor.clone(),
4711 };
4712
4713 let response = self.inner.get_trade_history(¶ms).await?;
4714 for execution in response.result.list {
4715 if execution.exec_type == BybitExecType::Funding {
4716 log::debug!(
4717 "Skipping funding execution: symbol={}, order_id={}, exec_id={}",
4718 execution.symbol,
4719 execution.order_id,
4720 execution.exec_id,
4721 );
4722 continue;
4723 }
4724
4725 all_executions.push(execution);
4726 total_executions += 1;
4727 }
4728
4729 cursor = cursor_walk.advance(BYBIT_EXECUTION_LIST, response.result.next_page_cursor)?;
4730 if cursor.is_none() {
4731 break;
4732 }
4733 }
4734
4735 let ts_init = self.generate_ts_init();
4736 let mut reports = Vec::new();
4737
4738 for execution in all_executions {
4739 let symbol_with_product =
4742 Symbol::from_ustr_unchecked(make_bybit_symbol(execution.symbol, product_type));
4743
4744 let Ok(instrument) = self.instrument_from_cache(&symbol_with_product) else {
4745 log::debug!(
4746 "Skipping fill report for instrument not in cache: symbol={}, full_symbol={}",
4747 execution.symbol,
4748 symbol_with_product
4749 );
4750 continue;
4751 };
4752
4753 match parse_fill_report(&execution, account_id, &instrument, ts_init) {
4754 Ok(report) => reports.push(report),
4755 Err(e) => {
4756 log::error!("Failed to parse fill report: {e}");
4757 }
4758 }
4759 }
4760
4761 Ok(reports)
4762 }
4763
4764 pub async fn request_position_status_reports(
4777 &self,
4778 account_id: AccountId,
4779 product_type: BybitProductType,
4780 instrument_id: Option<InstrumentId>,
4781 ) -> anyhow::Result<Vec<PositionStatusReport>> {
4782 if product_type == BybitProductType::Spot {
4784 if self.use_spot_position_reports.load(Ordering::Relaxed) {
4785 let Some(instrument_id) = instrument_id else {
4786 anyhow::bail!(
4787 "SPOT wallet balances carry no pair identity and cannot be attributed for a bulk position report request"
4788 );
4789 };
4790 return self
4791 .generate_spot_position_reports_from_wallet(account_id, instrument_id)
4792 .await;
4793 } else {
4794 return Ok(Vec::new());
4796 }
4797 }
4798
4799 let ts_init = self.generate_ts_init();
4800 let mut reports = Vec::new();
4801
4802 let symbol = if let Some(id) = instrument_id {
4804 let symbol_str = id.symbol.as_str();
4805 if symbol_str.is_empty() {
4806 anyhow::bail!("InstrumentId symbol is empty");
4807 }
4808 let bybit_symbol = BybitSymbol::new(symbol_str)?;
4809 Some(bybit_symbol.raw_symbol().to_string())
4810 } else {
4811 None
4812 };
4813
4814 if product_type == BybitProductType::Linear && symbol.is_none() {
4817 for settle_coin in ["USDT", "USDC"] {
4819 let mut cursor: Option<String> = None;
4820 let mut cursor_walk = CursorWalk::default();
4821
4822 loop {
4823 let params = BybitPositionListParams {
4824 category: product_type,
4825 symbol: None,
4826 base_coin: None,
4827 settle_coin: Some(settle_coin.to_string()),
4828 limit: Some(200), cursor: cursor.clone(),
4830 };
4831
4832 let response = self.inner.get_positions(¶ms).await?;
4833
4834 for position in response.result.list {
4835 if position.symbol.is_empty() {
4836 continue;
4837 }
4838
4839 let symbol_with_product = Symbol::new(format!(
4840 "{}{}",
4841 position.symbol.as_str(),
4842 product_type.suffix()
4843 ));
4844
4845 let Ok(instrument) = self.instrument_from_cache(&symbol_with_product)
4846 else {
4847 log::debug!(
4848 "Skipping position report for instrument not in cache: symbol={}, full_symbol={}",
4849 position.symbol,
4850 symbol_with_product
4851 );
4852 continue;
4853 };
4854
4855 match parse_position_status_report(
4856 &position,
4857 account_id,
4858 &instrument,
4859 ts_init,
4860 ) {
4861 Ok(report) => reports.push(report),
4862 Err(e) => {
4863 log::error!("Failed to parse position status report: {e}");
4864 }
4865 }
4866 }
4867
4868 cursor = cursor_walk
4869 .advance(BYBIT_POSITION_LIST, response.result.next_page_cursor)?;
4870
4871 if cursor.is_none() {
4872 break;
4873 }
4874 }
4875 }
4876 } else {
4877 let mut cursor: Option<String> = None;
4879 let mut cursor_walk = CursorWalk::default();
4880
4881 loop {
4882 let params = BybitPositionListParams {
4883 category: product_type,
4884 symbol: symbol.clone(),
4885 base_coin: None,
4886 settle_coin: None,
4887 limit: Some(200), cursor: cursor.clone(),
4889 };
4890
4891 let response = self.inner.get_positions(¶ms).await?;
4892
4893 for position in response.result.list {
4894 if position.symbol.is_empty() {
4895 continue;
4896 }
4897
4898 let symbol_with_product = Symbol::new(format!(
4899 "{}{}",
4900 position.symbol.as_str(),
4901 product_type.suffix()
4902 ));
4903
4904 let Ok(instrument) = self.instrument_from_cache(&symbol_with_product) else {
4905 log::debug!(
4906 "Skipping position report for instrument not in cache: symbol={}, full_symbol={}",
4907 position.symbol,
4908 symbol_with_product
4909 );
4910 continue;
4911 };
4912
4913 match parse_position_status_report(&position, account_id, &instrument, ts_init)
4914 {
4915 Ok(report) => reports.push(report),
4916 Err(e) => {
4917 log::error!("Failed to parse position status report: {e}");
4918 }
4919 }
4920 }
4921
4922 cursor =
4923 cursor_walk.advance(BYBIT_POSITION_LIST, response.result.next_page_cursor)?;
4924
4925 if cursor.is_none() {
4926 break;
4927 }
4928 }
4929 }
4930
4931 Ok(reports)
4932 }
4933
4934 async fn query_order_by_id(
4935 &self,
4936 product_type: BybitProductType,
4937 order_id: &str,
4938 endpoint: &str,
4939 context: &str,
4940 ) -> anyhow::Result<BybitOrder> {
4941 let mut query_params = BybitOpenOrdersParamsBuilder::default();
4942 query_params.category(product_type);
4943 query_params.order_id(order_id.to_string());
4944
4945 let query_params = query_params.build().build_anyhow()?;
4946 let order_response: BybitOpenOrdersResponse = self
4947 .inner
4948 .send_request(Method::GET, endpoint, Some(&query_params), None, true)
4949 .await?;
4950
4951 order_response
4952 .result
4953 .list
4954 .into_iter()
4955 .next()
4956 .ok_or_else(|| anyhow::anyhow!("No order returned {context}"))
4957 }
4958}
4959
4960#[cfg(test)]
4961mod tests {
4962 use nautilus_testkit::http::assert_http_redirect_rejected;
4963 use rstest::rstest;
4964
4965 use super::*;
4966
4967 #[tokio::test]
4968 async fn test_authenticated_client_rejects_redirects() {
4969 let client = BybitRawHttpClient::build_http_client(3, None).unwrap();
4970 assert_http_redirect_rejected(|url| async move {
4971 client
4972 .get(url, None, None, Some(3), None)
4973 .await
4974 .unwrap()
4975 .status
4976 .as_u16()
4977 })
4978 .await;
4979 }
4980
4981 #[rstest]
4982 fn test_client_creation() {
4983 let client = BybitHttpClient::new(None, 60, 3, 1000, 10_000, 5_000, None);
4984 assert!(client.is_ok());
4985
4986 let client = client.unwrap();
4987 assert!(client.base_url().contains("bybit.com"));
4988 assert!(client.credential().is_none());
4989 }
4990
4991 #[rstest]
4992 fn test_client_with_credentials() {
4993 let client = BybitHttpClient::with_credentials(
4994 "test_key".to_string(),
4995 "test_secret".to_string(),
4996 Some("https://api-testnet.bybit.com".to_string()),
4997 60,
4998 3,
4999 1000,
5000 10_000,
5001 5_000,
5002 None,
5003 );
5004 assert!(client.is_ok());
5005
5006 let client = client.unwrap();
5007 assert!(client.credential().is_some());
5008 }
5009
5010 #[rstest]
5011 fn test_build_path_with_params() {
5012 #[derive(Serialize)]
5013 struct TestParams {
5014 category: String,
5015 symbol: String,
5016 }
5017
5018 let params = TestParams {
5019 category: "linear".to_string(),
5020 symbol: "BTCUSDT".to_string(),
5021 };
5022
5023 let path = BybitRawHttpClient::build_path("/v5/market/test", ¶ms);
5024 assert!(path.is_ok());
5025 assert!(path.unwrap().contains("category=linear"));
5026 }
5027
5028 #[rstest]
5029 fn test_build_path_without_params() {
5030 let params = ();
5031 let path = BybitRawHttpClient::build_path("/v5/market/time", ¶ms);
5032 assert!(path.is_ok());
5033 assert_eq!(path.unwrap(), "/v5/market/time");
5034 }
5035
5036 #[rstest]
5037 fn test_params_serialization_matches_build_path() {
5038 #[derive(Serialize)]
5040 struct TestParams {
5041 category: String,
5042 limit: u32,
5043 }
5044
5045 let params = TestParams {
5046 category: "spot".to_string(),
5047 limit: 50,
5048 };
5049
5050 let old_path = BybitRawHttpClient::build_path(BYBIT_ORDER_REALTIME, ¶ms).unwrap();
5052 let old_query = old_path.split('?').nth(1).unwrap_or("");
5053
5054 let new_query = serde_urlencoded::to_string(¶ms).unwrap();
5056
5057 assert_eq!(old_query, new_query);
5059 }
5060
5061 #[rstest]
5062 fn test_params_serialization_order() {
5063 #[derive(Serialize)]
5065 struct OrderParams {
5066 category: String,
5067 symbol: String,
5068 limit: u32,
5069 }
5070
5071 let params = OrderParams {
5072 category: "spot".to_string(),
5073 symbol: "BTCUSDT".to_string(),
5074 limit: 50,
5075 };
5076
5077 let query1 = serde_urlencoded::to_string(¶ms).unwrap();
5079 let query2 = serde_urlencoded::to_string(¶ms).unwrap();
5080 let query3 = serde_urlencoded::to_string(¶ms).unwrap();
5081
5082 assert_eq!(query1, query2);
5083 assert_eq!(query2, query3);
5084
5085 assert!(query1.contains("category=spot"));
5087 assert!(query1.contains("symbol=BTCUSDT"));
5088 assert!(query1.contains("limit=50"));
5089 }
5090
5091 #[rstest]
5092 #[case(403, "Access too frequent", true)]
5093 #[case(403, "Forbidden", false)]
5094 #[case(429, "Access too frequent", false)]
5095 fn test_rate_limit_403_detection(
5096 #[case] status: u16,
5097 #[case] body: &str,
5098 #[case] expected: bool,
5099 ) {
5100 assert_eq!(
5101 BybitRawHttpClient::is_rate_limit_403(status, body),
5102 expected
5103 );
5104 }
5105
5106 #[rstest]
5107 #[case(
5108 "https://api-demo.bybit.com",
5109 BybitProductType::Linear,
5110 10001,
5111 "",
5112 "Bybit demo rejected the linear fee rate request via /v5/account/fee-rate \
5113 (error 10001, no message); demo derivatives fee rates appear unsupported, using defaults"
5114 )]
5115 #[case(
5116 "https://api-demo.bybit.com",
5117 BybitProductType::Inverse,
5118 10001,
5119 "",
5120 "Bybit demo rejected the inverse fee rate request via /v5/account/fee-rate \
5121 (error 10001, no message); demo derivatives fee rates appear unsupported, using defaults"
5122 )]
5123 #[case(
5124 "https://api.bybit.com",
5125 BybitProductType::Spot,
5126 10001,
5127 "Parameter error",
5128 "Fee rate request rejected for spot instruments via /v5/account/fee-rate \
5129 (error 10001: Parameter error), using defaults"
5130 )]
5131 #[case(
5132 "https://api-demo.bybit.com",
5133 BybitProductType::Spot,
5134 10001,
5135 "Parameter error",
5136 "Fee rate request rejected for spot instruments via /v5/account/fee-rate \
5137 (error 10001: Parameter error), using defaults"
5138 )]
5139 #[case(
5140 "https://api.bybit.com",
5141 BybitProductType::Linear,
5142 10001,
5143 "Parameter error",
5144 "Fee rate request rejected for linear instruments via /v5/account/fee-rate \
5145 (error 10001: Parameter error), using defaults"
5146 )]
5147 fn test_fee_rate_rejection_warning(
5148 #[case] base_url: &str,
5149 #[case] product_type: BybitProductType,
5150 #[case] error_code: i32,
5151 #[case] message: &str,
5152 #[case] expected: &str,
5153 ) {
5154 let client =
5155 BybitHttpClient::new(Some(base_url.to_string()), 60, 3, 1000, 10_000, 5_000, None)
5156 .unwrap();
5157
5158 let warning = client.fee_rate_rejection_warning(product_type, error_code, message);
5159
5160 assert_eq!(warning, expected);
5161 }
5162
5163 #[rstest]
5164 #[case(10001, "", "error 10001, no message")]
5165 #[case(10001, "Parameter error", "error 10001: Parameter error")]
5166 fn test_format_bybit_error_detail(
5167 #[case] error_code: i32,
5168 #[case] message: &str,
5169 #[case] expected: &str,
5170 ) {
5171 let detail = BybitHttpClient::format_bybit_error_detail(error_code, message);
5172
5173 assert_eq!(detail, expected);
5174 }
5175
5176 #[rstest]
5177 #[case(
5178 10001,
5179 "",
5180 "Option fee rate request rejected via /v5/account/fee-rate \
5181 (error 10001, no message), using defaults"
5182 )]
5183 #[case(
5184 10001,
5185 "Parameter error",
5186 "Option fee rate request rejected via /v5/account/fee-rate \
5187 (error 10001: Parameter error), using defaults"
5188 )]
5189 fn test_option_fee_rate_warning_message(
5190 #[case] error_code: i32,
5191 #[case] message: &str,
5192 #[case] expected: &str,
5193 ) {
5194 let error_detail = BybitHttpClient::format_bybit_error_detail(error_code, message);
5195 let warning = format!(
5196 "Option fee rate request rejected via /v5/account/fee-rate ({error_detail}), using defaults"
5197 );
5198
5199 assert_eq!(warning, expected);
5200 }
5201
5202 #[rstest]
5203 fn test_cursor_walk_stops_on_absent_or_empty_cursor() {
5204 let mut walk = CursorWalk::default();
5205
5206 assert_eq!(walk.advance("/endpoint", None).unwrap(), None);
5207 assert_eq!(
5208 walk.advance("/endpoint", Some(String::new())).unwrap(),
5209 None
5210 );
5211 assert!(walk.followed.is_empty());
5212 }
5213
5214 #[rstest]
5215 fn test_cursor_walk_follows_advancing_cursors() {
5216 let mut walk = CursorWalk::default();
5217
5218 for page in ["page-2", "page-3", "page-4"] {
5219 assert_eq!(
5220 walk.advance("/endpoint", Some(page.to_string())).unwrap(),
5221 Some(page.to_string())
5222 );
5223 }
5224 }
5225
5226 #[rstest]
5227 fn test_cursor_walk_rejects_a_cursor_that_does_not_advance() {
5228 let mut walk = CursorWalk::default();
5229 walk.advance("/endpoint", Some("page-2".to_string()))
5230 .unwrap();
5231
5232 let error = walk
5233 .advance("/endpoint", Some("page-2".to_string()))
5234 .expect_err("a cursor that addresses the page it came with has no next page");
5235
5236 assert_eq!(
5237 error.to_string(),
5238 r#"/endpoint pagination cursor did not advance from "page-2""#
5239 );
5240 }
5241
5242 #[rstest]
5243 #[case::two_cycle(vec!["page-2", "page-3"], "page-2")]
5244 #[case::long_cycle(vec!["page-2", "page-3", "page-4", "page-5"], "page-3")]
5245 fn test_cursor_walk_rejects_a_cursor_already_followed(
5246 #[case] followed: Vec<&str>,
5247 #[case] repeated: &str,
5248 ) {
5249 let mut walk = CursorWalk::default();
5250 for cursor in followed {
5251 walk.advance("/endpoint", Some(cursor.to_string())).unwrap();
5252 }
5253
5254 let error = walk
5255 .advance("/endpoint", Some(repeated.to_string()))
5256 .expect_err("a cursor already followed has no further page to offer");
5257
5258 assert_eq!(
5259 error.to_string(),
5260 format!(r#"/endpoint pagination repeated cursor "{repeated}""#)
5261 );
5262 }
5263}