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, consts::NAUTILUS_USER_AGENT, 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, Method, USER_AGENT},
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, BybitMarginMode,
96 BybitOpenOnly, BybitOrderFilter, BybitOrderSide, BybitOrderType, BybitPositionIdx,
97 BybitPositionMode, BybitProductType, BybitRepayStatus, BybitTpSlMode,
98 },
99 models::{BybitCursorListResponse, BybitErrorCheck, BybitResponseCheck},
100 parse::{
101 bar_spec_to_bybit_interval, bybit_rejection_due_post_only, make_bybit_symbol,
102 map_time_in_force, parse_account_state, parse_fill_report, parse_funding_rate,
103 parse_inverse_instrument, parse_kline_bar, parse_linear_instrument,
104 parse_option_instrument, parse_order_status_report, parse_orderbook,
105 parse_position_status_report, parse_spot_instrument, parse_trade_tick, spot_leverage,
106 spot_market_unit, trigger_direction,
107 },
108 rate_limit::{
109 BYBIT_RATE_LIMIT_HEADER, BYBIT_RATE_LIMIT_RESET_HEADER, BYBIT_RATE_LIMIT_STATUS_HEADER,
110 BybitRateLimiter, batch_call_limit, batch_endpoint_limit, batch_send_limit, batch_weight,
111 category_from_payload,
112 },
113 retry::should_retry_http,
114 symbol::BybitSymbol,
115 urls::bybit_http_base_url,
116};
117
118const DEFAULT_RECV_WINDOW_MS: u64 = 5_000;
119
120trait BuilderResultExt<T> {
121 fn build_anyhow(self) -> anyhow::Result<T>;
122}
123
124impl<T, E: Display> BuilderResultExt<T> for Result<T, E> {
125 fn build_anyhow(self) -> anyhow::Result<T> {
126 self.map_err(|e| anyhow::anyhow!("{e}"))
127 }
128}
129
130const BYBIT_ORDER_REALTIME: &str = "/v5/order/realtime";
131const BYBIT_ORDER_HISTORY: &str = "/v5/order/history";
132
133pub static BYBIT_REST_QUOTA: LazyLock<Quota> = LazyLock::new(|| {
135 Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant")
136});
137
138pub static BYBIT_REPAY_QUOTA: LazyLock<Quota> = LazyLock::new(|| {
140 Quota::per_second(NonZeroU32::new(1).expect("non-zero")).expect("valid constant")
141});
142
143#[cfg_attr(
148 feature = "python",
149 pyo3::pyclass(module = "nautilus_trader.adapters.bybit", from_py_object)
150)]
151#[cfg_attr(
152 feature = "python",
153 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.bybit")
154)]
155#[derive(Clone)]
156pub struct BybitRawHttpClient {
157 base_url: String,
158 client: Arc<ArcSwap<HttpClient>>,
159 rate_limiter: BybitRateLimiter,
160 credential: Option<Credential>,
161 recv_window_ms: u64,
162 timeout_secs: u64,
163 proxy_url: Option<String>,
164 session_generation: Arc<AtomicU64>,
165 retry_manager: RetryManager<BybitHttpError>,
166 cancellation_token: Arc<parking_lot::Mutex<CancellationToken>>,
167}
168
169impl Default for BybitRawHttpClient {
170 fn default() -> Self {
171 Self::new(None, 60, 3, 1000, 10_000, DEFAULT_RECV_WINDOW_MS, None)
172 .expect("Failed to create default BybitRawHttpClient")
173 }
174}
175
176impl Debug for BybitRawHttpClient {
177 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
178 f.debug_struct(stringify!(BybitRawHttpClient))
179 .field("base_url", &self.base_url)
180 .field("has_credentials", &self.credential.is_some())
181 .field("recv_window_ms", &self.recv_window_ms)
182 .finish()
183 }
184}
185
186impl BybitRawHttpClient {
187 pub fn cancel_all_requests(&self) {
189 self.cancellation_token.lock().cancel();
190 }
191
192 pub fn reset_cancellation_token(&self) {
195 let mut guard = self.cancellation_token.lock();
196 *guard = CancellationToken::new();
197 }
198
199 pub fn cancellation_token(&self) -> CancellationToken {
201 self.cancellation_token.lock().clone()
202 }
203
204 pub fn new(
210 base_url: Option<String>,
211 timeout_secs: u64,
212 max_retries: u32,
213 retry_delay_ms: u64,
214 retry_delay_max_ms: u64,
215 recv_window_ms: u64,
216 proxy_url: Option<String>,
217 ) -> Result<Self, BybitHttpError> {
218 let retry_config = RetryConfig {
219 max_retries,
220 initial_delay_ms: retry_delay_ms,
221 max_delay_ms: retry_delay_max_ms,
222 backoff_factor: 2.0,
223 jitter_ms: 1000,
224 operation_timeout_ms: Some(60_000),
225 immediate_first: false,
226 max_elapsed_ms: Some(180_000),
227 };
228
229 let retry_manager = RetryManager::new(retry_config);
230 let base_url =
231 base_url.unwrap_or_else(|| bybit_http_base_url(BybitEnvironment::Mainnet).to_string());
232 let rate_limiter = BybitRateLimiter::for_http(&base_url, None, proxy_url.as_deref());
233 let client = Self::build_http_client(timeout_secs, proxy_url.clone())?;
234 let session_generation = rate_limiter.http_session_generation();
235
236 Ok(Self {
237 base_url,
238 client: Arc::new(ArcSwap::from_pointee(client)),
239 rate_limiter,
240 credential: None,
241 recv_window_ms,
242 timeout_secs,
243 proxy_url,
244 session_generation: Arc::new(AtomicU64::new(session_generation)),
245 retry_manager,
246 cancellation_token: Arc::new(parking_lot::Mutex::new(CancellationToken::new())),
247 })
248 }
249
250 #[expect(clippy::too_many_arguments)]
256 pub fn with_credentials(
257 api_key: String,
258 api_secret: String,
259 base_url: Option<String>,
260 timeout_secs: u64,
261 max_retries: u32,
262 retry_delay_ms: u64,
263 retry_delay_max_ms: u64,
264 recv_window_ms: u64,
265 proxy_url: Option<String>,
266 ) -> Result<Self, BybitHttpError> {
267 let retry_config = RetryConfig {
268 max_retries,
269 initial_delay_ms: retry_delay_ms,
270 max_delay_ms: retry_delay_max_ms,
271 backoff_factor: 2.0,
272 jitter_ms: 1000,
273 operation_timeout_ms: Some(60_000),
274 immediate_first: false,
275 max_elapsed_ms: Some(180_000),
276 };
277
278 let retry_manager = RetryManager::new(retry_config);
279 let base_url =
280 base_url.unwrap_or_else(|| bybit_http_base_url(BybitEnvironment::Mainnet).to_string());
281 let credential = Credential::new(api_key, api_secret);
282 let rate_limiter =
283 BybitRateLimiter::for_http(&base_url, Some(credential.api_key()), proxy_url.as_deref());
284 let client = Self::build_http_client(timeout_secs, proxy_url.clone())?;
285 let session_generation = rate_limiter.http_session_generation();
286
287 Ok(Self {
288 base_url,
289 client: Arc::new(ArcSwap::from_pointee(client)),
290 rate_limiter,
291 credential: Some(credential),
292 recv_window_ms,
293 timeout_secs,
294 proxy_url,
295 session_generation: Arc::new(AtomicU64::new(session_generation)),
296 retry_manager,
297 cancellation_token: Arc::new(parking_lot::Mutex::new(CancellationToken::new())),
298 })
299 }
300
301 #[expect(clippy::too_many_arguments)]
313 pub fn new_with_env(
314 api_key: Option<String>,
315 api_secret: Option<String>,
316 base_url: Option<String>,
317 demo: bool,
318 testnet: bool,
319 timeout_secs: u64,
320 max_retries: u32,
321 retry_delay_ms: u64,
322 retry_delay_max_ms: u64,
323 recv_window_ms: u64,
324 proxy_url: Option<String>,
325 ) -> Result<Self, BybitHttpError> {
326 let environment = if demo {
327 BybitEnvironment::Demo
328 } else if testnet {
329 BybitEnvironment::Testnet
330 } else {
331 BybitEnvironment::Mainnet
332 };
333 let base_url =
334 Some(base_url.unwrap_or_else(|| bybit_http_base_url(environment).to_string()));
335 let (key_var, secret_var) = credential_env_vars(environment);
336 let key = get_or_env_var_opt(api_key, key_var);
337 let secret = get_or_env_var_opt(api_secret, secret_var);
338
339 if let (Some(k), Some(s)) = (key, secret) {
340 Self::with_credentials(
341 k,
342 s,
343 base_url,
344 timeout_secs,
345 max_retries,
346 retry_delay_ms,
347 retry_delay_max_ms,
348 recv_window_ms,
349 proxy_url,
350 )
351 } else {
352 Self::new(
353 base_url,
354 timeout_secs,
355 max_retries,
356 retry_delay_ms,
357 retry_delay_max_ms,
358 recv_window_ms,
359 proxy_url,
360 )
361 }
362 }
363
364 fn default_headers() -> HashMap<String, String> {
365 HashMap::from([
366 (USER_AGENT.to_string(), NAUTILUS_USER_AGENT.to_string()),
367 (
368 "X-Referer".to_string(),
369 BYBIT_NAUTILUS_BROKER_ID.to_string(),
370 ),
371 ])
372 }
373
374 fn build_http_client(
375 timeout_secs: u64,
376 proxy_url: Option<String>,
377 ) -> Result<HttpClient, BybitHttpError> {
378 HttpClient::builder()
379 .headers(Self::default_headers())
380 .header_keys(vec![
381 BYBIT_RATE_LIMIT_HEADER.to_string(),
382 BYBIT_RATE_LIMIT_STATUS_HEADER.to_string(),
383 BYBIT_RATE_LIMIT_RESET_HEADER.to_string(),
384 ])
385 .timeout_secs(timeout_secs)
386 .maybe_proxy_url(proxy_url)
387 .rate_limiters(Vec::new())
388 .build()
389 .map_err(|e| BybitHttpError::NetworkError(format!("Failed to create HTTP client: {e}")))
390 }
391
392 fn refresh_http_session(&self, generation: u64) -> Result<(), BybitHttpError> {
393 let current = self.session_generation.load(Ordering::Acquire);
394 if current == generation {
395 return Ok(());
396 }
397
398 let client = Self::build_http_client(self.timeout_secs, self.proxy_url.clone())?;
399 self.client.store(Arc::new(client));
400 self.session_generation.store(generation, Ordering::Release);
401 Ok(())
402 }
403
404 fn request_rate_limit(
405 endpoint: &str,
406 payload: Option<&str>,
407 ) -> (Option<BybitProductType>, u32) {
408 let category = category_from_payload(payload);
409 let weight = if endpoint.ends_with("-batch") {
410 let order_count = payload
411 .and_then(|payload| serde_json::from_str::<serde_json::Value>(payload).ok())
412 .and_then(|value| value.get("request")?.as_array().map(Vec::len))
413 .unwrap_or(1);
414 category.map_or(1, |category| batch_weight(category, order_count))
415 } else {
416 1
417 };
418 (category, weight)
419 }
420
421 fn observe_rate_limit_headers(
422 &self,
423 endpoint: &str,
424 category: Option<BybitProductType>,
425 headers: &HashMap<String, String>,
426 ) {
427 let Some(limit) = headers
428 .get(BYBIT_RATE_LIMIT_HEADER)
429 .and_then(|value| value.parse::<u32>().ok())
430 else {
431 return;
432 };
433 let Some(remaining) = headers
434 .get(BYBIT_RATE_LIMIT_STATUS_HEADER)
435 .and_then(|value| value.parse::<u32>().ok())
436 else {
437 return;
438 };
439 let reset_timestamp_ms = headers
440 .get(BYBIT_RATE_LIMIT_RESET_HEADER)
441 .and_then(|value| value.parse::<i64>().ok());
442 self.rate_limiter
443 .observe_account(endpoint, category, limit, remaining, reset_timestamp_ms);
444 }
445
446 fn is_rate_limit_403(status: u16, body: &str) -> bool {
447 status == 403 && body.to_ascii_lowercase().contains("access too frequent")
448 }
449
450 fn sign_request(
451 &self,
452 timestamp: &str,
453 params: Option<&str>,
454 ) -> Result<HashMap<String, String>, BybitHttpError> {
455 let credential = self
456 .credential
457 .as_ref()
458 .ok_or(BybitHttpError::MissingCredentials)?;
459
460 let signature = credential.sign_with_payload(timestamp, self.recv_window_ms, params);
461
462 let mut headers = HashMap::new();
463 headers.insert(
464 "X-BAPI-API-KEY".to_string(),
465 credential.api_key().to_string(),
466 );
467 headers.insert("X-BAPI-TIMESTAMP".to_string(), timestamp.to_string());
468 headers.insert("X-BAPI-SIGN".to_string(), signature);
469 headers.insert(
470 "X-BAPI-RECV-WINDOW".to_string(),
471 self.recv_window_ms.to_string(),
472 );
473
474 Ok(headers)
475 }
476
477 async fn send_request<T: DeserializeOwned + BybitResponseCheck, P: Serialize>(
478 &self,
479 method: Method,
480 endpoint: &str,
481 params: Option<&P>,
482 body: Option<Vec<u8>>,
483 authenticate: bool,
484 ) -> Result<T, BybitHttpError> {
485 let endpoint = endpoint.to_string();
486 let url = format!("{}{endpoint}", self.base_url);
487 let method_clone = method.clone();
488 let body_clone = body.clone();
489
490 let params_str = if method == Method::GET {
492 params
493 .map(serde_urlencoded::to_string)
494 .transpose()
495 .map_err(|e| {
496 BybitHttpError::JsonError(format!("Failed to serialize params: {e}"))
497 })?
498 } else {
499 None
500 };
501
502 let operation = || {
503 let url = url.clone();
504 let method = method_clone.clone();
505 let body = body_clone.clone();
506 let endpoint = endpoint.clone();
507 let params_str = params_str.clone();
508
509 async move {
510 let full_url = if let Some(ref query) = params_str {
511 if query.is_empty() {
512 url
513 } else {
514 format!("{url}?{query}")
515 }
516 } else {
517 url
518 };
519
520 let sign_payload = if method == Method::GET {
521 params_str.as_deref()
522 } else {
523 body.as_ref()
524 .and_then(|body| std::str::from_utf8(body).ok())
525 };
526 let (category, weight) = Self::request_rate_limit(&endpoint, sign_payload);
527
528 let generation = self.rate_limiter.http_session_generation();
529 self.refresh_http_session(generation)?;
530 self.rate_limiter
531 .acquire_http(&endpoint, category, weight, authenticate)
532 .await
533 .map_err(BybitHttpError::ValidationError)?;
534 let generation = self.rate_limiter.http_session_generation();
535 self.refresh_http_session(generation)?;
536
537 let mut headers = Self::default_headers();
538
539 if authenticate {
540 let timestamp = get_atomic_clock_realtime().get_time_ms().to_string();
541 let auth_headers = self.sign_request(×tamp, sign_payload)?;
542 headers.extend(auth_headers);
543 }
544
545 if method == Method::POST || method == Method::PUT {
546 headers.insert("Content-Type".to_string(), "application/json".to_string());
547 }
548
549 let response = self
550 .client
551 .load()
552 .request(method, full_url, None, Some(headers), body, None, None)
553 .await?;
554
555 self.observe_rate_limit_headers(&endpoint, category, &response.headers);
556
557 if response.status.as_u16() >= 400 {
558 let body = String::from_utf8_lossy(&response.body).to_string();
559 if Self::is_rate_limit_403(response.status.as_u16(), &body) {
560 let generation = self.rate_limiter.reset_http_sessions();
561 self.refresh_http_session(generation)?;
562 }
563 return Err(BybitHttpError::UnexpectedStatus {
564 status: response.status.as_u16(),
565 body,
566 });
567 }
568
569 match serde_json::from_slice::<T>(&response.body) {
571 Ok(result) => {
572 if result.ret_code() != 0 {
574 return Err(BybitHttpError::BybitError {
575 error_code: result.ret_code() as i32,
576 message: result.ret_msg().to_string(),
577 });
578 }
579 Ok(result)
580 }
581 Err(json_err) => {
582 if let Ok(error_check) =
585 serde_json::from_slice::<BybitErrorCheck>(&response.body)
586 && error_check.ret_code != 0
587 {
588 return Err(BybitHttpError::BybitError {
589 error_code: error_check.ret_code as i32,
590 message: error_check.ret_msg,
591 });
592 }
593 Err(json_err.into())
595 }
596 }
597 }
598 };
599
600 let create_error = |error: RetryError| -> BybitHttpError {
601 match error {
602 RetryError::Canceled => {
603 BybitHttpError::Canceled("Adapter disconnecting or shutting down".to_string())
604 }
605 error => BybitHttpError::NetworkError(error.to_string()),
606 }
607 };
608
609 let token = self.cancellation_token();
610
611 self.retry_manager
612 .execute_with_retry_with_cancel(
613 endpoint.as_str(),
614 operation,
615 should_retry_http,
616 create_error,
617 &token,
618 )
619 .await
620 }
621
622 #[cfg(test)]
623 fn build_path<S: Serialize>(base: &str, params: &S) -> Result<String, BybitHttpError> {
624 let query = serde_urlencoded::to_string(params)
625 .map_err(|e| BybitHttpError::JsonError(e.to_string()))?;
626
627 if query.is_empty() {
628 Ok(base.to_owned())
629 } else {
630 Ok(format!("{base}?{query}"))
631 }
632 }
633
634 pub async fn get_server_time(&self) -> Result<BybitServerTimeResponse, BybitHttpError> {
644 self.send_request::<_, ()>(Method::GET, "/v5/market/time", None, None, false)
645 .await
646 }
647
648 pub async fn get_instruments<T: DeserializeOwned + BybitResponseCheck>(
658 &self,
659 params: &BybitInstrumentsInfoParams,
660 ) -> Result<T, BybitHttpError> {
661 self.send_request(
662 Method::GET,
663 "/v5/market/instruments-info",
664 Some(params),
665 None,
666 false,
667 )
668 .await
669 }
670
671 pub async fn get_instruments_spot(
681 &self,
682 params: &BybitInstrumentsInfoParams,
683 ) -> Result<BybitInstrumentSpotResponse, BybitHttpError> {
684 self.get_instruments(params).await
685 }
686
687 pub async fn get_instruments_linear(
697 &self,
698 params: &BybitInstrumentsInfoParams,
699 ) -> Result<BybitInstrumentLinearResponse, BybitHttpError> {
700 self.get_instruments(params).await
701 }
702
703 pub async fn get_instruments_inverse(
713 &self,
714 params: &BybitInstrumentsInfoParams,
715 ) -> Result<BybitInstrumentInverseResponse, BybitHttpError> {
716 self.get_instruments(params).await
717 }
718
719 pub async fn get_instruments_option(
729 &self,
730 params: &BybitInstrumentsInfoParams,
731 ) -> Result<BybitInstrumentOptionResponse, BybitHttpError> {
732 self.get_instruments(params).await
733 }
734
735 pub async fn get_klines(
745 &self,
746 params: &BybitKlinesParams,
747 ) -> Result<BybitKlinesResponse, BybitHttpError> {
748 self.send_request(Method::GET, "/v5/market/kline", Some(params), None, false)
749 .await
750 }
751
752 pub async fn get_recent_trades(
762 &self,
763 params: &BybitTradesParams,
764 ) -> Result<BybitTradesResponse, BybitHttpError> {
765 self.send_request(
766 Method::GET,
767 "/v5/market/recent-trade",
768 Some(params),
769 None,
770 false,
771 )
772 .await
773 }
774
775 pub async fn get_funding_history(
785 &self,
786 params: &BybitFundingParams,
787 ) -> Result<BybitFundingResponse, BybitHttpError> {
788 self.send_request(
789 Method::GET,
790 "/v5/market/funding/history",
791 Some(params),
792 None,
793 false,
794 )
795 .await
796 }
797
798 pub async fn get_orderbook(
808 &self,
809 params: &BybitOrderbookParams,
810 ) -> Result<BybitOrderbookResponse, BybitHttpError> {
811 self.send_request(
812 Method::GET,
813 "/v5/market/orderbook",
814 Some(params),
815 None,
816 false,
817 )
818 .await
819 }
820
821 #[expect(clippy::too_many_arguments)]
835 pub async fn get_open_orders(
836 &self,
837 category: BybitProductType,
838 symbol: Option<String>,
839 base_coin: Option<String>,
840 settle_coin: Option<String>,
841 order_id: Option<String>,
842 order_link_id: Option<String>,
843 open_only: Option<BybitOpenOnly>,
844 order_filter: Option<BybitOrderFilter>,
845 limit: Option<u32>,
846 cursor: Option<String>,
847 ) -> Result<BybitOpenOrdersResponse, BybitHttpError> {
848 let mut builder = BybitOpenOrdersParamsBuilder::default();
849 builder.category(category);
850
851 if let Some(s) = symbol {
852 builder.symbol(s);
853 }
854
855 if let Some(bc) = base_coin {
856 builder.base_coin(bc);
857 }
858
859 if let Some(sc) = settle_coin {
860 builder.settle_coin(sc);
861 }
862
863 if let Some(oi) = order_id {
864 builder.order_id(oi);
865 }
866
867 if let Some(ol) = order_link_id {
868 builder.order_link_id(ol);
869 }
870
871 if let Some(oo) = open_only {
872 builder.open_only(oo);
873 }
874
875 if let Some(of) = order_filter {
876 builder.order_filter(of);
877 }
878
879 if let Some(l) = limit {
880 builder.limit(l);
881 }
882
883 if let Some(c) = cursor {
884 builder.cursor(c);
885 }
886
887 let params = builder
888 .build()
889 .expect("Failed to build BybitOpenOrdersParams");
890
891 self.send_request(Method::GET, BYBIT_ORDER_REALTIME, Some(¶ms), None, true)
892 .await
893 }
894
895 pub async fn place_order(
905 &self,
906 request: &serde_json::Value,
907 ) -> Result<BybitPlaceOrderResponse, BybitHttpError> {
908 let body = serde_json::to_vec(request)?;
909 self.send_request::<_, ()>(Method::POST, "/v5/order/create", None, Some(body), true)
910 .await
911 }
912
913 pub async fn get_wallet_balance(
923 &self,
924 params: &BybitWalletBalanceParams,
925 ) -> Result<BybitWalletBalanceResponse, BybitHttpError> {
926 self.send_request(
927 Method::GET,
928 "/v5/account/wallet-balance",
929 Some(params),
930 None,
931 true,
932 )
933 .await
934 }
935
936 pub async fn get_account_info(&self) -> Result<BybitAccountInfoResponse, BybitHttpError> {
946 self.send_request::<_, ()>(Method::GET, "/v5/account/info", None, None, true)
947 .await
948 }
949
950 pub async fn get_account_details(&self) -> Result<BybitAccountDetailsResponse, BybitHttpError> {
960 self.send_request::<_, ()>(Method::GET, "/v5/user/query-api", None, None, true)
961 .await
962 }
963
964 pub async fn update_sub_api_key(
974 &self,
975 params: &BybitUpdateSubApiParams,
976 ) -> Result<BybitUpdateSubApiResponse, BybitHttpError> {
977 let body = serde_json::to_vec(params)?;
978 self.send_request::<_, ()>(
979 Method::POST,
980 "/v5/user/update-sub-api",
981 None,
982 Some(body),
983 true,
984 )
985 .await
986 }
987
988 pub async fn update_master_api_key(
998 &self,
999 params: &BybitUpdateMasterApiParams,
1000 ) -> Result<BybitUpdateMasterApiResponse, BybitHttpError> {
1001 let body = serde_json::to_vec(params)?;
1002 self.send_request::<_, ()>(Method::POST, "/v5/user/update-api", None, Some(body), true)
1003 .await
1004 }
1005
1006 pub async fn get_sub_members(&self) -> Result<BybitSubMembersResponse, BybitHttpError> {
1016 self.send_request::<_, ()>(Method::GET, "/v5/user/query-sub-members", None, None, true)
1017 .await
1018 }
1019
1020 pub async fn get_sub_members_paged(
1030 &self,
1031 params: &BybitSubMembersPageParams,
1032 ) -> Result<BybitSubMembersPagedResponse, BybitHttpError> {
1033 self.send_request(Method::GET, "/v5/user/submembers", Some(params), None, true)
1034 .await
1035 }
1036
1037 pub async fn get_escrow_sub_members(
1047 &self,
1048 params: &BybitSubMembersPageParams,
1049 ) -> Result<BybitEscrowSubMembersResponse, BybitHttpError> {
1050 self.send_request(
1051 Method::GET,
1052 "/v5/user/escrow_sub_members",
1053 Some(params),
1054 None,
1055 true,
1056 )
1057 .await
1058 }
1059
1060 pub async fn get_sub_api_keys(
1070 &self,
1071 params: &BybitSubApiKeysParams,
1072 ) -> Result<BybitSubApiKeysResponse, BybitHttpError> {
1073 self.send_request(
1074 Method::GET,
1075 "/v5/user/sub-apikeys",
1076 Some(params),
1077 None,
1078 true,
1079 )
1080 .await
1081 }
1082
1083 pub async fn fetch_all_sub_members_paged(
1090 &self,
1091 page_size: Option<u32>,
1092 ) -> Result<Vec<BybitSubMember>, BybitHttpError> {
1093 let mut members = Vec::new();
1094 let mut cursor: Option<String> = None;
1095
1096 loop {
1097 let params = BybitSubMembersPageParams {
1098 page_size,
1099 next_cursor: cursor.take(),
1100 };
1101 let mut page = self.get_sub_members_paged(¶ms).await?;
1102 let next = page.result.continuation_cursor().map(str::to_owned);
1103 members.append(&mut page.result.sub_members);
1104
1105 match next {
1106 Some(c) => cursor = Some(c),
1107 None => break,
1108 }
1109 }
1110
1111 Ok(members)
1112 }
1113
1114 pub async fn fetch_all_escrow_sub_members(
1122 &self,
1123 page_size: Option<u32>,
1124 ) -> Result<Vec<BybitSubMember>, BybitHttpError> {
1125 let mut members = Vec::new();
1126 let mut cursor: Option<String> = None;
1127
1128 loop {
1129 let params = BybitSubMembersPageParams {
1130 page_size,
1131 next_cursor: cursor.take(),
1132 };
1133 let mut page = self.get_escrow_sub_members(¶ms).await?;
1134 let next = page.result.continuation_cursor().map(str::to_owned);
1135 members.append(&mut page.result.sub_members);
1136
1137 match next {
1138 Some(c) => cursor = Some(c),
1139 None => break,
1140 }
1141 }
1142
1143 Ok(members)
1144 }
1145
1146 pub async fn fetch_all_sub_api_keys(
1154 &self,
1155 sub_member_id: impl Into<String>,
1156 limit: Option<u32>,
1157 ) -> Result<Vec<BybitSubApiKeyInfo>, BybitHttpError> {
1158 let sub_member_id = sub_member_id.into();
1159 let mut keys = Vec::new();
1160 let mut cursor: Option<String> = None;
1161
1162 loop {
1163 let params = BybitSubApiKeysParams {
1164 sub_member_id: sub_member_id.clone(),
1165 limit,
1166 cursor: cursor.take(),
1167 };
1168 let mut page = self.get_sub_api_keys(¶ms).await?;
1169 let next = page.result.continuation_cursor().map(str::to_owned);
1170 keys.append(&mut page.result.keys);
1171
1172 match next {
1173 Some(c) => cursor = Some(c),
1174 None => break,
1175 }
1176 }
1177
1178 Ok(keys)
1179 }
1180
1181 pub async fn get_fee_rate(
1191 &self,
1192 params: &BybitFeeRateParams,
1193 ) -> Result<BybitFeeRateResponse, BybitHttpError> {
1194 self.send_request(
1195 Method::GET,
1196 "/v5/account/fee-rate",
1197 Some(params),
1198 None,
1199 true,
1200 )
1201 .await
1202 }
1203
1204 pub async fn set_margin_mode(
1221 &self,
1222 margin_mode: BybitMarginMode,
1223 ) -> Result<BybitSetMarginModeResponse, BybitHttpError> {
1224 let params = BybitSetMarginModeParamsBuilder::default()
1225 .set_margin_mode(margin_mode)
1226 .build()
1227 .expect("Failed to build BybitSetMarginModeParams");
1228
1229 let body = serde_json::to_vec(¶ms)?;
1230 self.send_request::<_, ()>(
1231 Method::POST,
1232 "/v5/account/set-margin-mode",
1233 None,
1234 Some(body),
1235 true,
1236 )
1237 .await
1238 }
1239
1240 pub async fn set_leverage(
1257 &self,
1258 product_type: BybitProductType,
1259 symbol: &str,
1260 buy_leverage: &str,
1261 sell_leverage: &str,
1262 ) -> Result<BybitSetLeverageResponse, BybitHttpError> {
1263 let params = BybitSetLeverageParamsBuilder::default()
1264 .category(product_type)
1265 .symbol(symbol.to_string())
1266 .buy_leverage(buy_leverage.to_string())
1267 .sell_leverage(sell_leverage.to_string())
1268 .build()
1269 .expect("Failed to build BybitSetLeverageParams");
1270
1271 let body = serde_json::to_vec(¶ms)?;
1272 self.send_request::<_, ()>(
1273 Method::POST,
1274 "/v5/position/set-leverage",
1275 None,
1276 Some(body),
1277 true,
1278 )
1279 .await
1280 }
1281
1282 pub async fn switch_mode(
1299 &self,
1300 product_type: BybitProductType,
1301 mode: BybitPositionMode,
1302 symbol: Option<String>,
1303 coin: Option<String>,
1304 ) -> Result<BybitSwitchModeResponse, BybitHttpError> {
1305 let mut builder = BybitSwitchModeParamsBuilder::default();
1306 builder.category(product_type);
1307 builder.mode(mode);
1308
1309 if let Some(s) = symbol {
1310 builder.symbol(s);
1311 }
1312
1313 if let Some(c) = coin {
1314 builder.coin(c);
1315 }
1316
1317 let params = builder
1318 .build()
1319 .expect("Failed to build BybitSwitchModeParams");
1320
1321 let body = serde_json::to_vec(¶ms)?;
1322 self.send_request::<_, ()>(
1323 Method::POST,
1324 "/v5/position/switch-mode",
1325 None,
1326 Some(body),
1327 true,
1328 )
1329 .await
1330 }
1331
1332 pub async fn set_trading_stop(
1345 &self,
1346 params: &BybitSetTradingStopParams,
1347 ) -> Result<BybitSetTradingStopResponse, BybitHttpError> {
1348 let body = serde_json::to_vec(params)?;
1349 self.send_request::<_, ()>(
1350 Method::POST,
1351 "/v5/position/trading-stop",
1352 None,
1353 Some(body),
1354 true,
1355 )
1356 .await
1357 }
1358
1359 pub async fn borrow(
1376 &self,
1377 coin: &str,
1378 amount: &str,
1379 ) -> Result<BybitBorrowResponse, BybitHttpError> {
1380 let params = BybitBorrowParamsBuilder::default()
1381 .coin(coin.to_string())
1382 .amount(amount.to_string())
1383 .build()
1384 .expect("Failed to build BybitBorrowParams");
1385
1386 let body = serde_json::to_vec(¶ms)?;
1387 self.send_request::<_, ()>(Method::POST, "/v5/account/borrow", None, Some(body), true)
1388 .await
1389 }
1390
1391 pub async fn no_convert_repay(
1409 &self,
1410 coin: &str,
1411 amount: Option<&str>,
1412 ) -> Result<BybitNoConvertRepayResponse, BybitHttpError> {
1413 let mut builder = BybitNoConvertRepayParamsBuilder::default();
1414 builder.coin(coin.to_string());
1415
1416 if let Some(amt) = amount {
1417 builder.amount(amt.to_string());
1418 }
1419
1420 let params = builder
1421 .build()
1422 .expect("Failed to build BybitNoConvertRepayParams");
1423
1424 if let Ok(params_json) = serde_json::to_string(¶ms) {
1425 log::debug!("Repay request params: {params_json}");
1426 }
1427
1428 let body = serde_json::to_vec(¶ms)?;
1429 let result = self
1430 .send_request::<_, ()>(
1431 Method::POST,
1432 "/v5/account/no-convert-repay",
1433 None,
1434 Some(body),
1435 true,
1436 )
1437 .await;
1438
1439 if let Err(ref e) = result
1440 && let Ok(params_json) = serde_json::to_string(¶ms)
1441 {
1442 log::error!("Repay request failed with params {params_json}: {e}");
1443 }
1444
1445 result
1446 }
1447
1448 pub async fn repay(
1466 &self,
1467 coin: Option<&str>,
1468 amount: Option<&str>,
1469 ) -> Result<BybitRepayResponse, BybitHttpError> {
1470 let mut builder = BybitRepayParamsBuilder::default();
1471
1472 if let Some(coin) = coin {
1473 builder.coin(coin.to_string());
1474 }
1475
1476 if let Some(amt) = amount {
1477 builder.amount(amt.to_string());
1478 }
1479
1480 let params = builder.build().expect("Failed to build BybitRepayParams");
1481
1482 if let Ok(params_json) = serde_json::to_string(¶ms) {
1483 log::debug!("Repay request params: {params_json}");
1484 }
1485
1486 let body = serde_json::to_vec(¶ms)?;
1487 let result = self
1488 .send_request::<_, ()>(Method::POST, "/v5/account/repay", None, Some(body), true)
1489 .await;
1490
1491 if let Err(ref e) = result
1492 && let Ok(params_json) = serde_json::to_string(¶ms)
1493 {
1494 log::error!("Repay request failed with params {params_json}: {e}");
1495 }
1496
1497 result
1498 }
1499
1500 pub async fn get_tickers<T: DeserializeOwned + BybitResponseCheck>(
1510 &self,
1511 params: &BybitTickersParams,
1512 ) -> Result<T, BybitHttpError> {
1513 self.send_request(Method::GET, "/v5/market/tickers", Some(params), None, false)
1514 .await
1515 }
1516
1517 pub async fn get_trade_history(
1527 &self,
1528 params: &BybitTradeHistoryParams,
1529 ) -> Result<BybitTradeHistoryResponse, BybitHttpError> {
1530 self.send_request(Method::GET, "/v5/execution/list", Some(params), None, true)
1531 .await
1532 }
1533
1534 pub async fn get_positions(
1547 &self,
1548 params: &BybitPositionListParams,
1549 ) -> Result<BybitPositionListResponse, BybitHttpError> {
1550 self.send_request(Method::GET, "/v5/position/list", Some(params), None, true)
1551 .await
1552 }
1553
1554 #[must_use]
1556 pub fn base_url(&self) -> &str {
1557 &self.base_url
1558 }
1559
1560 #[must_use]
1562 pub fn recv_window_ms(&self) -> u64 {
1563 self.recv_window_ms
1564 }
1565
1566 #[must_use]
1568 pub fn credential(&self) -> Option<&Credential> {
1569 self.credential.as_ref()
1570 }
1571}
1572
1573#[cfg_attr(
1575 feature = "python",
1576 pyo3::pyclass(module = "nautilus_trader.adapters.bybit", from_py_object)
1577)]
1578#[cfg_attr(
1579 feature = "python",
1580 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.bybit")
1581)]
1582pub struct BybitHttpClient {
1587 pub(crate) inner: Arc<BybitRawHttpClient>,
1588 pub(crate) instruments_cache: Arc<AtomicMap<Ustr, InstrumentAny>>,
1589 clock: &'static AtomicTime,
1590 cache_initialized: Arc<AtomicBool>,
1591 use_spot_position_reports: Arc<AtomicBool>,
1592}
1593
1594impl Clone for BybitHttpClient {
1595 fn clone(&self) -> Self {
1596 Self {
1597 inner: self.inner.clone(),
1598 instruments_cache: self.instruments_cache.clone(),
1599 cache_initialized: self.cache_initialized.clone(),
1600 use_spot_position_reports: self.use_spot_position_reports.clone(),
1601 clock: self.clock,
1602 }
1603 }
1604}
1605
1606impl Default for BybitHttpClient {
1607 fn default() -> Self {
1608 Self::new(None, 60, 3, 1000, 10_000, DEFAULT_RECV_WINDOW_MS, None)
1609 .expect("Failed to create default BybitHttpClient")
1610 }
1611}
1612
1613impl Debug for BybitHttpClient {
1614 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1615 f.debug_struct(stringify!(BybitHttpClient))
1616 .field("inner", &self.inner)
1617 .finish()
1618 }
1619}
1620
1621impl BybitHttpClient {
1622 pub fn new(
1628 base_url: Option<String>,
1629 timeout_secs: u64,
1630 max_retries: u32,
1631 retry_delay_ms: u64,
1632 retry_delay_max_ms: u64,
1633 recv_window_ms: u64,
1634 proxy_url: Option<String>,
1635 ) -> Result<Self, BybitHttpError> {
1636 Ok(Self {
1637 inner: Arc::new(BybitRawHttpClient::new(
1638 base_url,
1639 timeout_secs,
1640 max_retries,
1641 retry_delay_ms,
1642 retry_delay_max_ms,
1643 recv_window_ms,
1644 proxy_url,
1645 )?),
1646 instruments_cache: Arc::new(AtomicMap::new()),
1647 cache_initialized: Arc::new(AtomicBool::new(false)),
1648 use_spot_position_reports: Arc::new(AtomicBool::new(false)),
1649 clock: get_atomic_clock_realtime(),
1650 })
1651 }
1652
1653 #[expect(clippy::too_many_arguments)]
1659 pub fn with_credentials(
1660 api_key: String,
1661 api_secret: String,
1662 base_url: Option<String>,
1663 timeout_secs: u64,
1664 max_retries: u32,
1665 retry_delay_ms: u64,
1666 retry_delay_max_ms: u64,
1667 recv_window_ms: u64,
1668 proxy_url: Option<String>,
1669 ) -> Result<Self, BybitHttpError> {
1670 Ok(Self {
1671 inner: Arc::new(BybitRawHttpClient::with_credentials(
1672 api_key,
1673 api_secret,
1674 base_url,
1675 timeout_secs,
1676 max_retries,
1677 retry_delay_ms,
1678 retry_delay_max_ms,
1679 recv_window_ms,
1680 proxy_url,
1681 )?),
1682 instruments_cache: Arc::new(AtomicMap::new()),
1683 cache_initialized: Arc::new(AtomicBool::new(false)),
1684 use_spot_position_reports: Arc::new(AtomicBool::new(false)),
1685 clock: get_atomic_clock_realtime(),
1686 })
1687 }
1688
1689 #[expect(clippy::too_many_arguments)]
1702 pub fn new_with_env(
1703 api_key: Option<String>,
1704 api_secret: Option<String>,
1705 base_url: Option<String>,
1706 demo: bool,
1707 testnet: bool,
1708 timeout_secs: u64,
1709 max_retries: u32,
1710 retry_delay_ms: u64,
1711 retry_delay_max_ms: u64,
1712 recv_window_ms: u64,
1713 proxy_url: Option<String>,
1714 ) -> Result<Self, BybitHttpError> {
1715 Ok(Self {
1716 inner: Arc::new(BybitRawHttpClient::new_with_env(
1717 api_key,
1718 api_secret,
1719 base_url,
1720 demo,
1721 testnet,
1722 timeout_secs,
1723 max_retries,
1724 retry_delay_ms,
1725 retry_delay_max_ms,
1726 recv_window_ms,
1727 proxy_url,
1728 )?),
1729 instruments_cache: Arc::new(AtomicMap::new()),
1730 cache_initialized: Arc::new(AtomicBool::new(false)),
1731 use_spot_position_reports: Arc::new(AtomicBool::new(false)),
1732 clock: get_atomic_clock_realtime(),
1733 })
1734 }
1735
1736 #[must_use]
1737 pub fn base_url(&self) -> &str {
1738 self.inner.base_url()
1739 }
1740
1741 #[must_use]
1742 pub fn recv_window_ms(&self) -> u64 {
1743 self.inner.recv_window_ms()
1744 }
1745
1746 #[must_use]
1747 pub fn credential(&self) -> Option<&Credential> {
1748 self.inner.credential()
1749 }
1750
1751 pub fn set_use_spot_position_reports(&self, use_spot_position_reports: bool) {
1752 self.use_spot_position_reports
1753 .store(use_spot_position_reports, Ordering::Relaxed);
1754 }
1755
1756 pub fn cancel_all_requests(&self) {
1757 self.inner.cancel_all_requests();
1758 }
1759
1760 pub fn reset_cancellation_token(&self) {
1761 self.inner.reset_cancellation_token();
1762 }
1763
1764 pub fn cancellation_token(&self) -> CancellationToken {
1765 self.inner.cancellation_token()
1766 }
1767
1768 pub fn cache_instrument(&self, instrument: InstrumentAny) {
1770 self.instruments_cache
1771 .insert(instrument.symbol().inner(), instrument);
1772 self.cache_initialized.store(true, Ordering::Release);
1773 }
1774
1775 pub fn cache_instruments(&self, instruments: &[InstrumentAny]) {
1777 self.instruments_cache.rcu(|m| {
1778 for instrument in instruments {
1779 m.insert(instrument.symbol().inner(), instrument.clone());
1780 }
1781 });
1782 self.cache_initialized.store(true, Ordering::Release);
1783 }
1784
1785 pub fn get_instrument(&self, symbol: &Ustr) -> Option<InstrumentAny> {
1786 self.instruments_cache.get_cloned(symbol)
1787 }
1788
1789 fn instrument_from_cache(&self, symbol: &Symbol) -> anyhow::Result<InstrumentAny> {
1790 self.get_instrument(&symbol.inner()).ok_or_else(|| {
1791 anyhow::anyhow!(
1792 "Instrument {symbol} not found in cache, ensure instruments loaded first"
1793 )
1794 })
1795 }
1796
1797 #[must_use]
1798 fn generate_ts_init(&self) -> UnixNanos {
1799 self.clock.get_time_ns()
1800 }
1801
1802 pub async fn get_server_time(&self) -> Result<BybitServerTimeResponse, BybitHttpError> {
1814 self.inner.get_server_time().await
1815 }
1816
1817 pub async fn get_instruments<T: DeserializeOwned + BybitResponseCheck>(
1829 &self,
1830 params: &BybitInstrumentsInfoParams,
1831 ) -> Result<T, BybitHttpError> {
1832 self.inner.get_instruments(params).await
1833 }
1834
1835 pub async fn get_instruments_spot(
1847 &self,
1848 params: &BybitInstrumentsInfoParams,
1849 ) -> Result<BybitInstrumentSpotResponse, BybitHttpError> {
1850 self.inner.get_instruments_spot(params).await
1851 }
1852
1853 pub async fn get_instruments_linear(
1865 &self,
1866 params: &BybitInstrumentsInfoParams,
1867 ) -> Result<BybitInstrumentLinearResponse, BybitHttpError> {
1868 self.inner.get_instruments_linear(params).await
1869 }
1870
1871 pub async fn get_instruments_inverse(
1883 &self,
1884 params: &BybitInstrumentsInfoParams,
1885 ) -> Result<BybitInstrumentInverseResponse, BybitHttpError> {
1886 self.inner.get_instruments_inverse(params).await
1887 }
1888
1889 pub async fn get_instruments_option(
1901 &self,
1902 params: &BybitInstrumentsInfoParams,
1903 ) -> Result<BybitInstrumentOptionResponse, BybitHttpError> {
1904 self.inner.get_instruments_option(params).await
1905 }
1906
1907 pub async fn get_klines(
1919 &self,
1920 params: &BybitKlinesParams,
1921 ) -> Result<BybitKlinesResponse, BybitHttpError> {
1922 self.inner.get_klines(params).await
1923 }
1924
1925 pub async fn get_recent_trades(
1937 &self,
1938 params: &BybitTradesParams,
1939 ) -> Result<BybitTradesResponse, BybitHttpError> {
1940 self.inner.get_recent_trades(params).await
1941 }
1942
1943 #[expect(clippy::too_many_arguments)]
1955 pub async fn get_open_orders(
1956 &self,
1957 category: BybitProductType,
1958 symbol: Option<String>,
1959 base_coin: Option<String>,
1960 settle_coin: Option<String>,
1961 order_id: Option<String>,
1962 order_link_id: Option<String>,
1963 open_only: Option<BybitOpenOnly>,
1964 order_filter: Option<BybitOrderFilter>,
1965 limit: Option<u32>,
1966 cursor: Option<String>,
1967 ) -> Result<BybitOpenOrdersResponse, BybitHttpError> {
1968 self.inner
1969 .get_open_orders(
1970 category,
1971 symbol,
1972 base_coin,
1973 settle_coin,
1974 order_id,
1975 order_link_id,
1976 open_only,
1977 order_filter,
1978 limit,
1979 cursor,
1980 )
1981 .await
1982 }
1983
1984 pub async fn place_order(
1996 &self,
1997 request: &serde_json::Value,
1998 ) -> Result<BybitPlaceOrderResponse, BybitHttpError> {
1999 self.inner.place_order(request).await
2000 }
2001
2002 pub async fn get_wallet_balance(
2014 &self,
2015 params: &BybitWalletBalanceParams,
2016 ) -> Result<BybitWalletBalanceResponse, BybitHttpError> {
2017 self.inner.get_wallet_balance(params).await
2018 }
2019
2020 pub async fn get_account_info(&self) -> Result<BybitAccountInfoResponse, BybitHttpError> {
2032 self.inner.get_account_info().await
2033 }
2034
2035 pub async fn get_account_details(&self) -> Result<BybitAccountDetailsResponse, BybitHttpError> {
2047 self.inner.get_account_details().await
2048 }
2049
2050 pub async fn update_sub_api_key(
2062 &self,
2063 params: &BybitUpdateSubApiParams,
2064 ) -> Result<BybitUpdateSubApiResponse, BybitHttpError> {
2065 self.inner.update_sub_api_key(params).await
2066 }
2067
2068 pub async fn update_master_api_key(
2080 &self,
2081 params: &BybitUpdateMasterApiParams,
2082 ) -> Result<BybitUpdateMasterApiResponse, BybitHttpError> {
2083 self.inner.update_master_api_key(params).await
2084 }
2085
2086 pub async fn get_sub_members(&self) -> Result<BybitSubMembersResponse, BybitHttpError> {
2098 self.inner.get_sub_members().await
2099 }
2100
2101 pub async fn get_sub_members_paged(
2113 &self,
2114 params: &BybitSubMembersPageParams,
2115 ) -> Result<BybitSubMembersPagedResponse, BybitHttpError> {
2116 self.inner.get_sub_members_paged(params).await
2117 }
2118
2119 pub async fn get_escrow_sub_members(
2131 &self,
2132 params: &BybitSubMembersPageParams,
2133 ) -> Result<BybitEscrowSubMembersResponse, BybitHttpError> {
2134 self.inner.get_escrow_sub_members(params).await
2135 }
2136
2137 pub async fn get_sub_api_keys(
2149 &self,
2150 params: &BybitSubApiKeysParams,
2151 ) -> Result<BybitSubApiKeysResponse, BybitHttpError> {
2152 self.inner.get_sub_api_keys(params).await
2153 }
2154
2155 pub async fn get_positions(
2168 &self,
2169 params: &BybitPositionListParams,
2170 ) -> Result<BybitPositionListResponse, BybitHttpError> {
2171 self.inner.get_positions(params).await
2172 }
2173
2174 pub async fn get_fee_rate(
2187 &self,
2188 params: &BybitFeeRateParams,
2189 ) -> Result<BybitFeeRateResponse, BybitHttpError> {
2190 self.inner.get_fee_rate(params).await
2191 }
2192
2193 pub async fn set_margin_mode(
2206 &self,
2207 margin_mode: BybitMarginMode,
2208 ) -> Result<BybitSetMarginModeResponse, BybitHttpError> {
2209 self.inner.set_margin_mode(margin_mode).await
2210 }
2211
2212 pub async fn set_leverage(
2225 &self,
2226 product_type: BybitProductType,
2227 symbol: &str,
2228 buy_leverage: &str,
2229 sell_leverage: &str,
2230 ) -> Result<BybitSetLeverageResponse, BybitHttpError> {
2231 self.inner
2232 .set_leverage(product_type, symbol, buy_leverage, sell_leverage)
2233 .await
2234 }
2235
2236 pub async fn switch_mode(
2249 &self,
2250 product_type: BybitProductType,
2251 mode: BybitPositionMode,
2252 symbol: Option<String>,
2253 coin: Option<String>,
2254 ) -> Result<BybitSwitchModeResponse, BybitHttpError> {
2255 self.inner
2256 .switch_mode(product_type, mode, symbol, coin)
2257 .await
2258 }
2259
2260 pub async fn set_trading_stop(
2273 &self,
2274 params: &BybitSetTradingStopParams,
2275 ) -> Result<BybitSetTradingStopResponse, BybitHttpError> {
2276 self.inner.set_trading_stop(params).await
2277 }
2278
2279 pub async fn get_spot_borrow_amount(&self, coin: &str) -> anyhow::Result<Decimal> {
2294 let params = BybitWalletBalanceParams {
2295 account_type: BybitAccountType::Unified,
2296 coin: Some(coin.to_string()),
2297 };
2298
2299 let response = self.inner.get_wallet_balance(¶ms).await?;
2300
2301 let borrow_amount = response
2302 .result
2303 .list
2304 .first()
2305 .and_then(|wallet| wallet.coin.iter().find(|c| c.coin.as_str() == coin))
2306 .map_or(Decimal::ZERO, |balance| balance.spot_borrow);
2307
2308 Ok(borrow_amount)
2309 }
2310
2311 pub async fn borrow_spot(
2327 &self,
2328 coin: &str,
2329 amount: Quantity,
2330 ) -> anyhow::Result<BybitBorrowResponse> {
2331 let amount_str = amount.to_string();
2332 self.inner
2333 .borrow(coin, &amount_str)
2334 .await
2335 .map_err(|e| anyhow::anyhow!("Failed to borrow {amount} {coin}: {e}"))
2336 }
2337
2338 pub async fn repay_spot_borrow(
2355 &self,
2356 coin: &str,
2357 amount: Option<Quantity>,
2358 ) -> anyhow::Result<BybitNoConvertRepayResponse> {
2359 let amount_str = amount.as_ref().map(|q| q.to_string());
2360 let response = self
2361 .inner
2362 .no_convert_repay(coin, amount_str.as_deref())
2363 .await
2364 .map_err(|e| anyhow::anyhow!("Failed to repay spot borrow for {coin}: {e}"))?;
2365 Self::ensure_repay_accepted(coin, response.result.result_status)?;
2366 Ok(response)
2367 }
2368
2369 pub async fn repay_spot_borrow_with_conversion(
2387 &self,
2388 coin: &str,
2389 amount: Option<Quantity>,
2390 ) -> anyhow::Result<BybitRepayResponse> {
2391 let amount_str = amount.as_ref().map(|q| q.to_string());
2392 let response = self
2393 .inner
2394 .repay(Some(coin), amount_str.as_deref())
2395 .await
2396 .map_err(|e| {
2397 anyhow::anyhow!("Failed to repay spot borrow (with conversion) for {coin}: {e}")
2398 })?;
2399 Self::ensure_repay_accepted(coin, response.result.result_status)?;
2400 Ok(response)
2401 }
2402
2403 fn ensure_repay_accepted(coin: &str, status: BybitRepayStatus) -> anyhow::Result<()> {
2404 anyhow::ensure!(
2405 status != BybitRepayStatus::Failed,
2406 "Bybit repay for {coin} returned result status {status}"
2407 );
2408 Ok(())
2409 }
2410
2411 async fn generate_spot_position_reports_from_wallet(
2419 &self,
2420 account_id: AccountId,
2421 instrument_id: InstrumentId,
2422 ) -> anyhow::Result<Vec<PositionStatusReport>> {
2423 let params = BybitWalletBalanceParams {
2424 account_type: BybitAccountType::Unified,
2425 coin: None,
2426 };
2427
2428 let response = self.inner.get_wallet_balance(¶ms).await?;
2429 let ts_init = self.generate_ts_init();
2430
2431 let mut wallet_by_coin: HashMap<Ustr, Decimal> = HashMap::new();
2432
2433 for wallet in &response.result.list {
2434 for coin_balance in &wallet.coin {
2435 let balance = coin_balance.wallet_balance - coin_balance.spot_borrow;
2436 *wallet_by_coin
2437 .entry(coin_balance.coin)
2438 .or_insert(Decimal::ZERO) += balance;
2439 }
2440 }
2441
2442 let mut reports = Vec::new();
2443
2444 if let Some(instrument) = self
2445 .instruments_cache
2446 .get_cloned(&instrument_id.symbol.inner())
2447 {
2448 let base_currency = instrument
2449 .base_currency()
2450 .expect("SPOT instrument should have base currency");
2451 let coin = base_currency.code;
2452 let wallet_balance = wallet_by_coin.get(&coin).copied().unwrap_or(Decimal::ZERO);
2453
2454 let side = if wallet_balance > Decimal::ZERO {
2455 PositionSide::Long
2456 } else if wallet_balance < Decimal::ZERO {
2457 PositionSide::Short
2458 } else {
2459 PositionSide::Flat
2460 };
2461
2462 let abs_balance = wallet_balance.abs();
2463 let quantity = Quantity::from_decimal_dp(abs_balance, instrument.size_precision())?;
2464
2465 let report = PositionStatusReport::new(
2466 account_id,
2467 instrument_id,
2468 side,
2469 quantity,
2470 ts_init,
2471 ts_init,
2472 None,
2473 None,
2474 None,
2475 );
2476
2477 reports.push(report);
2478 }
2479
2480 Ok(reports)
2481 }
2482
2483 #[expect(clippy::too_many_arguments)]
2494 pub async fn submit_order(
2495 &self,
2496 account_id: AccountId,
2497 product_type: BybitProductType,
2498 instrument_id: InstrumentId,
2499 client_order_id: ClientOrderId,
2500 order_side: OrderSide,
2501 order_type: OrderType,
2502 quantity: Quantity,
2503 time_in_force: Option<TimeInForce>,
2504 price: Option<Price>,
2505 trigger_price: Option<Price>,
2506 post_only: Option<bool>,
2507 reduce_only: bool,
2508 is_quote_quantity: bool,
2509 is_leverage: bool,
2510 position_idx: Option<BybitPositionIdx>,
2511 bbo_side_type: Option<BybitBboSideType>,
2512 bbo_level: Option<String>,
2513 native_tp_sl: Option<&BybitNativeTpSlParams>,
2514 ) -> anyhow::Result<OrderStatusReport> {
2515 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
2516 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2517
2518 let bybit_side = match order_side {
2519 OrderSide::Buy => BybitOrderSide::Buy,
2520 OrderSide::Sell => BybitOrderSide::Sell,
2521 };
2522
2523 let (bybit_order_type, is_stop_order) = match order_type {
2525 OrderType::Market => (BybitOrderType::Market, false),
2526 OrderType::Limit => (BybitOrderType::Limit, false),
2527 OrderType::StopMarket | OrderType::MarketIfTouched => (BybitOrderType::Market, true),
2528 OrderType::StopLimit | OrderType::LimitIfTouched => (BybitOrderType::Limit, true),
2529 _ => anyhow::bail!("Unsupported order type: {order_type:?}"),
2530 };
2531
2532 let bybit_tif = map_time_in_force(bybit_order_type, time_in_force, post_only)
2533 .map_err(|tif| anyhow::anyhow!("Unsupported time in force: {tif:?}"))?;
2534 let market_unit = spot_market_unit(product_type, bybit_order_type, is_quote_quantity);
2535 let trigger_dir = trigger_direction(order_type, order_side, is_stop_order);
2536
2537 let mut order_entry = BybitBatchPlaceOrderEntryBuilder::default();
2538 order_entry.symbol(bybit_symbol.raw_symbol().to_string());
2539 order_entry.side(bybit_side);
2540 order_entry.order_type(bybit_order_type);
2541 order_entry.qty(quantity.to_string());
2542 order_entry.time_in_force(bybit_tif);
2543 order_entry.order_link_id(client_order_id.to_string());
2544 order_entry.market_unit(market_unit);
2545 order_entry.trigger_direction(trigger_dir);
2546
2547 if bbo_side_type.is_none()
2548 && let Some(price) = price
2549 {
2550 order_entry.price(Some(price.to_string()));
2551 }
2552
2553 if let Some(trigger_price) = trigger_price {
2554 order_entry.trigger_price(Some(trigger_price.to_string()));
2555 }
2556
2557 if reduce_only {
2558 order_entry.reduce_only(Some(true));
2559 }
2560
2561 order_entry.is_leverage(spot_leverage(product_type, is_leverage));
2562
2563 if let Some(idx) = position_idx {
2564 order_entry.position_idx(Some(idx));
2565 }
2566
2567 order_entry.bbo_side_type(bbo_side_type);
2568 order_entry.bbo_level(bbo_level);
2569
2570 if let Some(tp_sl) = native_tp_sl {
2571 if let Some(ref tp) = tp_sl.take_profit {
2572 order_entry.take_profit(Some(tp.clone()));
2573 }
2574
2575 if let Some(ref sl) = tp_sl.stop_loss {
2576 order_entry.stop_loss(Some(sl.clone()));
2577 }
2578
2579 if let Some(tp_trigger) = tp_sl.tp_trigger_by {
2580 order_entry.tp_trigger_by(Some(tp_trigger));
2581 }
2582
2583 if let Some(sl_trigger) = tp_sl.sl_trigger_by {
2584 order_entry.sl_trigger_by(Some(sl_trigger));
2585 }
2586
2587 if let Some(tp_ot) = tp_sl.tp_order_type {
2588 order_entry.tp_order_type(Some(tp_ot));
2589 }
2590
2591 if let Some(sl_ot) = tp_sl.sl_order_type {
2592 order_entry.sl_order_type(Some(sl_ot));
2593 }
2594
2595 if let Some(ref tp_lp) = tp_sl.tp_limit_price {
2596 order_entry.tp_limit_price(Some(tp_lp.clone()));
2597 }
2598
2599 if let Some(ref sl_lp) = tp_sl.sl_limit_price {
2600 order_entry.sl_limit_price(Some(sl_lp.clone()));
2601 }
2602
2603 let mode = tp_sl.tpsl_mode.or_else(|| {
2606 (tp_sl.take_profit.is_some() || tp_sl.stop_loss.is_some())
2607 .then_some(BybitTpSlMode::Full)
2608 });
2609
2610 if let Some(m) = mode {
2611 order_entry.tpsl_mode(Some(m));
2612 }
2613
2614 if let Some(close) = tp_sl.close_on_trigger {
2615 order_entry.close_on_trigger(Some(close));
2616 }
2617
2618 if let Some(ref iv) = tp_sl.order_iv {
2619 order_entry.order_iv(Some(iv.clone()));
2620 }
2621
2622 if let Some(mmp) = tp_sl.mmp {
2623 order_entry.mmp(Some(mmp));
2624 }
2625 }
2626
2627 let order_entry = order_entry.build().build_anyhow()?;
2628
2629 let mut params = BybitPlaceOrderParamsBuilder::default();
2630 params.category(product_type);
2631 params.order(order_entry);
2632
2633 let params = params.build().build_anyhow()?;
2634
2635 let body = serde_json::to_value(¶ms)?;
2636 let response = self.inner.place_order(&body).await?;
2637
2638 let order_id = response
2639 .result
2640 .order_id
2641 .ok_or(BybitSubmitOrderError::MissingOrderId)?;
2642
2643 let order = self
2644 .query_order_by_id(
2645 product_type,
2646 order_id.as_str(),
2647 BYBIT_ORDER_REALTIME,
2648 "after submission",
2649 )
2650 .await
2651 .map_err(|source| BybitSubmitOrderError::PostSubmitLookup { source })?;
2652
2653 let is_rejection = order.order_status == crate::common::enums::BybitOrderStatus::Rejected
2658 || (order.order_status == crate::common::enums::BybitOrderStatus::Canceled
2659 && bybit_rejection_due_post_only(order.reject_reason.as_str()));
2660 if is_rejection && (order.cum_exec_qty.as_str() == "0" || order.cum_exec_qty.is_empty()) {
2661 return Err(BybitSubmitOrderError::Rejected {
2662 reason: order.reject_reason.to_string(),
2663 }
2664 .into());
2665 }
2666
2667 let ts_init = self.generate_ts_init();
2668
2669 parse_order_status_report(&order, &instrument, account_id, ts_init)
2670 }
2671
2672 pub async fn cancel_order(
2682 &self,
2683 account_id: AccountId,
2684 product_type: BybitProductType,
2685 instrument_id: InstrumentId,
2686 client_order_id: Option<ClientOrderId>,
2687 venue_order_id: Option<VenueOrderId>,
2688 ) -> anyhow::Result<OrderStatusReport> {
2689 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
2690 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2691
2692 let mut cancel_entry = BybitBatchCancelOrderEntryBuilder::default();
2693 cancel_entry.symbol(bybit_symbol.raw_symbol().to_string());
2694
2695 if let Some(venue_order_id) = venue_order_id {
2696 cancel_entry.order_id(venue_order_id.to_string());
2697 } else if let Some(client_order_id) = client_order_id {
2698 cancel_entry.order_link_id(client_order_id.to_string());
2699 } else {
2700 anyhow::bail!("Either client_order_id or venue_order_id must be provided");
2701 }
2702
2703 let cancel_entry = cancel_entry.build().build_anyhow()?;
2704
2705 let mut params = BybitCancelOrderParamsBuilder::default();
2706 params.category(product_type);
2707 params.order(cancel_entry);
2708
2709 let params = params.build().build_anyhow()?;
2710 let body = serde_json::to_vec(¶ms)?;
2711
2712 let response: BybitPlaceOrderResponse = self
2713 .inner
2714 .send_request::<_, ()>(Method::POST, "/v5/order/cancel", None, Some(body), true)
2715 .await?;
2716
2717 let order_id = response
2718 .result
2719 .order_id
2720 .ok_or(BybitCancelOrderError::MissingOrderId)?;
2721
2722 let order = self
2723 .query_order_by_id(
2724 product_type,
2725 order_id.as_str(),
2726 BYBIT_ORDER_HISTORY,
2727 "after cancellation",
2728 )
2729 .await
2730 .map_err(|source| BybitCancelOrderError::PostCancelLookup { source })?;
2731
2732 let ts_init = self.generate_ts_init();
2733
2734 parse_order_status_report(&order, &instrument, account_id, ts_init)
2735 }
2736
2737 pub async fn batch_cancel_orders(
2747 &self,
2748 account_id: AccountId,
2749 product_type: BybitProductType,
2750 instrument_ids: Vec<InstrumentId>,
2751 client_order_ids: Vec<Option<ClientOrderId>>,
2752 venue_order_ids: Vec<Option<VenueOrderId>>,
2753 ) -> anyhow::Result<Vec<OrderStatusReport>> {
2754 if instrument_ids.len() != client_order_ids.len()
2755 || instrument_ids.len() != venue_order_ids.len()
2756 {
2757 anyhow::bail!(
2758 "instrument_ids, client_order_ids, and venue_order_ids must have the same length"
2759 );
2760 }
2761
2762 if instrument_ids.is_empty() {
2763 return Ok(Vec::new());
2764 }
2765
2766 let call_limit = batch_call_limit(product_type);
2767 if instrument_ids.len() > call_limit {
2768 anyhow::bail!(
2769 "Batch cancel limit is {call_limit} orders for {}",
2770 product_type.as_str()
2771 );
2772 }
2773
2774 let mut cancel_entries = Vec::new();
2775
2776 for ((instrument_id, client_order_id), venue_order_id) in instrument_ids
2777 .iter()
2778 .zip(client_order_ids.iter())
2779 .zip(venue_order_ids.iter())
2780 {
2781 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2782 let mut cancel_entry = BybitBatchCancelOrderEntryBuilder::default();
2783 cancel_entry.symbol(bybit_symbol.raw_symbol().to_string());
2784
2785 if let Some(venue_order_id) = venue_order_id {
2786 cancel_entry.order_id(venue_order_id.to_string());
2787 } else if let Some(client_order_id) = client_order_id {
2788 cancel_entry.order_link_id(client_order_id.to_string());
2789 } else {
2790 anyhow::bail!(
2791 "Either client_order_id or venue_order_id must be provided for each order"
2792 );
2793 }
2794
2795 cancel_entries.push(cancel_entry.build().build_anyhow()?);
2796 }
2797
2798 let chunk_limit = batch_endpoint_limit(product_type).min(batch_send_limit(product_type));
2799 for chunk in cancel_entries.chunks(chunk_limit) {
2800 let mut params = BybitBatchCancelOrderParamsBuilder::default();
2801 params.category(product_type);
2802 params.request(chunk.to_vec());
2803
2804 let params = params.build().build_anyhow()?;
2805 let body = serde_json::to_vec(¶ms)?;
2806
2807 let _response: BybitPlaceOrderResponse = self
2808 .inner
2809 .send_request::<_, ()>(
2810 Method::POST,
2811 "/v5/order/cancel-batch",
2812 None,
2813 Some(body),
2814 true,
2815 )
2816 .await?;
2817 }
2818
2819 let mut reports = Vec::new();
2821
2822 for (instrument_id, (client_order_id, venue_order_id)) in instrument_ids
2823 .iter()
2824 .zip(client_order_ids.iter().zip(venue_order_ids.iter()))
2825 {
2826 let Ok(instrument) = self.instrument_from_cache(&instrument_id.symbol) else {
2827 log::debug!(
2828 "Skipping cancelled order report for instrument not in cache: symbol={}",
2829 instrument_id.symbol
2830 );
2831 continue;
2832 };
2833
2834 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2835
2836 let mut query_params = BybitOpenOrdersParamsBuilder::default();
2837 query_params.category(product_type);
2838 query_params.symbol(bybit_symbol.raw_symbol().to_string());
2839
2840 if let Some(venue_order_id) = venue_order_id {
2841 query_params.order_id(venue_order_id.to_string());
2842 } else if let Some(client_order_id) = client_order_id {
2843 query_params.order_link_id(client_order_id.to_string());
2844 }
2845
2846 let query_params = query_params.build().build_anyhow()?;
2847 let order_response: BybitOrderHistoryResponse = self
2848 .inner
2849 .send_request(
2850 Method::GET,
2851 BYBIT_ORDER_HISTORY,
2852 Some(&query_params),
2853 None,
2854 true,
2855 )
2856 .await?;
2857
2858 if let Some(order) = order_response.result.list.into_iter().next() {
2859 let ts_init = self.generate_ts_init();
2860 let report = parse_order_status_report(&order, &instrument, account_id, ts_init)?;
2861 reports.push(report);
2862 }
2863 }
2864
2865 Ok(reports)
2866 }
2867
2868 pub async fn cancel_all_orders(
2877 &self,
2878 account_id: AccountId,
2879 product_type: BybitProductType,
2880 instrument_id: InstrumentId,
2881 ) -> anyhow::Result<Vec<OrderStatusReport>> {
2882 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
2883 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2884
2885 let mut params = BybitCancelAllOrdersParamsBuilder::default();
2886 params.category(product_type);
2887 params.symbol(bybit_symbol.raw_symbol().to_string());
2888
2889 let params = params.build().build_anyhow()?;
2890 let body = serde_json::to_vec(¶ms)?;
2891
2892 let _response: crate::common::models::BybitListResponse<serde_json::Value> = self
2893 .inner
2894 .send_request::<_, ()>(Method::POST, "/v5/order/cancel-all", None, Some(body), true)
2895 .await?;
2896
2897 let mut query_params = BybitOrderHistoryParamsBuilder::default();
2899 query_params.category(product_type);
2900 query_params.symbol(bybit_symbol.raw_symbol().to_string());
2901 query_params.limit(50u32);
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 let ts_init = self.generate_ts_init();
2916
2917 let mut reports = Vec::new();
2918
2919 for order in order_response.result.list {
2920 if let Ok(report) = parse_order_status_report(&order, &instrument, account_id, ts_init)
2921 {
2922 reports.push(report);
2923 }
2924 }
2925
2926 Ok(reports)
2927 }
2928
2929 #[expect(clippy::too_many_arguments)]
2940 pub async fn modify_order(
2941 &self,
2942 account_id: AccountId,
2943 product_type: BybitProductType,
2944 instrument_id: InstrumentId,
2945 client_order_id: Option<ClientOrderId>,
2946 venue_order_id: Option<VenueOrderId>,
2947 quantity: Option<Quantity>,
2948 price: Option<Price>,
2949 ) -> anyhow::Result<OrderStatusReport> {
2950 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
2951 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
2952
2953 let mut amend_entry = BybitBatchAmendOrderEntryBuilder::default();
2954 amend_entry.symbol(bybit_symbol.raw_symbol().to_string());
2955
2956 if let Some(venue_order_id) = venue_order_id {
2957 amend_entry.order_id(venue_order_id.to_string());
2958 } else if let Some(client_order_id) = client_order_id {
2959 amend_entry.order_link_id(client_order_id.to_string());
2960 } else {
2961 anyhow::bail!("Either client_order_id or venue_order_id must be provided");
2962 }
2963
2964 if let Some(quantity) = quantity {
2965 amend_entry.qty(Some(quantity.to_string()));
2966 }
2967
2968 if let Some(price) = price {
2969 amend_entry.price(Some(price.to_string()));
2970 }
2971
2972 let amend_entry = amend_entry.build().build_anyhow()?;
2973
2974 let mut params = BybitAmendOrderParamsBuilder::default();
2975 params.category(product_type);
2976 params.order(amend_entry);
2977
2978 let params = params.build().build_anyhow()?;
2979 let body = serde_json::to_vec(¶ms)?;
2980
2981 let response: BybitPlaceOrderResponse = self
2982 .inner
2983 .send_request::<_, ()>(Method::POST, "/v5/order/amend", None, Some(body), true)
2984 .await?;
2985
2986 let order_id = response
2987 .result
2988 .order_id
2989 .ok_or(BybitModifyOrderError::MissingOrderId)?;
2990
2991 let order = self
2992 .query_order_by_id(
2993 product_type,
2994 order_id.as_str(),
2995 BYBIT_ORDER_REALTIME,
2996 "after amendment",
2997 )
2998 .await
2999 .map_err(|source| BybitModifyOrderError::PostModifyLookup { source })?;
3000
3001 let ts_init = self.generate_ts_init();
3002
3003 parse_order_status_report(&order, &instrument, account_id, ts_init)
3004 }
3005
3006 pub async fn query_order(
3015 &self,
3016 account_id: AccountId,
3017 product_type: BybitProductType,
3018 instrument_id: InstrumentId,
3019 client_order_id: Option<ClientOrderId>,
3020 venue_order_id: Option<VenueOrderId>,
3021 ) -> anyhow::Result<Option<OrderStatusReport>> {
3022 log::debug!(
3023 "query_order: instrument_id={instrument_id}, client_order_id={client_order_id:?}, venue_order_id={venue_order_id:?}"
3024 );
3025
3026 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
3027
3028 let mut params = BybitOpenOrdersParamsBuilder::default();
3029 params.category(product_type);
3030 params.symbol(bybit_symbol.raw_symbol().to_string());
3032
3033 if let Some(venue_order_id) = venue_order_id {
3034 params.order_id(venue_order_id.to_string());
3035 } else if let Some(client_order_id) = client_order_id {
3036 params.order_link_id(client_order_id.to_string());
3037 } else {
3038 anyhow::bail!("Either client_order_id or venue_order_id must be provided");
3039 }
3040
3041 let params = params.build().build_anyhow()?;
3042 let mut response: BybitOpenOrdersResponse = self
3043 .inner
3044 .send_request(Method::GET, BYBIT_ORDER_REALTIME, Some(¶ms), None, true)
3045 .await?;
3046
3047 if response.result.list.is_empty() && product_type != BybitProductType::Option {
3049 log::debug!("Order not found in open orders, trying with StopOrder filter");
3050
3051 let mut stop_params = BybitOpenOrdersParamsBuilder::default();
3052 stop_params.category(product_type);
3053 stop_params.symbol(bybit_symbol.raw_symbol().to_string());
3054 stop_params.order_filter(BybitOrderFilter::StopOrder);
3055
3056 if let Some(venue_order_id) = venue_order_id {
3057 stop_params.order_id(venue_order_id.to_string());
3058 } else if let Some(client_order_id) = client_order_id {
3059 stop_params.order_link_id(client_order_id.to_string());
3060 }
3061
3062 let stop_params = stop_params.build().build_anyhow()?;
3063 response = self
3064 .inner
3065 .send_request(
3066 Method::GET,
3067 BYBIT_ORDER_REALTIME,
3068 Some(&stop_params),
3069 None,
3070 true,
3071 )
3072 .await?;
3073 }
3074
3075 if response.result.list.is_empty() {
3077 log::debug!("Order not found in open orders, checking order history");
3078
3079 let mut history_params = BybitOrderHistoryParamsBuilder::default();
3080 history_params.category(product_type);
3081 history_params.symbol(bybit_symbol.raw_symbol().to_string());
3082
3083 if let Some(venue_order_id) = venue_order_id {
3084 history_params.order_id(venue_order_id.to_string());
3085 } else if let Some(client_order_id) = client_order_id {
3086 history_params.order_link_id(client_order_id.to_string());
3087 }
3088
3089 let history_params = history_params.build().build_anyhow()?;
3090
3091 let mut history_response: BybitOrderHistoryResponse = self
3092 .inner
3093 .send_request(
3094 Method::GET,
3095 BYBIT_ORDER_HISTORY,
3096 Some(&history_params),
3097 None,
3098 true,
3099 )
3100 .await?;
3101
3102 if history_response.result.list.is_empty() && product_type == BybitProductType::Option {
3103 log::debug!("Option order not found in order history");
3104 return Ok(None);
3105 }
3106
3107 if history_response.result.list.is_empty() && product_type != BybitProductType::Option {
3109 log::debug!("Order not found in order history, trying with StopOrder filter");
3110
3111 let mut stop_history_params = BybitOrderHistoryParamsBuilder::default();
3112 stop_history_params.category(product_type);
3113 stop_history_params.symbol(bybit_symbol.raw_symbol().to_string());
3114 stop_history_params.order_filter(BybitOrderFilter::StopOrder);
3115
3116 if let Some(venue_order_id) = venue_order_id {
3117 stop_history_params.order_id(venue_order_id.to_string());
3118 } else if let Some(client_order_id) = client_order_id {
3119 stop_history_params.order_link_id(client_order_id.to_string());
3120 }
3121
3122 let stop_history_params = stop_history_params
3123 .build()
3124 .map_err(|e| anyhow::anyhow!(e))?;
3125
3126 history_response = self
3127 .inner
3128 .send_request(
3129 Method::GET,
3130 BYBIT_ORDER_HISTORY,
3131 Some(&stop_history_params),
3132 None,
3133 true,
3134 )
3135 .await?;
3136
3137 if history_response.result.list.is_empty() {
3138 log::debug!("Order not found in order history with StopOrder filter either");
3139 return Ok(None);
3140 }
3141 }
3142
3143 response.result.list = history_response.result.list;
3145 }
3146
3147 let order = &response.result.list[0];
3148 let ts_init = self.generate_ts_init();
3149
3150 log::debug!(
3151 "Query order response: symbol={}, order_id={}, order_link_id={}",
3152 order.symbol.as_str(),
3153 order.order_id.as_str(),
3154 order.order_link_id.as_str()
3155 );
3156
3157 let instrument = self
3158 .instrument_from_cache(&instrument_id.symbol)
3159 .map_err(|e| {
3160 log::error!(
3161 "Instrument cache miss for symbol '{}': {}",
3162 instrument_id.symbol.as_str(),
3163 e
3164 );
3165 anyhow::anyhow!(
3166 "Failed to query order {}: {}",
3167 client_order_id
3168 .as_ref()
3169 .map(|id| id.to_string())
3170 .or_else(|| venue_order_id.as_ref().map(|id| id.to_string()))
3171 .unwrap_or_else(|| "unknown".to_string()),
3172 e
3173 )
3174 })?;
3175
3176 log::debug!("Retrieved instrument from cache: id={}", instrument.id());
3177
3178 let report =
3179 parse_order_status_report(order, &instrument, account_id, ts_init).map_err(|e| {
3180 log::error!(
3181 "Failed to parse order status report for {}: {}",
3182 order.order_link_id.as_str(),
3183 e
3184 );
3185 e
3186 })?;
3187
3188 log::debug!(
3189 "Successfully created OrderStatusReport for {}",
3190 order.order_link_id.as_str()
3191 );
3192
3193 Ok(Some(report))
3194 }
3195
3196 async fn fetch_fee_map(
3197 &self,
3198 product_type: BybitProductType,
3199 base_coin: Option<Ustr>,
3200 ) -> anyhow::Result<AHashMap<Ustr, BybitFeeRate>> {
3201 let mut fee_params = BybitFeeRateParamsBuilder::default();
3202 fee_params.category(product_type);
3203 if let Some(bc) = base_coin {
3204 fee_params.base_coin(bc.to_string());
3205 }
3206 let Ok(params) = fee_params.build() else {
3207 return Ok(AHashMap::new());
3208 };
3209
3210 match self.inner.get_fee_rate(¶ms).await {
3211 Ok(response) => Ok(response
3212 .result
3213 .list
3214 .into_iter()
3215 .map(|f| (f.symbol, f))
3216 .collect()),
3217 Err(BybitHttpError::MissingCredentials) => {
3218 log::warn!("Missing credentials for fee rates, using defaults");
3219 Ok(AHashMap::new())
3220 }
3221 Err(BybitHttpError::BybitError {
3222 error_code,
3223 ref message,
3224 }) => {
3225 log::warn!(
3226 "{}",
3227 self.fee_rate_rejection_warning(product_type, error_code, message)
3228 );
3229 Ok(AHashMap::new())
3230 }
3231 Err(e) => Err(e.into()),
3232 }
3233 }
3234
3235 async fn fetch_option_fee_map(
3236 &self,
3237 base_coin: Option<Ustr>,
3238 ) -> anyhow::Result<AHashMap<Ustr, BybitFeeRate>> {
3239 let mut fee_params = BybitFeeRateParamsBuilder::default();
3240 fee_params.category(BybitProductType::Option);
3241 if let Some(bc) = base_coin {
3242 fee_params.base_coin(bc.to_string());
3243 }
3244 let Ok(params) = fee_params.build() else {
3245 return Ok(AHashMap::new());
3246 };
3247
3248 match self.inner.get_fee_rate(¶ms).await {
3249 Ok(response) => Ok(response
3250 .result
3251 .list
3252 .into_iter()
3253 .filter_map(|f| f.base_coin.map(|bc| (bc, f)))
3254 .collect()),
3255 Err(BybitHttpError::MissingCredentials) => {
3256 log::warn!("Missing credentials for option fee rates, using defaults");
3257 Ok(AHashMap::new())
3258 }
3259 Err(BybitHttpError::BybitError {
3260 error_code,
3261 ref message,
3262 }) => {
3263 let error_detail = Self::format_bybit_error_detail(error_code, message);
3264 log::warn!(
3265 "Option fee rate request rejected via /v5/account/fee-rate ({error_detail}), using defaults"
3266 );
3267 Ok(AHashMap::new())
3268 }
3269 Err(e) => {
3270 log::warn!("Option fee rate request failed ({e}), using defaults");
3271 Ok(AHashMap::new())
3272 }
3273 }
3274 }
3275
3276 fn fee_rate_rejection_warning(
3277 &self,
3278 product_type: BybitProductType,
3279 error_code: i32,
3280 message: &str,
3281 ) -> String {
3282 let product_type = product_type.as_ref().to_ascii_lowercase();
3283 let error_detail = Self::format_bybit_error_detail(error_code, message);
3284
3285 if self
3286 .base_url()
3287 .starts_with(bybit_http_base_url(BybitEnvironment::Demo))
3288 && matches!(product_type.as_str(), "linear" | "inverse")
3289 && error_code == 10001
3290 {
3291 format!(
3292 "Bybit demo rejected the {product_type} fee rate request via \
3293 /v5/account/fee-rate ({error_detail}); demo derivatives fee rates appear \
3294 unsupported, using defaults"
3295 )
3296 } else {
3297 format!(
3298 "Fee rate request rejected for {product_type} instruments via \
3299 /v5/account/fee-rate ({error_detail}), using defaults"
3300 )
3301 }
3302 }
3303
3304 fn format_bybit_error_detail(error_code: i32, message: &str) -> String {
3305 let message = message.trim();
3306 if message.is_empty() {
3307 format!("error {error_code}, no message")
3308 } else {
3309 format!("error {error_code}: {message}")
3310 }
3311 }
3312
3313 async fn paginate_instruments<D, F>(
3314 &self,
3315 product_type: BybitProductType,
3316 symbol: &Option<String>,
3317 base_coin: Option<Ustr>,
3318 mut parse: F,
3319 ) -> anyhow::Result<Vec<InstrumentAny>>
3320 where
3321 D: DeserializeOwned,
3322 BybitCursorListResponse<D>: BybitResponseCheck,
3323 F: FnMut(&D) -> Option<InstrumentAny>,
3324 {
3325 let mut instruments = Vec::new();
3326 let mut cursor: Option<String> = None;
3327 let mut prev_cursor: Option<String> = None;
3328
3329 loop {
3330 let params = BybitInstrumentsInfoParams {
3331 category: product_type,
3332 symbol: symbol.clone(),
3333 status: None,
3334 base_coin: base_coin.map(|u| u.to_string()),
3335 limit: Some(1000),
3336 cursor: cursor.clone(),
3337 };
3338
3339 let response: BybitCursorListResponse<D> = self.inner.get_instruments(¶ms).await?;
3340
3341 for definition in &response.result.list {
3342 if let Some(instrument) = parse(definition) {
3343 instruments.push(instrument);
3344 }
3345 }
3346
3347 cursor = response.result.next_page_cursor;
3348 if cursor.as_ref().is_none_or(|c| c.is_empty()) || cursor == prev_cursor {
3349 break;
3350 }
3351 prev_cursor = cursor.clone();
3352 }
3353
3354 Ok(instruments)
3355 }
3356
3357 pub async fn request_instrument_statuses(
3367 &self,
3368 product_type: BybitProductType,
3369 ) -> anyhow::Result<AHashMap<InstrumentId, MarketStatusAction>> {
3370 let mut statuses = AHashMap::new();
3371 let mut cursor: Option<String> = None;
3372
3373 loop {
3374 let params = BybitInstrumentsInfoParams {
3375 category: product_type,
3376 symbol: None,
3377 status: None,
3378 base_coin: None,
3379 limit: Some(1000),
3380 cursor: cursor.clone(),
3381 };
3382
3383 match product_type {
3384 BybitProductType::Spot => {
3385 let response: BybitCursorListResponse<BybitInstrumentSpot> =
3386 self.inner.get_instruments(¶ms).await?;
3387
3388 for def in &response.result.list {
3389 let symbol = make_bybit_symbol(def.symbol, product_type);
3390 let id = InstrumentId::new(Symbol::from(symbol), *BYBIT_VENUE);
3391 statuses.insert(id, MarketStatusAction::from(def.status));
3392 }
3393 cursor = response.result.next_page_cursor;
3394 }
3395 BybitProductType::Linear => {
3396 let response: BybitCursorListResponse<BybitInstrumentLinear> =
3397 self.inner.get_instruments(¶ms).await?;
3398
3399 for def in &response.result.list {
3400 let symbol = make_bybit_symbol(def.symbol, product_type);
3401 let id = InstrumentId::new(Symbol::from(symbol), *BYBIT_VENUE);
3402 let status = MarketStatusAction::from(def.status);
3403 if status == MarketStatusAction::Trading
3404 && def.contract_type == BybitContractType::LinearPerpetual
3405 && def.delivery_time != "0"
3406 {
3407 statuses.insert(id, MarketStatusAction::PreClose);
3408 } else {
3409 statuses.insert(id, status);
3410 }
3411 }
3412 cursor = response.result.next_page_cursor;
3413 }
3414 BybitProductType::Inverse => {
3415 let response: BybitCursorListResponse<BybitInstrumentInverse> =
3416 self.inner.get_instruments(¶ms).await?;
3417
3418 for def in &response.result.list {
3419 let symbol = make_bybit_symbol(def.symbol, product_type);
3420 let id = InstrumentId::new(Symbol::from(symbol), *BYBIT_VENUE);
3421 let status = MarketStatusAction::from(def.status);
3422 if status == MarketStatusAction::Trading
3423 && def.contract_type == BybitContractType::InversePerpetual
3424 && def.delivery_time != "0"
3425 {
3426 statuses.insert(id, MarketStatusAction::PreClose);
3427 } else {
3428 statuses.insert(id, status);
3429 }
3430 }
3431 cursor = response.result.next_page_cursor;
3432 }
3433 BybitProductType::Option => {
3434 let response: BybitCursorListResponse<BybitInstrumentOption> =
3435 self.inner.get_instruments(¶ms).await?;
3436
3437 for def in &response.result.list {
3438 let symbol = make_bybit_symbol(def.symbol, product_type);
3439 let id = InstrumentId::new(Symbol::from(symbol), *BYBIT_VENUE);
3440 statuses.insert(id, MarketStatusAction::from(def.status));
3441 }
3442 cursor = response.result.next_page_cursor;
3443 }
3444 }
3445
3446 if cursor.as_ref().is_none_or(|c| c.is_empty()) {
3447 break;
3448 }
3449 }
3450
3451 Ok(statuses)
3452 }
3453
3454 pub async fn request_instruments(
3464 &self,
3465 product_type: BybitProductType,
3466 symbol: Option<String>,
3467 base_coin: Option<Ustr>,
3468 ) -> anyhow::Result<Vec<InstrumentAny>> {
3469 let ts_init = self.generate_ts_init();
3470
3471 let default_fee_rate = |symbol: Ustr| BybitFeeRate {
3472 symbol,
3473 taker_fee_rate: "0.001".to_string(),
3474 maker_fee_rate: "0.001".to_string(),
3475 base_coin: None,
3476 };
3477
3478 let instruments = match product_type {
3479 BybitProductType::Spot => {
3480 let fee_map = self.fetch_fee_map(product_type, base_coin).await?;
3481 self.paginate_instruments::<BybitInstrumentSpot, _>(
3482 product_type,
3483 &symbol,
3484 base_coin,
3485 |def| {
3486 let fee = fee_map
3487 .get(&def.symbol)
3488 .cloned()
3489 .unwrap_or_else(|| default_fee_rate(def.symbol));
3490 parse_spot_instrument(def, &fee, ts_init, ts_init).ok()
3491 },
3492 )
3493 .await?
3494 }
3495 BybitProductType::Linear => {
3496 let fee_map = self.fetch_fee_map(product_type, base_coin).await?;
3497 self.paginate_instruments::<BybitInstrumentLinear, _>(
3498 product_type,
3499 &symbol,
3500 base_coin,
3501 |def| {
3502 let fee = fee_map
3503 .get(&def.symbol)
3504 .cloned()
3505 .unwrap_or_else(|| default_fee_rate(def.symbol));
3506 parse_linear_instrument(def, &fee, ts_init, ts_init).ok()
3507 },
3508 )
3509 .await?
3510 }
3511 BybitProductType::Inverse => {
3512 let fee_map = self.fetch_fee_map(product_type, base_coin).await?;
3513 self.paginate_instruments::<BybitInstrumentInverse, _>(
3514 product_type,
3515 &symbol,
3516 base_coin,
3517 |def| {
3518 let fee = fee_map
3519 .get(&def.symbol)
3520 .cloned()
3521 .unwrap_or_else(|| default_fee_rate(def.symbol));
3522 parse_inverse_instrument(def, &fee, ts_init, ts_init).ok()
3523 },
3524 )
3525 .await?
3526 }
3527 BybitProductType::Option => {
3528 let fee_map = self.fetch_option_fee_map(base_coin).await?;
3529 self.paginate_instruments::<BybitInstrumentOption, _>(
3530 product_type,
3531 &symbol,
3532 base_coin,
3533 |def| {
3534 let fee = fee_map.get(&def.base_coin);
3535 parse_option_instrument(def, fee, ts_init, ts_init).ok()
3536 },
3537 )
3538 .await?
3539 }
3540 };
3541
3542 self.cache_instruments(&instruments);
3543
3544 Ok(instruments)
3545 }
3546
3547 pub async fn request_instruments_with_statuses(
3557 &self,
3558 product_type: BybitProductType,
3559 ) -> anyhow::Result<(
3560 Vec<InstrumentAny>,
3561 AHashMap<InstrumentId, MarketStatusAction>,
3562 )> {
3563 let ts_init = self.generate_ts_init();
3564 let mut statuses = AHashMap::new();
3565
3566 let default_fee_rate = |symbol: Ustr| BybitFeeRate {
3567 symbol,
3568 taker_fee_rate: "0.001".to_string(),
3569 maker_fee_rate: "0.001".to_string(),
3570 base_coin: None,
3571 };
3572
3573 let perp_status = |status: MarketStatusAction, is_scheduled_perp: bool| {
3575 if status == MarketStatusAction::Trading && is_scheduled_perp {
3576 MarketStatusAction::PreClose
3577 } else {
3578 status
3579 }
3580 };
3581
3582 let instruments = match product_type {
3583 BybitProductType::Spot => {
3584 let fee_map = self.fetch_fee_map(product_type, None).await?;
3585 self.paginate_instruments::<BybitInstrumentSpot, _>(
3586 product_type,
3587 &None::<String>,
3588 None,
3589 |def| {
3590 let id = InstrumentId::new(
3591 Symbol::from(make_bybit_symbol(def.symbol, product_type)),
3592 *BYBIT_VENUE,
3593 );
3594 statuses.insert(id, MarketStatusAction::from(def.status));
3595 let fee = fee_map
3596 .get(&def.symbol)
3597 .cloned()
3598 .unwrap_or_else(|| default_fee_rate(def.symbol));
3599 parse_spot_instrument(def, &fee, ts_init, ts_init).ok()
3600 },
3601 )
3602 .await?
3603 }
3604 BybitProductType::Linear => {
3605 let fee_map = self.fetch_fee_map(product_type, None).await?;
3606 self.paginate_instruments::<BybitInstrumentLinear, _>(
3607 product_type,
3608 &None::<String>,
3609 None,
3610 |def| {
3611 let id = InstrumentId::new(
3612 Symbol::from(make_bybit_symbol(def.symbol, product_type)),
3613 *BYBIT_VENUE,
3614 );
3615 let scheduled = def.contract_type == BybitContractType::LinearPerpetual
3616 && def.delivery_time != "0";
3617 statuses.insert(id, perp_status(def.status.into(), scheduled));
3618 let fee = fee_map
3619 .get(&def.symbol)
3620 .cloned()
3621 .unwrap_or_else(|| default_fee_rate(def.symbol));
3622 parse_linear_instrument(def, &fee, ts_init, ts_init).ok()
3623 },
3624 )
3625 .await?
3626 }
3627 BybitProductType::Inverse => {
3628 let fee_map = self.fetch_fee_map(product_type, None).await?;
3629 self.paginate_instruments::<BybitInstrumentInverse, _>(
3630 product_type,
3631 &None::<String>,
3632 None,
3633 |def| {
3634 let id = InstrumentId::new(
3635 Symbol::from(make_bybit_symbol(def.symbol, product_type)),
3636 *BYBIT_VENUE,
3637 );
3638 let scheduled = def.contract_type == BybitContractType::InversePerpetual
3639 && def.delivery_time != "0";
3640 statuses.insert(id, perp_status(def.status.into(), scheduled));
3641 let fee = fee_map
3642 .get(&def.symbol)
3643 .cloned()
3644 .unwrap_or_else(|| default_fee_rate(def.symbol));
3645 parse_inverse_instrument(def, &fee, ts_init, ts_init).ok()
3646 },
3647 )
3648 .await?
3649 }
3650 BybitProductType::Option => {
3651 let fee_map = self.fetch_option_fee_map(None).await?;
3652 self.paginate_instruments::<BybitInstrumentOption, _>(
3653 product_type,
3654 &None::<String>,
3655 None,
3656 |def| {
3657 let id = InstrumentId::new(
3658 Symbol::from(make_bybit_symbol(def.symbol, product_type)),
3659 *BYBIT_VENUE,
3660 );
3661 statuses.insert(id, MarketStatusAction::from(def.status));
3662 let fee = fee_map.get(&def.base_coin);
3663 parse_option_instrument(def, fee, ts_init, ts_init).ok()
3664 },
3665 )
3666 .await?
3667 }
3668 };
3669
3670 self.cache_instruments(&instruments);
3671
3672 Ok((instruments, statuses))
3673 }
3674
3675 pub async fn request_tickers(
3688 &self,
3689 params: &BybitTickersParams,
3690 ) -> anyhow::Result<Vec<BybitTickerData>> {
3691 use super::models::{
3692 BybitTickersLinearResponse, BybitTickersOptionResponse, BybitTickersSpotResponse,
3693 };
3694
3695 match params.category {
3696 BybitProductType::Spot => {
3697 let response: BybitTickersSpotResponse = self.inner.get_tickers(params).await?;
3698 Ok(response.result.list.into_iter().map(Into::into).collect())
3699 }
3700 BybitProductType::Linear | BybitProductType::Inverse => {
3701 let response: BybitTickersLinearResponse = self.inner.get_tickers(params).await?;
3702 Ok(response.result.list.into_iter().map(Into::into).collect())
3703 }
3704 BybitProductType::Option => {
3705 let response: BybitTickersOptionResponse = self.inner.get_tickers(params).await?;
3706 Ok(response.result.list.into_iter().map(Into::into).collect())
3707 }
3708 }
3709 }
3710
3711 pub async fn request_option_tickers_raw(
3720 &self,
3721 base_coin: &str,
3722 ) -> anyhow::Result<Vec<BybitTickerOption>> {
3723 let params = BybitTickersParams {
3724 category: BybitProductType::Option,
3725 symbol: None,
3726 base_coin: Some(base_coin.to_string()),
3727 exp_date: None,
3728 };
3729 let response: BybitTickersOptionResponse = self.inner.get_tickers(¶ms).await?;
3730 Ok(response.result.list)
3731 }
3732
3733 pub async fn request_option_tickers_raw_with_params(
3742 &self,
3743 params: &BybitTickersParams,
3744 ) -> anyhow::Result<Vec<BybitTickerOption>> {
3745 let response: BybitTickersOptionResponse = self.inner.get_tickers(params).await?;
3746 Ok(response.result.list)
3747 }
3748
3749 pub async fn request_trades(
3769 &self,
3770 product_type: BybitProductType,
3771 instrument_id: InstrumentId,
3772 limit: Option<u32>,
3773 ) -> anyhow::Result<Vec<TradeTick>> {
3774 let instrument = self.instrument_from_cache_by_id(instrument_id)?;
3775 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
3776
3777 let mut params_builder = BybitTradesParamsBuilder::default();
3778 params_builder.category(product_type);
3779 params_builder.symbol(bybit_symbol.raw_symbol().to_string());
3780
3781 if let Some(limit_val) = limit {
3782 params_builder.limit(limit_val);
3783 }
3784
3785 let params = params_builder.build().build_anyhow()?;
3786 let response = self.inner.get_recent_trades(¶ms).await?;
3787
3788 let mut trades = Vec::new();
3789
3790 for trade in response.result.list {
3791 if let Ok(trade_tick) = parse_trade_tick(&trade, &instrument, None) {
3792 trades.push(trade_tick);
3793 }
3794 }
3795
3796 Ok(trades)
3797 }
3798
3799 pub async fn request_funding_rates(
3812 &self,
3813 product_type: BybitProductType,
3814 instrument_id: InstrumentId,
3815 start: Option<Timestamp>,
3816 end: Option<Timestamp>,
3817 limit: Option<u32>,
3818 ) -> anyhow::Result<Vec<FundingRateUpdate>> {
3819 let instrument = self.instrument_from_cache_by_id(instrument_id)?;
3820 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
3821
3822 let start_ms = start.map(|dt| dt.as_millisecond());
3823 let mut seen_timestamps: AHashSet<i64> = AHashSet::new();
3824
3825 let mut raw_funding_rates = Vec::new();
3826
3827 let mut current_end_ms = match (start, end) {
3829 (Some(_), None) => Some(Timestamp::now().as_millisecond()),
3830 _ => end.map(|dt| dt.as_millisecond()),
3831 };
3832
3833 loop {
3834 let mut params_builder = BybitFundingParamsBuilder::default();
3835 params_builder.category(product_type);
3836 params_builder.symbol(bybit_symbol.raw_symbol().to_string());
3837 params_builder.limit(limit.unwrap_or(200).clamp(0, 200)); if let Some(start_val) = start_ms {
3840 params_builder.start_time(start_val);
3841 }
3842
3843 if let Some(end_val) = current_end_ms {
3844 params_builder.end_time(end_val);
3845 }
3846
3847 let params = params_builder.build().build_anyhow()?;
3848 let response = self.inner.get_funding_history(¶ms).await?;
3849
3850 let funding_rates = response.result.list;
3851
3852 let mut new_funding_rates_with_ts: Vec<(i64, _)> = funding_rates
3853 .into_iter()
3854 .filter_map(|f| {
3855 let Ok(ts) = f.funding_rate_timestamp.parse::<i64>() else {
3856 return None;
3857 };
3858
3859 seen_timestamps.insert(ts).then_some((ts, f))
3860 })
3861 .collect();
3862
3863 new_funding_rates_with_ts.sort_by_key(|(ts, _)| Reverse(*ts));
3864
3865 let earliest_funding_time = match new_funding_rates_with_ts.last() {
3866 Some((last_ts, _)) => *last_ts,
3867 None => break,
3868 };
3869
3870 let new_funding_rates = new_funding_rates_with_ts.into_iter().map(|(_, f)| f);
3871 raw_funding_rates.extend(new_funding_rates);
3872
3873 if let Some(limit_val) = limit
3875 && raw_funding_rates.len() >= limit_val as usize
3876 {
3877 break;
3878 }
3879
3880 if let Some(start_val) = start_ms
3881 && earliest_funding_time <= start_val
3882 {
3883 break;
3884 }
3885
3886 current_end_ms = Some(earliest_funding_time - 1);
3888 }
3889
3890 if let Some(limit_val) = limit {
3891 raw_funding_rates.truncate(limit_val as usize);
3892 }
3893 let mut rates: Vec<FundingRateUpdate> = Vec::with_capacity(raw_funding_rates.len());
3894
3895 for window in raw_funding_rates.windows(2) {
3896 let raw = &window[0];
3897 let timestamp = raw
3898 .funding_rate_timestamp
3899 .parse::<i64>()
3900 .map_err(|_| anyhow::anyhow!("invalid funding_rate_timestamp"))?;
3901 let older_timestamp = window[1]
3902 .funding_rate_timestamp
3903 .parse::<i64>()
3904 .map_err(|_| anyhow::anyhow!("invalid funding_rate_timestamp"))?;
3905
3906 let interval_millis = timestamp - older_timestamp;
3907 let rate = parse_funding_rate(raw, &instrument, Some(interval_millis))?;
3908
3909 rates.push(rate);
3910 }
3911
3912 if let Some(last_raw) = raw_funding_rates.last() {
3913 let rate = parse_funding_rate(last_raw, &instrument, None)?;
3914 rates.push(rate);
3915 }
3916
3917 rates.reverse();
3918
3919 Ok(rates)
3920 }
3921
3922 pub async fn request_orderbook_snapshot(
3940 &self,
3941 product_type: BybitProductType,
3942 instrument_id: InstrumentId,
3943 limit: Option<u32>,
3944 ) -> anyhow::Result<OrderBookDeltas> {
3945 let instrument = self.instrument_from_cache_by_id(instrument_id)?;
3946 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
3947
3948 let mut params_builder = BybitOrderbookParamsBuilder::default();
3949 params_builder.category(product_type);
3950 params_builder.symbol(bybit_symbol.raw_symbol().to_string());
3951
3952 if let Some(limit) = limit {
3953 let max_limit = match product_type {
3954 BybitProductType::Spot => 200,
3955 BybitProductType::Option => 25,
3956 BybitProductType::Linear | BybitProductType::Inverse => 500,
3957 };
3958 let clamped_limit = limit.min(max_limit);
3959 if limit > max_limit {
3960 log::warn!(
3961 "Bybit orderbook snapshot request depth limit exceeds venue maximum; clamping: limit={limit}, clamped_limit={clamped_limit}",
3962 );
3963 }
3964 params_builder.limit(clamped_limit);
3965 }
3966
3967 let params = params_builder.build().build_anyhow()?;
3968 let response = self.inner.get_orderbook(¶ms).await?;
3969
3970 let deltas = parse_orderbook(&response.result, &instrument, None)?;
3971
3972 Ok(deltas)
3973 }
3974
3975 pub async fn request_bars(
3988 &self,
3989 product_type: BybitProductType,
3990 bar_type: BarType,
3991 start: Option<Timestamp>,
3992 end: Option<Timestamp>,
3993 limit: Option<u32>,
3994 timestamp_on_close: bool,
3995 ) -> anyhow::Result<Vec<Bar>> {
3996 let instrument_id = bar_type.instrument_id();
3997 let instrument = self.instrument_from_cache_by_id(instrument_id)?;
3998 let bybit_symbol = BybitSymbol::new(instrument_id.symbol.as_str())?;
3999
4000 let interval = bar_spec_to_bybit_interval(
4002 bar_type.spec().aggregation,
4003 bar_type.spec().step.get() as u64,
4004 )?;
4005
4006 let start_ms = start.map(|dt| dt.as_millisecond());
4007 let mut seen_timestamps: AHashSet<i64> = AHashSet::new();
4008 let current_time_ms = get_atomic_clock_realtime().get_time_ms() as i64;
4009
4010 let mut pages: Vec<Vec<Bar>> = Vec::new();
4020 let mut total_bars = 0usize;
4021 let mut current_end = end.map(|dt| dt.as_millisecond());
4022 let mut page_count = 0;
4023
4024 loop {
4025 page_count += 1;
4026
4027 let mut params_builder = BybitKlinesParamsBuilder::default();
4028 params_builder.category(product_type);
4029 params_builder.symbol(bybit_symbol.raw_symbol().to_string());
4030 params_builder.interval(interval);
4031 params_builder.limit(1000u32); if let Some(start_val) = start_ms {
4034 params_builder.start(start_val);
4035 }
4036
4037 if let Some(end_val) = current_end {
4038 params_builder.end(end_val);
4039 }
4040
4041 let params = params_builder.build().build_anyhow()?;
4042 let response = self.inner.get_klines(¶ms).await?;
4043
4044 let klines = response.result.list;
4045 if klines.is_empty() {
4046 break;
4047 }
4048
4049 let mut klines_with_ts: Vec<(i64, _)> = klines
4051 .into_iter()
4052 .filter_map(|k| k.start.parse::<i64>().ok().map(|ts| (ts, k)))
4053 .collect();
4054
4055 klines_with_ts.sort_by_key(|(ts, _)| *ts);
4056
4057 let has_new = klines_with_ts
4059 .iter()
4060 .any(|(ts, _)| !seen_timestamps.contains(ts));
4061
4062 if !has_new {
4063 break;
4064 }
4065
4066 let mut page_bars = Vec::with_capacity(klines_with_ts.len());
4067
4068 let mut earliest_ts: Option<i64> = None;
4069
4070 for (start_time, kline) in &klines_with_ts {
4071 if earliest_ts.is_none_or(|ts| *start_time < ts) {
4073 earliest_ts = Some(*start_time);
4074 }
4075
4076 let bar_end_time = interval.bar_end_time_ms(*start_time);
4077 if bar_end_time > current_time_ms {
4078 continue;
4079 }
4080
4081 if !seen_timestamps.contains(start_time)
4082 && let Ok(bar) =
4083 parse_kline_bar(kline, &instrument, bar_type, timestamp_on_close, None)
4084 {
4085 page_bars.push(bar);
4086 seen_timestamps.insert(*start_time);
4087 }
4088 }
4089
4090 total_bars += page_bars.len();
4093 pages.push(page_bars);
4094
4095 if let Some(limit_val) = limit
4097 && total_bars >= limit_val as usize
4098 {
4099 break;
4100 }
4101
4102 let Some(earliest_bar_time) = earliest_ts else {
4105 break;
4106 };
4107
4108 if let Some(start_val) = start_ms
4109 && earliest_bar_time <= start_val
4110 {
4111 break;
4112 }
4113
4114 current_end = Some(earliest_bar_time - 1);
4115
4116 if page_count > 100 {
4118 break;
4119 }
4120 }
4121
4122 let mut all_bars: Vec<Bar> = Vec::with_capacity(total_bars);
4124 for page in pages.into_iter().rev() {
4125 all_bars.extend(page);
4126 }
4127
4128 if let Some(limit_val) = limit {
4130 let limit_usize = limit_val as usize;
4131 if all_bars.len() > limit_usize {
4132 let start_idx = all_bars.len() - limit_usize;
4133 return Ok(all_bars[start_idx..].to_vec());
4134 }
4135 }
4136
4137 Ok(all_bars)
4138 }
4139
4140 fn instrument_from_cache_by_id(
4141 &self,
4142 instrument_id: InstrumentId,
4143 ) -> anyhow::Result<InstrumentAny> {
4144 self.get_instrument(&instrument_id.symbol.inner())
4145 .ok_or_else(|| InstrumentLookupError::not_found(instrument_id).into())
4146 }
4147
4148 pub async fn request_fee_rates(
4160 &self,
4161 product_type: BybitProductType,
4162 symbol: Option<String>,
4163 base_coin: Option<String>,
4164 ) -> anyhow::Result<Vec<BybitFeeRate>> {
4165 let params = BybitFeeRateParams {
4166 category: product_type,
4167 symbol,
4168 base_coin,
4169 };
4170
4171 let response = self.inner.get_fee_rate(¶ms).await?;
4172 Ok(response.result.list)
4173 }
4174
4175 pub async fn request_account_state(
4187 &self,
4188 account_type: BybitAccountType,
4189 account_id: AccountId,
4190 ) -> anyhow::Result<AccountState> {
4191 let params = BybitWalletBalanceParams {
4192 account_type,
4193 coin: None,
4194 };
4195
4196 let response = self.inner.get_wallet_balance(¶ms).await?;
4197 let ts_init = self.generate_ts_init();
4198
4199 let wallet_balance = response
4201 .result
4202 .list
4203 .first()
4204 .ok_or_else(|| anyhow::anyhow!("No wallet balance found in response"))?;
4205
4206 parse_account_state(wallet_balance, account_id, ts_init)
4207 }
4208
4209 #[expect(clippy::too_many_arguments)]
4225 pub async fn request_order_status_reports(
4226 &self,
4227 account_id: AccountId,
4228 product_type: BybitProductType,
4229 instrument_id: Option<InstrumentId>,
4230 open_only: bool,
4231 start: Option<Timestamp>,
4232 end: Option<Timestamp>,
4233 limit: Option<u32>,
4234 ) -> anyhow::Result<Vec<OrderStatusReport>> {
4235 let symbol_param = if let Some(id) = instrument_id.as_ref() {
4237 let symbol_str = id.symbol.as_str();
4238 if symbol_str.is_empty() {
4239 None
4240 } else {
4241 Some(BybitSymbol::new(symbol_str)?.raw_symbol().to_string())
4242 }
4243 } else {
4244 None
4245 };
4246
4247 let settle_coins_to_query: Vec<Option<String>> =
4250 if product_type == BybitProductType::Linear && symbol_param.is_none() {
4251 vec![Some("USDT".to_string()), Some("USDC".to_string())]
4252 } else {
4253 match product_type {
4254 BybitProductType::Inverse => vec![None],
4255 _ => vec![None],
4256 }
4257 };
4258
4259 let mut all_collected_orders = Vec::new();
4260 let mut total_collected_across_coins = 0;
4261
4262 for settle_coin in settle_coins_to_query {
4263 let remaining_limit = if let Some(limit) = limit {
4264 let remaining = (limit as usize).saturating_sub(total_collected_across_coins);
4265 if remaining == 0 {
4266 break;
4267 }
4268 Some(remaining as u32)
4269 } else {
4270 None
4271 };
4272
4273 let orders_for_coin = if open_only {
4274 let mut all_orders = Vec::new();
4275 let mut seen_ids: AHashSet<Ustr> = AHashSet::new();
4276
4277 let order_filters: Vec<Option<BybitOrderFilter>> =
4280 if product_type == BybitProductType::Option {
4281 vec![None]
4282 } else {
4283 vec![None, Some(BybitOrderFilter::StopOrder)]
4284 };
4285
4286 let open_only_modes = [None, Some(BybitOpenOnly::ClosedRecent)];
4287
4288 for oo in open_only_modes {
4289 for order_filter in &order_filters {
4290 let mut cursor: Option<String> = None;
4291
4292 loop {
4293 let remaining = if let Some(limit) = remaining_limit {
4294 (limit as usize).saturating_sub(all_orders.len())
4295 } else {
4296 usize::MAX
4297 };
4298
4299 if remaining == 0 {
4300 break;
4301 }
4302
4303 let page_limit = std::cmp::min(remaining, 50);
4305
4306 let mut p = BybitOpenOrdersParamsBuilder::default();
4307 p.category(product_type);
4308
4309 if let Some(symbol) = symbol_param.clone() {
4310 p.symbol(symbol);
4311 }
4312
4313 if let Some(coin) = settle_coin.clone() {
4314 p.settle_coin(coin);
4315 }
4316
4317 if let Some(of) = order_filter {
4318 p.order_filter(*of);
4319 }
4320
4321 if let Some(oo) = oo {
4322 p.open_only(oo);
4323 }
4324 p.limit(page_limit as u32);
4325
4326 if let Some(c) = cursor {
4327 p.cursor(c);
4328 }
4329 let params = p.build().build_anyhow()?;
4330 let response: BybitOpenOrdersResponse = self
4331 .inner
4332 .send_request(
4333 Method::GET,
4334 BYBIT_ORDER_REALTIME,
4335 Some(¶ms),
4336 None,
4337 true,
4338 )
4339 .await?;
4340
4341 for order in response.result.list {
4342 if seen_ids.insert(order.order_id) {
4343 all_orders.push(order);
4344 }
4345 }
4346
4347 if oo.is_some() {
4349 break;
4350 }
4351
4352 cursor = response.result.next_page_cursor;
4353 if cursor.as_ref().is_none_or(|c| c.is_empty()) {
4354 break;
4355 }
4356 }
4357 }
4358 }
4359
4360 all_orders
4361 } else {
4362 let mut all_orders = Vec::new();
4365 let mut open_orders = Vec::new();
4366 let mut seen_open_ids: AHashSet<Ustr> = AHashSet::new();
4367
4368 let order_filters: Vec<Option<BybitOrderFilter>> =
4371 if product_type == BybitProductType::Option {
4372 vec![None]
4373 } else {
4374 vec![None, Some(BybitOrderFilter::StopOrder)]
4375 };
4376
4377 for order_filter in &order_filters {
4378 let mut cursor: Option<String> = None;
4379
4380 loop {
4381 let remaining = if let Some(limit) = remaining_limit {
4382 (limit as usize).saturating_sub(open_orders.len())
4383 } else {
4384 usize::MAX
4385 };
4386
4387 if remaining == 0 {
4388 break;
4389 }
4390
4391 let page_limit = std::cmp::min(remaining, 50);
4393
4394 let mut open_params = BybitOpenOrdersParamsBuilder::default();
4395 open_params.category(product_type);
4396
4397 if let Some(symbol) = symbol_param.clone() {
4398 open_params.symbol(symbol);
4399 }
4400
4401 if let Some(coin) = settle_coin.clone() {
4402 open_params.settle_coin(coin);
4403 }
4404
4405 if let Some(of) = order_filter {
4406 open_params.order_filter(*of);
4407 }
4408 open_params.limit(page_limit as u32);
4409
4410 if let Some(c) = cursor {
4411 open_params.cursor(c);
4412 }
4413 let open_params = open_params.build().build_anyhow()?;
4414 let open_response: BybitOpenOrdersResponse = self
4415 .inner
4416 .send_request(
4417 Method::GET,
4418 BYBIT_ORDER_REALTIME,
4419 Some(&open_params),
4420 None,
4421 true,
4422 )
4423 .await?;
4424
4425 for order in open_response.result.list {
4426 if !seen_open_ids.contains(&order.order_id) {
4427 seen_open_ids.insert(order.order_id);
4428 open_orders.push(order);
4429 }
4430 }
4431
4432 cursor = open_response.result.next_page_cursor;
4433 if cursor.as_ref().is_none_or(|c| c.is_empty()) {
4434 break;
4435 }
4436 }
4437 }
4438
4439 let seen_order_ids: AHashSet<Ustr> = seen_open_ids;
4440 let total_open_orders = open_orders.len();
4441
4442 all_orders.extend(open_orders);
4443
4444 let mut total_history_orders = 0;
4445
4446 for order_filter in &order_filters {
4447 let mut cursor: Option<String> = None;
4448
4449 loop {
4450 let total_orders = total_open_orders + total_history_orders;
4451 let remaining = if let Some(limit) = remaining_limit {
4452 (limit as usize).saturating_sub(total_orders)
4453 } else {
4454 usize::MAX
4455 };
4456
4457 if remaining == 0 {
4458 break;
4459 }
4460
4461 let page_limit = std::cmp::min(remaining, 50);
4463
4464 let mut history_params = BybitOrderHistoryParamsBuilder::default();
4465 history_params.category(product_type);
4466
4467 if let Some(symbol) = symbol_param.clone() {
4468 history_params.symbol(symbol);
4469 }
4470
4471 if let Some(coin) = settle_coin.clone() {
4472 history_params.settle_coin(coin);
4473 }
4474
4475 if let Some(of) = order_filter {
4476 history_params.order_filter(*of);
4477 }
4478
4479 if let Some(start) = start {
4480 history_params.start_time(start.as_millisecond());
4481 }
4482
4483 if let Some(end) = end {
4484 history_params.end_time(end.as_millisecond());
4485 }
4486 history_params.limit(page_limit as u32);
4487
4488 if let Some(c) = cursor {
4489 history_params.cursor(c);
4490 }
4491 let history_params = history_params.build().build_anyhow()?;
4492 let history_response: BybitOrderHistoryResponse = self
4493 .inner
4494 .send_request(
4495 Method::GET,
4496 BYBIT_ORDER_HISTORY,
4497 Some(&history_params),
4498 None,
4499 true,
4500 )
4501 .await?;
4502
4503 for order in history_response.result.list {
4505 if !seen_order_ids.contains(&order.order_id) {
4506 all_orders.push(order);
4507 total_history_orders += 1;
4508 }
4509 }
4510
4511 cursor = history_response.result.next_page_cursor;
4512 if cursor.as_ref().is_none_or(|c| c.is_empty()) {
4513 break;
4514 }
4515 }
4516 }
4517
4518 all_orders
4519 };
4520
4521 total_collected_across_coins += orders_for_coin.len();
4522 all_collected_orders.extend(orders_for_coin);
4523 }
4524
4525 let ts_init = self.generate_ts_init();
4526
4527 let mut reports = Vec::new();
4528
4529 for order in all_collected_orders {
4530 if let Some(ref instrument_id) = instrument_id {
4531 let instrument = self.instrument_from_cache(&instrument_id.symbol)?;
4532
4533 if let Ok(report) =
4534 parse_order_status_report(&order, &instrument, account_id, ts_init)
4535 {
4536 reports.push(report);
4537 }
4538 } else {
4539 if !order.symbol.is_empty() {
4542 let symbol_with_product =
4543 Symbol::from_ustr_unchecked(make_bybit_symbol(order.symbol, product_type));
4544
4545 let Ok(instrument) = self.instrument_from_cache(&symbol_with_product) else {
4546 log::debug!(
4547 "Skipping order report for instrument not in cache: symbol={}, full_symbol={}",
4548 order.symbol,
4549 symbol_with_product
4550 );
4551 continue;
4552 };
4553
4554 match parse_order_status_report(&order, &instrument, account_id, ts_init) {
4555 Ok(report) => reports.push(report),
4556 Err(e) => {
4557 log::error!("Failed to parse order status report: {e}");
4558 }
4559 }
4560 }
4561 }
4562 }
4563
4564 Ok(reports)
4565 }
4566
4567 pub async fn request_fill_reports(
4579 &self,
4580 account_id: AccountId,
4581 product_type: BybitProductType,
4582 instrument_id: Option<InstrumentId>,
4583 start: Option<i64>,
4584 end: Option<i64>,
4585 limit: Option<u32>,
4586 ) -> anyhow::Result<Vec<FillReport>> {
4587 let symbol = if let Some(id) = instrument_id {
4589 let bybit_symbol = BybitSymbol::new(id.symbol.as_str())?;
4590 Some(bybit_symbol.raw_symbol().to_string())
4591 } else {
4592 None
4593 };
4594
4595 let mut all_executions = Vec::new();
4597 let mut cursor: Option<String> = None;
4598 let mut total_executions = 0;
4599
4600 loop {
4601 let remaining = if let Some(limit) = limit {
4603 (limit as usize).saturating_sub(total_executions)
4604 } else {
4605 usize::MAX
4606 };
4607
4608 if remaining == 0 {
4610 break;
4611 }
4612
4613 let page_limit = std::cmp::min(remaining, 100);
4615
4616 let params = BybitTradeHistoryParams {
4617 category: product_type,
4618 symbol: symbol.clone(),
4619 base_coin: None,
4620 order_id: None,
4621 order_link_id: None,
4622 start_time: start,
4623 end_time: end,
4624 exec_type: None,
4625 limit: Some(page_limit as u32),
4626 cursor: cursor.clone(),
4627 };
4628
4629 let response = self.inner.get_trade_history(¶ms).await?;
4630 let list_len = response.result.list.len();
4631 all_executions.extend(response.result.list);
4632 total_executions += list_len;
4633
4634 cursor = response.result.next_page_cursor;
4635 if cursor.is_none() || cursor.as_ref().is_none_or(|c| c.is_empty()) {
4636 break;
4637 }
4638 }
4639
4640 let ts_init = self.generate_ts_init();
4641 let mut reports = Vec::new();
4642
4643 for execution in all_executions {
4644 let symbol_with_product =
4647 Symbol::from_ustr_unchecked(make_bybit_symbol(execution.symbol, product_type));
4648
4649 let Ok(instrument) = self.instrument_from_cache(&symbol_with_product) else {
4650 log::debug!(
4651 "Skipping fill report for instrument not in cache: symbol={}, full_symbol={}",
4652 execution.symbol,
4653 symbol_with_product
4654 );
4655 continue;
4656 };
4657
4658 match parse_fill_report(&execution, account_id, &instrument, ts_init) {
4659 Ok(report) => reports.push(report),
4660 Err(e) => {
4661 log::error!("Failed to parse fill report: {e}");
4662 }
4663 }
4664 }
4665
4666 Ok(reports)
4667 }
4668
4669 pub async fn request_position_status_reports(
4682 &self,
4683 account_id: AccountId,
4684 product_type: BybitProductType,
4685 instrument_id: Option<InstrumentId>,
4686 ) -> anyhow::Result<Vec<PositionStatusReport>> {
4687 if product_type == BybitProductType::Spot {
4689 if self.use_spot_position_reports.load(Ordering::Relaxed) {
4690 let Some(instrument_id) = instrument_id else {
4691 anyhow::bail!(
4692 "SPOT wallet balances carry no pair identity and cannot be attributed for a bulk position report request"
4693 );
4694 };
4695 return self
4696 .generate_spot_position_reports_from_wallet(account_id, instrument_id)
4697 .await;
4698 } else {
4699 return Ok(Vec::new());
4701 }
4702 }
4703
4704 let ts_init = self.generate_ts_init();
4705 let mut reports = Vec::new();
4706
4707 let symbol = if let Some(id) = instrument_id {
4709 let symbol_str = id.symbol.as_str();
4710 if symbol_str.is_empty() {
4711 anyhow::bail!("InstrumentId symbol is empty");
4712 }
4713 let bybit_symbol = BybitSymbol::new(symbol_str)?;
4714 Some(bybit_symbol.raw_symbol().to_string())
4715 } else {
4716 None
4717 };
4718
4719 if product_type == BybitProductType::Linear && symbol.is_none() {
4722 for settle_coin in ["USDT", "USDC"] {
4724 let mut cursor: Option<String> = None;
4725
4726 loop {
4727 let params = BybitPositionListParams {
4728 category: product_type,
4729 symbol: None,
4730 base_coin: None,
4731 settle_coin: Some(settle_coin.to_string()),
4732 limit: Some(200), cursor: cursor.clone(),
4734 };
4735
4736 let response = self.inner.get_positions(¶ms).await?;
4737
4738 for position in response.result.list {
4739 if position.symbol.is_empty() {
4740 continue;
4741 }
4742
4743 let symbol_with_product = Symbol::new(format!(
4744 "{}{}",
4745 position.symbol.as_str(),
4746 product_type.suffix()
4747 ));
4748
4749 let Ok(instrument) = self.instrument_from_cache(&symbol_with_product)
4750 else {
4751 log::debug!(
4752 "Skipping position report for instrument not in cache: symbol={}, full_symbol={}",
4753 position.symbol,
4754 symbol_with_product
4755 );
4756 continue;
4757 };
4758
4759 match parse_position_status_report(
4760 &position,
4761 account_id,
4762 &instrument,
4763 ts_init,
4764 ) {
4765 Ok(report) => reports.push(report),
4766 Err(e) => {
4767 log::error!("Failed to parse position status report: {e}");
4768 }
4769 }
4770 }
4771
4772 cursor = response.result.next_page_cursor;
4773 if cursor.as_ref().is_none_or(|c| c.is_empty()) {
4774 break;
4775 }
4776 }
4777 }
4778 } else {
4779 let mut cursor: Option<String> = None;
4781
4782 loop {
4783 let params = BybitPositionListParams {
4784 category: product_type,
4785 symbol: symbol.clone(),
4786 base_coin: None,
4787 settle_coin: None,
4788 limit: Some(200), cursor: cursor.clone(),
4790 };
4791
4792 let response = self.inner.get_positions(¶ms).await?;
4793
4794 for position in response.result.list {
4795 if position.symbol.is_empty() {
4796 continue;
4797 }
4798
4799 let symbol_with_product = Symbol::new(format!(
4800 "{}{}",
4801 position.symbol.as_str(),
4802 product_type.suffix()
4803 ));
4804
4805 let Ok(instrument) = self.instrument_from_cache(&symbol_with_product) else {
4806 log::debug!(
4807 "Skipping position report for instrument not in cache: symbol={}, full_symbol={}",
4808 position.symbol,
4809 symbol_with_product
4810 );
4811 continue;
4812 };
4813
4814 match parse_position_status_report(&position, account_id, &instrument, ts_init)
4815 {
4816 Ok(report) => reports.push(report),
4817 Err(e) => {
4818 log::error!("Failed to parse position status report: {e}");
4819 }
4820 }
4821 }
4822
4823 cursor = response.result.next_page_cursor;
4824 if cursor.is_none() || cursor.as_ref().is_none_or(|c| c.is_empty()) {
4825 break;
4826 }
4827 }
4828 }
4829
4830 Ok(reports)
4831 }
4832
4833 async fn query_order_by_id(
4834 &self,
4835 product_type: BybitProductType,
4836 order_id: &str,
4837 endpoint: &str,
4838 context: &str,
4839 ) -> anyhow::Result<BybitOrder> {
4840 let mut query_params = BybitOpenOrdersParamsBuilder::default();
4841 query_params.category(product_type);
4842 query_params.order_id(order_id.to_string());
4843
4844 let query_params = query_params.build().build_anyhow()?;
4845 let order_response: BybitOpenOrdersResponse = self
4846 .inner
4847 .send_request(Method::GET, endpoint, Some(&query_params), None, true)
4848 .await?;
4849
4850 order_response
4851 .result
4852 .list
4853 .into_iter()
4854 .next()
4855 .ok_or_else(|| anyhow::anyhow!("No order returned {context}"))
4856 }
4857}
4858
4859#[cfg(test)]
4860mod tests {
4861 use rstest::rstest;
4862
4863 use super::*;
4864
4865 #[rstest]
4866 fn test_client_creation() {
4867 let client = BybitHttpClient::new(None, 60, 3, 1000, 10_000, 5_000, None);
4868 assert!(client.is_ok());
4869
4870 let client = client.unwrap();
4871 assert!(client.base_url().contains("bybit.com"));
4872 assert!(client.credential().is_none());
4873 }
4874
4875 #[rstest]
4876 fn test_client_with_credentials() {
4877 let client = BybitHttpClient::with_credentials(
4878 "test_key".to_string(),
4879 "test_secret".to_string(),
4880 Some("https://api-testnet.bybit.com".to_string()),
4881 60,
4882 3,
4883 1000,
4884 10_000,
4885 5_000,
4886 None,
4887 );
4888 assert!(client.is_ok());
4889
4890 let client = client.unwrap();
4891 assert!(client.credential().is_some());
4892 }
4893
4894 #[rstest]
4895 fn test_build_path_with_params() {
4896 #[derive(Serialize)]
4897 struct TestParams {
4898 category: String,
4899 symbol: String,
4900 }
4901
4902 let params = TestParams {
4903 category: "linear".to_string(),
4904 symbol: "BTCUSDT".to_string(),
4905 };
4906
4907 let path = BybitRawHttpClient::build_path("/v5/market/test", ¶ms);
4908 assert!(path.is_ok());
4909 assert!(path.unwrap().contains("category=linear"));
4910 }
4911
4912 #[rstest]
4913 fn test_build_path_without_params() {
4914 let params = ();
4915 let path = BybitRawHttpClient::build_path("/v5/market/time", ¶ms);
4916 assert!(path.is_ok());
4917 assert_eq!(path.unwrap(), "/v5/market/time");
4918 }
4919
4920 #[rstest]
4921 fn test_params_serialization_matches_build_path() {
4922 #[derive(Serialize)]
4924 struct TestParams {
4925 category: String,
4926 limit: u32,
4927 }
4928
4929 let params = TestParams {
4930 category: "spot".to_string(),
4931 limit: 50,
4932 };
4933
4934 let old_path = BybitRawHttpClient::build_path(BYBIT_ORDER_REALTIME, ¶ms).unwrap();
4936 let old_query = old_path.split('?').nth(1).unwrap_or("");
4937
4938 let new_query = serde_urlencoded::to_string(¶ms).unwrap();
4940
4941 assert_eq!(old_query, new_query);
4943 }
4944
4945 #[rstest]
4946 fn test_params_serialization_order() {
4947 #[derive(Serialize)]
4949 struct OrderParams {
4950 category: String,
4951 symbol: String,
4952 limit: u32,
4953 }
4954
4955 let params = OrderParams {
4956 category: "spot".to_string(),
4957 symbol: "BTCUSDT".to_string(),
4958 limit: 50,
4959 };
4960
4961 let query1 = serde_urlencoded::to_string(¶ms).unwrap();
4963 let query2 = serde_urlencoded::to_string(¶ms).unwrap();
4964 let query3 = serde_urlencoded::to_string(¶ms).unwrap();
4965
4966 assert_eq!(query1, query2);
4967 assert_eq!(query2, query3);
4968
4969 assert!(query1.contains("category=spot"));
4971 assert!(query1.contains("symbol=BTCUSDT"));
4972 assert!(query1.contains("limit=50"));
4973 }
4974
4975 #[rstest]
4976 #[case(403, "Access too frequent", true)]
4977 #[case(403, "Forbidden", false)]
4978 #[case(429, "Access too frequent", false)]
4979 fn test_rate_limit_403_detection(
4980 #[case] status: u16,
4981 #[case] body: &str,
4982 #[case] expected: bool,
4983 ) {
4984 assert_eq!(
4985 BybitRawHttpClient::is_rate_limit_403(status, body),
4986 expected
4987 );
4988 }
4989
4990 #[rstest]
4991 #[case(
4992 "https://api-demo.bybit.com",
4993 BybitProductType::Linear,
4994 10001,
4995 "",
4996 "Bybit demo rejected the linear fee rate request via /v5/account/fee-rate \
4997 (error 10001, no message); demo derivatives fee rates appear unsupported, using defaults"
4998 )]
4999 #[case(
5000 "https://api-demo.bybit.com",
5001 BybitProductType::Inverse,
5002 10001,
5003 "",
5004 "Bybit demo rejected the inverse fee rate request via /v5/account/fee-rate \
5005 (error 10001, no message); demo derivatives fee rates appear unsupported, using defaults"
5006 )]
5007 #[case(
5008 "https://api.bybit.com",
5009 BybitProductType::Spot,
5010 10001,
5011 "Parameter error",
5012 "Fee rate request rejected for spot instruments via /v5/account/fee-rate \
5013 (error 10001: Parameter error), using defaults"
5014 )]
5015 #[case(
5016 "https://api-demo.bybit.com",
5017 BybitProductType::Spot,
5018 10001,
5019 "Parameter error",
5020 "Fee rate request rejected for spot instruments via /v5/account/fee-rate \
5021 (error 10001: Parameter error), using defaults"
5022 )]
5023 #[case(
5024 "https://api.bybit.com",
5025 BybitProductType::Linear,
5026 10001,
5027 "Parameter error",
5028 "Fee rate request rejected for linear instruments via /v5/account/fee-rate \
5029 (error 10001: Parameter error), using defaults"
5030 )]
5031 fn test_fee_rate_rejection_warning(
5032 #[case] base_url: &str,
5033 #[case] product_type: BybitProductType,
5034 #[case] error_code: i32,
5035 #[case] message: &str,
5036 #[case] expected: &str,
5037 ) {
5038 let client =
5039 BybitHttpClient::new(Some(base_url.to_string()), 60, 3, 1000, 10_000, 5_000, None)
5040 .unwrap();
5041
5042 let warning = client.fee_rate_rejection_warning(product_type, error_code, message);
5043
5044 assert_eq!(warning, expected);
5045 }
5046
5047 #[rstest]
5048 #[case(10001, "", "error 10001, no message")]
5049 #[case(10001, "Parameter error", "error 10001: Parameter error")]
5050 fn test_format_bybit_error_detail(
5051 #[case] error_code: i32,
5052 #[case] message: &str,
5053 #[case] expected: &str,
5054 ) {
5055 let detail = BybitHttpClient::format_bybit_error_detail(error_code, message);
5056
5057 assert_eq!(detail, expected);
5058 }
5059
5060 #[rstest]
5061 #[case(
5062 10001,
5063 "",
5064 "Option fee rate request rejected via /v5/account/fee-rate \
5065 (error 10001, no message), using defaults"
5066 )]
5067 #[case(
5068 10001,
5069 "Parameter error",
5070 "Option fee rate request rejected via /v5/account/fee-rate \
5071 (error 10001: Parameter error), using defaults"
5072 )]
5073 fn test_option_fee_rate_warning_message(
5074 #[case] error_code: i32,
5075 #[case] message: &str,
5076 #[case] expected: &str,
5077 ) {
5078 let error_detail = BybitHttpClient::format_bybit_error_detail(error_code, message);
5079 let warning = format!(
5080 "Option fee rate request rejected via /v5/account/fee-rate ({error_detail}), using defaults"
5081 );
5082
5083 assert_eq!(warning, expected);
5084 }
5085}