1use super::BinanceMargin;
7use crate::{
8 cash_flow::{
9 CashFlow, CashFlowId, CashFlowKind, CashFlowPeriod, CashFlowsRead, InterestTiming,
10 },
11 client::binance::shared::{
12 RequestKind, classify_rest_query_error, parse_side, response_decode_error,
13 rest_call_with_retry,
14 },
15 error::UnindexedClientError,
16 order::id::OrderId,
17};
18use binance_sdk::margin_trading::rest_api::{
19 GetForceLiquidationRecordParams, GetForceLiquidationRecordResponseRowsInner,
20 GetInterestHistoryParams, GetInterestHistoryResponseRowsInner,
21};
22use chrono::{DateTime, TimeDelta, TimeZone, Utc};
23use fnv::FnvHashSet;
24use futures::{StreamExt as _, TryStreamExt as _};
25use rust_decimal::Decimal;
26use rustrade_instrument::{
27 Side, asset::name::AssetNameExchange, instrument::name::InstrumentNameExchange,
28};
29use serde::{Deserialize, Serialize};
30use smol_str::{SmolStr, format_smolstr};
31use std::{future::Future, str::FromStr};
32use tracing::{debug, warn};
33
34const INTEREST_LOOKBACK: TimeDelta = TimeDelta::days(90);
36
37const INTEREST_LOOKBACK_MARGIN: TimeDelta = TimeDelta::minutes(1);
40
41fn interest_held_since(start: DateTime<Utc>, now: DateTime<Utc>) -> Option<DateTime<Utc>> {
45 let held_since = now - INTEREST_LOOKBACK + INTEREST_LOOKBACK_MARGIN;
46 (start < held_since).then_some(held_since)
47}
48
49const INTEREST_WINDOW_MS: i64 = 29 * 24 * 60 * 60 * 1000;
52
53const PAGE_SIZE: i64 = 100;
55
56const MAX_PAGES: i64 = 500;
60
61const CONCURRENT_PAIRS: usize = 8;
63
64#[non_exhaustive]
69#[derive(Debug, Clone, PartialEq, Eq, Hash, Deserialize, Serialize)]
70pub struct MarginForcedLiquidation {
71 pub order_id: OrderId,
77 pub instrument: InstrumentNameExchange,
79 pub side: Side,
81 pub price: Option<Decimal>,
83 pub quantity: Option<Decimal>,
85 pub filled_quantity: Decimal,
87 pub avg_price: Option<Decimal>,
89 pub time_exchange: DateTime<Utc>,
91}
92
93impl BinanceMargin {
94 pub async fn fetch_margin_interest(
195 &self,
196 start: DateTime<Utc>,
197 end: DateTime<Utc>,
198 ) -> Result<CashFlowsRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError>
199 {
200 let now = Utc::now();
201 let held_since = interest_held_since(start, now);
202 if let Some(held_since) = held_since {
203 debug!(
204 requested = %start,
205 %held_since,
206 "BinanceMargin fetch_margin_interest: Binance serves 90 days of interest \
207 history — reading from the earliest time it serves"
208 );
209 }
210 let start = held_since.unwrap_or(start);
211 let end = end.min(now);
212 if start > end {
213 return Ok(CashFlowsRead::complete(Vec::new()).with_held_since(held_since));
214 }
215 let (start_ms, end_ms) = (start.timestamp_millis(), end.timestamp_millis());
216
217 let rows = self
218 .read_each_account(|pair| self.interest_rows(pair, start_ms, end_ms))
219 .await?;
220
221 let mut seen = FnvHashSet::default();
222 let mut flows: Vec<_> = rows
223 .into_iter()
224 .flat_map(|(pair, rows)| {
225 rows.into_iter()
226 .filter_map(move |row| convert_interest(row, pair.as_ref()))
227 })
228 .filter(|flow| (start..=end).contains(&flow.time_exchange))
229 .filter(|flow| seen.insert(ChargeKey::of(flow)))
232 .collect();
233 flows.sort_by_key(|flow| flow.time_exchange);
234 Ok(CashFlowsRead::complete(flows).with_held_since(held_since))
235 }
236
237 pub async fn fetch_forced_liquidations(
268 &self,
269 start: DateTime<Utc>,
270 end: DateTime<Utc>,
271 ) -> Result<Vec<MarginForcedLiquidation>, UnindexedClientError> {
272 let end = end.min(Utc::now());
273 if start > end {
274 return Ok(Vec::new());
275 }
276 let (start_ms, end_ms) = (start.timestamp_millis(), end.timestamp_millis());
277
278 let is_isolated = self.config.is_isolated;
279 let rows = self
280 .read_each_account(|pair| {
281 read_all_pages(move |current| {
282 self.forced_liquidation_page(pair.clone(), start_ms, end_ms, current)
283 })
284 })
285 .await?;
286
287 let mut seen = FnvHashSet::default();
288 let mut liquidations: Vec<_> = rows
289 .into_iter()
290 .flat_map(|(_, rows)| rows)
291 .filter_map(|row| convert_forced_liquidation(row, is_isolated))
292 .filter(|liquidation| (start..=end).contains(&liquidation.time_exchange))
293 .filter(|liquidation| {
296 seen.insert((liquidation.instrument.clone(), liquidation.order_id.clone()))
297 })
298 .collect();
299 liquidations.sort_by_key(|liquidation| liquidation.time_exchange);
300 Ok(liquidations)
301 }
302
303 async fn read_each_account<'a, Row, Fut>(
306 &'a self,
307 read: impl Fn(Option<InstrumentNameExchange>) -> Fut,
308 ) -> Result<Vec<(Option<InstrumentNameExchange>, Vec<Row>)>, UnindexedClientError>
309 where
310 Fut: Future<Output = Result<Vec<Row>, UnindexedClientError>> + 'a,
311 {
312 let accounts: Vec<Option<InstrumentNameExchange>> = if self.config.is_isolated {
313 self.config
314 .isolated_symbols
315 .iter()
316 .cloned()
317 .map(Some)
318 .collect()
319 } else {
320 vec![None]
321 };
322 futures::stream::iter(accounts.into_iter().map(|pair| {
323 let rows = read(pair.clone());
324 async move { Ok::<_, UnindexedClientError>((pair, rows.await?)) }
325 }))
326 .buffer_unordered(CONCURRENT_PAIRS)
327 .try_collect()
328 .await
329 }
330
331 async fn interest_rows(
333 &self,
334 pair: Option<InstrumentNameExchange>,
335 start_ms: i64,
336 end_ms: i64,
337 ) -> Result<Vec<GetInterestHistoryResponseRowsInner>, UnindexedClientError> {
338 let mut rows = Vec::new();
339 let mut window_start = start_ms;
340 loop {
341 let window_end = window_start.saturating_add(INTEREST_WINDOW_MS).min(end_ms);
342 let pair = &pair;
343 rows.extend(
344 read_all_pages(|current| {
345 self.interest_page(pair.clone(), window_start, window_end, current)
346 })
347 .await?,
348 );
349 if window_end >= end_ms {
350 break;
351 }
352 window_start = window_end;
355 }
356 debug!(
357 pair = ?pair,
358 rows = rows.len(),
359 "BinanceMargin fetch_margin_interest: account read"
360 );
361 Ok(rows)
362 }
363
364 async fn interest_page(
366 &self,
367 pair: Option<InstrumentNameExchange>,
368 start_ms: i64,
369 end_ms: i64,
370 current: i64,
371 ) -> Result<Vec<GetInterestHistoryResponseRowsInner>, UnindexedClientError> {
372 let symbol = pair.as_ref().map(|pair| pair.name().to_string());
373 let response =
374 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
375 let symbol = symbol.clone();
376 Box::pin(async move {
377 let params = GetInterestHistoryParams::builder()
378 .isolated_symbol(symbol)
379 .start_time(start_ms)
380 .end_time(end_ms)
381 .current(current)
382 .size(PAGE_SIZE)
383 .build()?;
384 rest.get_interest_history(params).await
385 })
386 })
387 .await
388 .map_err(|e| classify_rest_query_error(&e, pair.as_ref()))?;
389 let page = response.data().await.map_err(response_decode_error)?;
390 Ok(page.rows.unwrap_or_default())
391 }
392
393 async fn forced_liquidation_page(
395 &self,
396 pair: Option<InstrumentNameExchange>,
397 start_ms: i64,
398 end_ms: i64,
399 current: i64,
400 ) -> Result<Vec<GetForceLiquidationRecordResponseRowsInner>, UnindexedClientError> {
401 let symbol = pair.as_ref().map(|pair| pair.name().to_string());
402 let response =
403 rest_call_with_retry(&self.rest, &self.rate_limiter, RequestKind::Query, |rest| {
404 let symbol = symbol.clone();
405 Box::pin(async move {
406 let params = GetForceLiquidationRecordParams::builder()
407 .isolated_symbol(symbol)
408 .start_time(start_ms)
409 .end_time(end_ms)
410 .current(current)
411 .size(PAGE_SIZE)
412 .build()?;
413 rest.get_force_liquidation_record(params).await
414 })
415 })
416 .await
417 .map_err(|e| classify_rest_query_error(&e, pair.as_ref()))?;
418 let page = response.data().await.map_err(response_decode_error)?;
419 Ok(page.rows.unwrap_or_default())
420 }
421}
422
423async fn read_all_pages<Row, Fut>(
426 mut page: impl FnMut(i64) -> Fut,
427) -> Result<Vec<Row>, UnindexedClientError>
428where
429 Fut: Future<Output = Result<Vec<Row>, UnindexedClientError>>,
430{
431 let mut rows = Vec::new();
432 for current in 1..=MAX_PAGES {
433 let batch = page(current).await?;
434 let full = i64::try_from(batch.len()).is_ok_and(|len| len >= PAGE_SIZE);
435 rows.extend(batch);
436 if !full {
437 return Ok(rows);
438 }
439 }
440 Err(UnindexedClientError::Internal(format!(
441 "Binance margin history query still full after {MAX_PAGES} pages of {PAGE_SIZE}: \
442 narrow the span"
443 )))
444}
445
446#[derive(PartialEq, Eq, Hash)]
449enum ChargeKey {
450 Id(
451 AssetNameExchange,
452 Option<InstrumentNameExchange>,
453 CashFlowId,
454 ),
455 Flow(CashFlow<AssetNameExchange, InstrumentNameExchange>),
456}
457
458impl ChargeKey {
459 fn of(flow: &CashFlow<AssetNameExchange, InstrumentNameExchange>) -> Self {
460 match &flow.id {
461 Some(id) => Self::Id(flow.asset.clone(), flow.instrument.clone(), id.clone()),
462 None => Self::Flow(flow.clone()),
463 }
464 }
465}
466
467fn optional_decimal(record: &str, field: &str, value: Option<&str>) -> Option<Decimal> {
470 let value = value.filter(|value| !value.is_empty())?;
471 Decimal::from_str(value)
472 .inspect_err(|error| {
473 warn!(
474 record,
475 field,
476 value,
477 %error,
478 "BinanceMargin record field does not parse — left out"
479 );
480 })
481 .ok()
482}
483
484fn interest_timing(r#type: &str) -> InterestTiming {
487 match r#type {
488 "PERIODIC" | "PERIODIC_CONVERTED" | "PORTFOLIO" => InterestTiming::Periodic,
489 "ON_BORROW" | "ON_BORROW_CONVERTED" => InterestTiming::OnBorrow,
490 other => InterestTiming::Other(SmolStr::new(other)),
491 }
492}
493
494const INTEREST_HOUR: TimeDelta = TimeDelta::hours(1);
496
497fn interest_period(
502 r#type: Option<&str>,
503 time: DateTime<Utc>,
504 tx_id: Option<i64>,
505) -> Option<CashFlowPeriod> {
506 let hourly = matches!(
507 r#type,
508 Some("PERIODIC" | "PERIODIC_CONVERTED" | "ON_BORROW" | "ON_BORROW_CONVERTED")
509 );
510 if !hourly {
511 return None;
512 }
513 if time
514 .timestamp_millis()
515 .rem_euclid(INTEREST_HOUR.num_milliseconds())
516 != 0
517 {
518 warn!(
519 ?tx_id,
520 r#type,
521 %time,
522 "BinanceMargin hourly interest record not stamped on the hour — no period given"
523 );
524 return None;
525 }
526 CashFlowPeriod::new(time, time + INTEREST_HOUR)
527}
528
529fn convert_interest(
532 row: GetInterestHistoryResponseRowsInner,
533 pair: Option<&InstrumentNameExchange>,
534) -> Option<CashFlow<AssetNameExchange, InstrumentNameExchange>> {
535 let tx_id = row.tx_id;
536 let Some(asset) = row.asset.as_deref() else {
537 warn!(
538 ?tx_id,
539 ?pair,
540 "BinanceMargin interest record missing asset — skipped"
541 );
542 return None;
543 };
544 let Some(interest) = row
545 .interest
546 .as_deref()
547 .and_then(|s| Decimal::from_str(s).ok())
548 else {
549 warn!(
550 ?tx_id,
551 ?pair,
552 "BinanceMargin interest record missing/unparseable interest — skipped"
553 );
554 return None;
555 };
556 let Some(time_exchange) = row
557 .interest_accured_time
558 .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
559 else {
560 warn!(
561 ?tx_id,
562 ?pair,
563 "BinanceMargin interest record missing/unparseable time — skipped"
564 );
565 return None;
566 };
567 let instrument = row
568 .isolated_symbol
569 .as_deref()
570 .map(InstrumentNameExchange::new)
571 .or_else(|| pair.cloned());
572 let period = interest_period(row.r#type.as_deref(), time_exchange, tx_id);
573 Some(
574 CashFlow::new(
575 CashFlowKind::MarginInterest {
576 rate: optional_decimal("interest", "interestRate", row.interest_rate.as_deref()),
577 principal: optional_decimal("interest", "principal", row.principal.as_deref()),
578 timing: row.r#type.as_deref().map(interest_timing),
579 },
580 AssetNameExchange::new(asset),
581 -interest,
582 instrument,
583 time_exchange,
584 tx_id.map(|id| CashFlowId(format_smolstr!("{id}"))),
585 )
586 .with_period(period),
587 )
588}
589
590fn convert_forced_liquidation(
594 row: GetForceLiquidationRecordResponseRowsInner,
595 is_isolated: bool,
596) -> Option<MarginForcedLiquidation> {
597 let Some(order_id) = row.order_id else {
598 warn!(symbol = ?row.symbol, "BinanceMargin forced liquidation record missing orderId — skipped");
599 return None;
600 };
601 let Some(symbol) = row.symbol.as_deref() else {
602 warn!(
603 order_id,
604 "BinanceMargin forced liquidation record missing symbol — skipped"
605 );
606 return None;
607 };
608 if !is_isolated && row.is_isolated == Some(true) {
611 warn!(
612 order_id,
613 symbol,
614 "BinanceMargin cross forced liquidation read returned an isolated record — left out"
615 );
616 return None;
617 }
618 let Some(side) = row.side.as_deref().and_then(parse_side) else {
619 warn!(
620 order_id,
621 symbol, "BinanceMargin forced liquidation record missing/unknown side — skipped"
622 );
623 return None;
624 };
625 let Some(filled_quantity) = row
626 .executed_qty
627 .as_deref()
628 .and_then(|s| Decimal::from_str(s).ok())
629 else {
630 warn!(
631 order_id,
632 symbol,
633 "BinanceMargin forced liquidation record missing/unparseable executedQty — skipped"
634 );
635 return None;
636 };
637 let Some(time_exchange) = row
638 .updated_time
639 .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
640 else {
641 warn!(
642 order_id,
643 symbol,
644 "BinanceMargin forced liquidation record missing/unparseable updatedTime — skipped"
645 );
646 return None;
647 };
648 Some(MarginForcedLiquidation {
649 order_id: OrderId(format_smolstr!("{order_id}")),
650 instrument: InstrumentNameExchange::new(symbol),
651 side,
652 price: optional_decimal("forced liquidation", "price", row.price.as_deref()),
653 quantity: optional_decimal("forced liquidation", "qty", row.qty.as_deref()),
654 avg_price: (!filled_quantity.is_zero())
655 .then(|| optional_decimal("forced liquidation", "avgPrice", row.avg_price.as_deref()))
656 .flatten(),
657 filled_quantity,
658 time_exchange,
659 })
660}
661
662#[cfg(test)]
663#[allow(clippy::unwrap_used)] mod tests {
665 use super::{super::tests::margin_client_at, *};
666 use crate::error::ApiError;
667 use chrono::DurationRound;
668 use rust_decimal_macros::dec;
669 use serde_json::{Value, json};
670 use std::collections::HashMap;
671 use wiremock::{
672 Mock, MockServer, ResponseTemplate,
673 matchers::{method, path, query_param},
674 };
675
676 const INTEREST_PATH: &str = "/sapi/v1/margin/interestHistory";
677 const LIQUIDATION_PATH: &str = "/sapi/v1/margin/forceLiquidationRec";
678
679 fn interest_row(tx_id: i64, time: DateTime<Utc>, isolated: Option<&str>) -> Value {
680 let mut row = json!({
681 "txId": tx_id, "interestAccuredTime": time.timestamp_millis(), "asset": "USDT",
682 "rawAsset": "USDT", "principal": "1000", "interest": "0.0125",
683 "interestRate": "0.0000125", "type": "PERIODIC",
684 });
685 if let Some(symbol) = isolated {
686 row["isolatedSymbol"] = json!(symbol);
687 }
688 row
689 }
690
691 fn liquidation_row(
692 order_id: i64,
693 side: &str,
694 filled: &str,
695 time: DateTime<Utc>,
696 is_isolated: bool,
697 ) -> Value {
698 json!({
699 "avgPrice": "60000", "executedQty": filled, "orderId": order_id, "price": "59000",
700 "qty": "0.5", "side": side, "symbol": "BTCUSDT", "timeInForce": "GTC",
701 "isIsolated": is_isolated, "updatedTime": time.timestamp_millis(),
702 })
703 }
704
705 fn page(rows: Vec<Value>) -> ResponseTemplate {
706 let total = rows.len();
707 ResponseTemplate::new(200).set_body_json(json!({ "rows": rows, "total": total }))
708 }
709
710 async fn mount(server: &MockServer, at: &str, response: ResponseTemplate) {
711 Mock::given(method("GET"))
712 .and(path(at))
713 .respond_with(response)
714 .mount(server)
715 .await;
716 }
717
718 async fn mount_page(server: &MockServer, at: &str, current: &str, response: ResponseTemplate) {
720 Mock::given(method("GET"))
721 .and(path(at))
722 .and(query_param("current", current))
723 .respond_with(response)
724 .mount(server)
725 .await;
726 }
727
728 async fn queries(server: &MockServer) -> Vec<HashMap<String, String>> {
730 server
731 .received_requests()
732 .await
733 .unwrap()
734 .iter()
735 .map(|request| request.url.query_pairs().into_owned().collect())
736 .collect()
737 }
738
739 fn millis(query: &HashMap<String, String>, key: &str) -> i64 {
740 query[key].parse().unwrap()
741 }
742
743 #[tokio::test]
744 async fn interest_is_read_window_by_window_as_negative_flows_oldest_first_once_each() {
745 let server = MockServer::start().await;
746 let now = Utc::now();
747 let start = now - TimeDelta::days(40);
748 let boundary = start + TimeDelta::milliseconds(INTEREST_WINDOW_MS);
749 let mut no_interest = interest_row(9, now - TimeDelta::days(2), None);
750 no_interest.as_object_mut().unwrap().remove("interest");
751 let mut bad_rate = interest_row(3, now - TimeDelta::days(3), None);
752 bad_rate["interestRate"] = json!("n/a");
753 mount(
756 &server,
757 INTEREST_PATH,
758 page(vec![
759 interest_row(2, now - TimeDelta::days(1), None),
760 no_interest,
761 bad_rate,
762 interest_row(4, boundary, None),
763 interest_row(1, now - TimeDelta::days(35), None),
764 interest_row(0, now - TimeDelta::days(50), None),
765 ]),
766 )
767 .await;
768 let client = margin_client_at(&server, false);
769
770 let read = client
771 .fetch_margin_interest(start, now + TimeDelta::days(5))
772 .await
773 .unwrap();
774 assert_eq!((read.resume, read.held_since), (None, None));
775 let flows = read.flows;
776
777 let ids: Vec<_> = flows.iter().map(|flow| flow.id.clone().unwrap()).collect();
778 assert_eq!(ids, ["1", "4", "3", "2"].map(CashFlowId::new));
779 let flow = &flows[0];
780 assert_eq!(flow.amount, dec!(-0.0125), "a charge is paid");
781 assert_eq!(flow.asset, AssetNameExchange::new("USDT"));
782 assert_eq!(flow.instrument, None, "a cross charge names no pair");
783 assert_eq!(
784 flow.kind,
785 CashFlowKind::MarginInterest {
786 rate: Some(dec!(0.0000125)),
787 principal: Some(dec!(1000)),
788 timing: Some(InterestTiming::Periodic),
789 }
790 );
791 assert_eq!(
792 flows[2].kind,
793 CashFlowKind::MarginInterest {
794 rate: None,
795 principal: Some(dec!(1000)),
796 timing: Some(InterestTiming::Periodic),
797 },
798 "a rate that does not parse is left out, not the charge"
799 );
800
801 let queries = queries(&server).await;
802 let [first, second] = queries.as_slice() else {
803 panic!("two windows: {queries:?}");
804 };
805 assert_eq!(millis(first, "startTime"), start.timestamp_millis());
806 assert_eq!(millis(first, "endTime"), boundary.timestamp_millis());
807 assert_eq!(millis(second, "startTime"), boundary.timestamp_millis());
808 assert!(
809 millis(second, "endTime") <= Utc::now().timestamp_millis(),
810 "read up to the local clock, not `end`"
811 );
812 for query in [first, second] {
813 assert_eq!(query["current"], "1");
814 assert_eq!(query["size"], "100");
815 assert!(!query.contains_key("isolatedSymbol"), "{query:?}");
816 }
817 }
818
819 #[test]
820 fn interest_timing_is_read_from_the_record_type() {
821 for (r#type, expected) in [
822 ("PERIODIC", InterestTiming::Periodic),
823 ("PERIODIC_CONVERTED", InterestTiming::Periodic),
824 ("PORTFOLIO", InterestTiming::Periodic),
825 ("ON_BORROW", InterestTiming::OnBorrow),
826 ("ON_BORROW_CONVERTED", InterestTiming::OnBorrow),
827 (
828 "ON_LIQUIDATION",
829 InterestTiming::Other("ON_LIQUIDATION".into()),
830 ),
831 ] {
832 assert_eq!(interest_timing(r#type), expected, "{type}");
833 }
834 }
835
836 #[test]
837 fn an_hourly_charge_has_the_hour_from_its_stamp_as_its_period() {
838 let hour = Utc.with_ymd_and_hms(2026, 10, 22, 9, 0, 0).unwrap();
839 for r#type in [
840 "PERIODIC",
841 "PERIODIC_CONVERTED",
842 "ON_BORROW",
843 "ON_BORROW_CONVERTED",
844 ] {
845 assert_eq!(
846 interest_period(Some(r#type), hour, Some(1)),
847 CashFlowPeriod::new(hour, hour + TimeDelta::hours(1)),
848 "{type}"
849 );
850 }
851 }
852
853 #[test]
854 fn a_charge_of_another_type_or_off_the_hour_has_no_period() {
855 let hour = Utc.with_ymd_and_hms(2026, 10, 22, 9, 0, 0).unwrap();
856 for r#type in [Some("PORTFOLIO"), Some("ON_LIQUIDATION"), None] {
857 assert_eq!(interest_period(r#type, hour, Some(1)), None, "{type:?}");
858 }
859 for off in [TimeDelta::milliseconds(1), TimeDelta::minutes(31)] {
860 assert_eq!(
861 interest_period(Some("ON_BORROW"), hour + off, Some(1)),
862 None,
863 "{off}"
864 );
865 }
866 }
867
868 #[tokio::test]
869 async fn a_charge_on_borrow_read_from_binance_has_its_hour_as_its_period() {
870 let server = MockServer::start().await;
871 let now = Utc::now();
872 let hour = now.duration_trunc(TimeDelta::hours(1)).unwrap() - TimeDelta::hours(2);
874 let mut row = interest_row(1, hour, None);
875 row["type"] = json!("ON_BORROW");
876 mount(&server, INTEREST_PATH, page(vec![row])).await;
877 let client = margin_client_at(&server, false);
878
879 let flows = client
880 .fetch_margin_interest(now - TimeDelta::days(1), now)
881 .await
882 .unwrap()
883 .flows;
884
885 assert_eq!(flows.len(), 1);
886 assert_eq!(flows[0].time_exchange, hour);
887 assert_eq!(
888 flows[0].period,
889 CashFlowPeriod::new(hour, hour + TimeDelta::hours(1))
890 );
891 }
892
893 #[tokio::test]
894 async fn a_charge_on_borrow_has_its_timing() {
895 let server = MockServer::start().await;
896 let now = Utc::now();
897 let mut row = interest_row(1, now - TimeDelta::hours(1), None);
898 row["type"] = json!("ON_BORROW");
899 mount(&server, INTEREST_PATH, page(vec![row])).await;
900 let client = margin_client_at(&server, false);
901
902 let flows = client
903 .fetch_margin_interest(now - TimeDelta::days(1), now)
904 .await
905 .unwrap()
906 .flows;
907
908 let [flow] = flows.as_slice() else {
909 panic!("one flow: {flows:?}");
910 };
911 let CashFlowKind::MarginInterest { timing, .. } = &flow.kind else {
912 panic!("margin interest: {flow:?}");
913 };
914 assert_eq!(*timing, Some(InterestTiming::OnBorrow));
915 }
916
917 #[tokio::test]
920 async fn a_charge_converted_into_bnb_is_in_bnb() {
921 let server = MockServer::start().await;
922 let now = Utc::now();
923 let mut row = interest_row(1, now - TimeDelta::hours(1), None);
924 row["asset"] = json!("BNB");
925 row["principal"] = json!("0");
926 row["interest"] = json!("0.00002");
927 row["type"] = json!("PERIODIC_CONVERTED");
928 mount(&server, INTEREST_PATH, page(vec![row])).await;
929 let client = margin_client_at(&server, false);
930
931 let flows = client
932 .fetch_margin_interest(now - TimeDelta::days(1), now)
933 .await
934 .unwrap()
935 .flows;
936
937 let [flow] = flows.as_slice() else {
938 panic!("one flow: {flows:?}");
939 };
940 assert_eq!(flow.asset, AssetNameExchange::new("BNB"));
941 assert_eq!(flow.amount, dec!(-0.00002));
942 let CashFlowKind::MarginInterest {
943 principal, timing, ..
944 } = &flow.kind
945 else {
946 panic!("margin interest: {flow:?}");
947 };
948 assert_eq!(*principal, Some(Decimal::ZERO));
949 assert_eq!(*timing, Some(InterestTiming::Periodic));
950 }
951
952 #[tokio::test]
953 async fn a_charge_without_a_type_has_no_timing() {
954 let server = MockServer::start().await;
955 let now = Utc::now();
956 let mut row = interest_row(1, now - TimeDelta::hours(1), None);
957 row.as_object_mut().unwrap().remove("type");
958 mount(&server, INTEREST_PATH, page(vec![row])).await;
959 let client = margin_client_at(&server, false);
960
961 let flows = client
962 .fetch_margin_interest(now - TimeDelta::days(1), now)
963 .await
964 .unwrap()
965 .flows;
966
967 let [flow] = flows.as_slice() else {
968 panic!("one flow: {flows:?}");
969 };
970 let CashFlowKind::MarginInterest { timing, .. } = &flow.kind else {
971 panic!("margin interest: {flow:?}");
972 };
973 assert_eq!(*timing, None);
974 }
975
976 #[tokio::test]
977 async fn a_charge_repeated_with_its_id_is_kept_once() {
978 let server = MockServer::start().await;
979 let now = Utc::now();
980 let mut corrected = interest_row(1, now - TimeDelta::hours(1), None);
981 corrected["interest"] = json!("0.02");
982 mount(
983 &server,
984 INTEREST_PATH,
985 page(vec![
986 corrected,
987 interest_row(1, now - TimeDelta::hours(1), None),
988 ]),
989 )
990 .await;
991 let client = margin_client_at(&server, false);
992
993 let flows = client
994 .fetch_margin_interest(now - TimeDelta::days(1), now)
995 .await
996 .unwrap()
997 .flows;
998
999 let [flow] = flows.as_slice() else {
1000 panic!("one charge: {flows:?}");
1001 };
1002 assert_eq!(flow.amount, dec!(-0.02), "the first copy read is kept");
1003 }
1004
1005 #[tokio::test]
1006 async fn a_span_of_one_instant_is_read_in_one_request() {
1007 let server = MockServer::start().await;
1008 let at =
1010 DateTime::from_timestamp_millis((Utc::now() - TimeDelta::hours(1)).timestamp_millis())
1011 .unwrap();
1012 mount(
1013 &server,
1014 INTEREST_PATH,
1015 page(vec![interest_row(1, at, None)]),
1016 )
1017 .await;
1018 let client = margin_client_at(&server, false);
1019
1020 let flows = client.fetch_margin_interest(at, at).await.unwrap().flows;
1021
1022 assert_eq!(flows.len(), 1);
1023 let queries = queries(&server).await;
1024 let [query] = queries.as_slice() else {
1025 panic!("one request: {queries:?}");
1026 };
1027 assert_eq!(millis(query, "startTime"), at.timestamp_millis());
1028 assert_eq!(millis(query, "endTime"), at.timestamp_millis());
1029 }
1030
1031 #[tokio::test]
1032 async fn a_full_page_reads_the_next() {
1033 let server = MockServer::start().await;
1034 let now = Utc::now();
1035 let time = now - TimeDelta::hours(1);
1036 let full = (0..100).map(|id| interest_row(id, time, None)).collect();
1037 mount_page(&server, INTEREST_PATH, "1", page(full)).await;
1038 mount_page(
1039 &server,
1040 INTEREST_PATH,
1041 "2",
1042 page(vec![interest_row(100, time, None)]),
1043 )
1044 .await;
1045 let client = margin_client_at(&server, false);
1046
1047 let flows = client
1048 .fetch_margin_interest(now - TimeDelta::days(1), now)
1049 .await
1050 .unwrap()
1051 .flows;
1052
1053 assert_eq!(flows.len(), 101);
1054 assert_eq!(queries(&server).await.len(), 2);
1055 }
1056
1057 #[tokio::test]
1058 async fn a_query_still_full_after_the_most_pages_fails() {
1059 let server = MockServer::start().await;
1060 let now = Utc::now();
1061 let time = now - TimeDelta::hours(1);
1062 let full = (0..100).map(|id| interest_row(id, time, None)).collect();
1064 mount(&server, INTEREST_PATH, page(full)).await;
1065 let client = margin_client_at(&server, false);
1066
1067 let result = client
1068 .fetch_margin_interest(now - TimeDelta::days(1), now)
1069 .await;
1070
1071 assert!(
1072 matches!(result, Err(UnindexedClientError::Internal(_))),
1073 "{result:?}"
1074 );
1075 assert_eq!(
1076 i64::try_from(queries(&server).await.len()).unwrap(),
1077 MAX_PAGES
1078 );
1079 }
1080
1081 #[tokio::test]
1082 async fn isolated_interest_is_read_per_pair_and_attributed_to_it() {
1083 let server = MockServer::start().await;
1084 let now = Utc::now();
1085 mount(
1086 &server,
1087 INTEREST_PATH,
1088 page(vec![
1089 interest_row(2, now - TimeDelta::hours(1), Some("BTCUSDT")),
1090 interest_row(1, now - TimeDelta::hours(2), None),
1092 ]),
1093 )
1094 .await;
1095 let client = margin_client_at(&server, true);
1096
1097 let flows = client
1098 .fetch_margin_interest(now - TimeDelta::days(1), now)
1099 .await
1100 .unwrap()
1101 .flows;
1102
1103 assert_eq!(flows.len(), 2);
1104 for flow in flows {
1105 assert_eq!(
1106 flow.instrument,
1107 Some(InstrumentNameExchange::new("BTCUSDT"))
1108 );
1109 }
1110 let queries = queries(&server).await;
1111 assert_eq!(queries.len(), 1);
1112 assert_eq!(queries[0]["isolatedSymbol"], "BTCUSDT");
1113 }
1114
1115 #[tokio::test]
1116 async fn interest_older_than_binance_serves_is_read_from_its_horizon() {
1117 let server = MockServer::start().await;
1118 mount(&server, INTEREST_PATH, page(Vec::new())).await;
1119 let client = margin_client_at(&server, false);
1120 let before = Utc::now();
1121
1122 let read = client
1123 .fetch_margin_interest(before - TimeDelta::days(200), before)
1124 .await
1125 .unwrap();
1126 let after = Utc::now();
1127
1128 assert!(read.flows.is_empty());
1129 let queries = queries(&server).await;
1130 let first = millis(&queries[0], "startTime");
1131 let horizon = |now: DateTime<Utc>| {
1132 (now - INTEREST_LOOKBACK + INTEREST_LOOKBACK_MARGIN).timestamp_millis()
1133 };
1134 assert!(
1135 (horizon(before)..=horizon(after)).contains(&first),
1136 "{first}"
1137 );
1138 assert_eq!(
1139 read.held_since
1140 .map(|held_since| held_since.timestamp_millis()),
1141 Some(first),
1142 "held_since is where the read started"
1143 );
1144 assert_eq!(queries.len(), 4, "90 days in windows of 29");
1145 }
1146
1147 #[test]
1148 fn interest_is_held_since_a_minute_inside_ninety_days_only_for_an_older_start() {
1149 let now = Utc.with_ymd_and_hms(2026, 10, 25, 12, 0, 0).unwrap();
1150 let held_since = now - TimeDelta::days(90) + TimeDelta::minutes(1);
1151 let before = held_since - TimeDelta::milliseconds(1);
1152
1153 assert_eq!(interest_held_since(before, now), Some(held_since));
1154 assert_eq!(
1155 interest_held_since(now - TimeDelta::days(90), now),
1156 Some(held_since),
1157 "a request for the last 90 days misses its first minute"
1158 );
1159 assert_eq!(interest_held_since(held_since, now), None);
1160 assert_eq!(interest_held_since(now, now), None);
1161 }
1162
1163 #[tokio::test]
1164 async fn interest_entirely_older_than_binance_serves_reads_nothing_and_says_so() {
1165 let server = MockServer::start().await;
1166 let client = margin_client_at(&server, false);
1167 let end = Utc::now() - TimeDelta::days(100);
1168
1169 let read = client
1170 .fetch_margin_interest(end - TimeDelta::days(1), end)
1171 .await
1172 .unwrap();
1173
1174 assert!(read.flows.is_empty());
1175 assert!(
1176 read.held_since.is_some_and(|held_since| held_since > end),
1177 "{:?}",
1178 read.held_since
1179 );
1180 assert!(queries(&server).await.is_empty());
1181 }
1182
1183 #[tokio::test]
1184 async fn an_empty_span_reads_nothing() {
1185 let server = MockServer::start().await;
1186 let client = margin_client_at(&server, false);
1187 let now = Utc::now();
1188
1189 let interest = client
1190 .fetch_margin_interest(now, now - TimeDelta::hours(1))
1191 .await
1192 .unwrap();
1193 let liquidations = client
1194 .fetch_forced_liquidations(now + TimeDelta::hours(1), now + TimeDelta::hours(2))
1195 .await
1196 .unwrap();
1197
1198 assert!(interest.flows.is_empty() && liquidations.is_empty());
1199 assert_eq!((interest.resume, interest.held_since), (None, None));
1200 assert!(queries(&server).await.is_empty());
1201 }
1202
1203 #[tokio::test]
1204 async fn a_refused_read_fails_the_whole_read() {
1205 let server = MockServer::start().await;
1206 let refused =
1207 ResponseTemplate::new(400).set_body_json(json!({"code": -1100, "msg": "refused"}));
1208 mount(&server, INTEREST_PATH, refused.clone()).await;
1209 mount(&server, LIQUIDATION_PATH, refused).await;
1210 let client = margin_client_at(&server, false);
1211 let now = Utc::now();
1212
1213 let interest = client
1214 .fetch_margin_interest(now - TimeDelta::days(1), now)
1215 .await;
1216 let liquidations = client
1217 .fetch_forced_liquidations(now - TimeDelta::days(1), now)
1218 .await;
1219
1220 assert!(
1221 matches!(
1222 interest,
1223 Err(UnindexedClientError::Api(ApiError::RequestRejected(_)))
1224 ),
1225 "{interest:?}"
1226 );
1227 assert!(
1228 matches!(
1229 liquidations,
1230 Err(UnindexedClientError::Api(ApiError::RequestRejected(_)))
1231 ),
1232 "{liquidations:?}"
1233 );
1234 }
1235
1236 #[tokio::test]
1237 async fn forced_liquidations_of_the_clients_account_are_read_oldest_first_once_each() {
1238 for is_isolated in [false, true] {
1239 let server = MockServer::start().await;
1240 let now = Utc::now();
1241 let row = |order_id, side, filled, time| {
1242 liquidation_row(order_id, side, filled, time, is_isolated)
1243 };
1244 let mut no_order_id = row(9, "SELL", "0.5", now);
1245 no_order_id.as_object_mut().unwrap().remove("orderId");
1246 mount(
1247 &server,
1248 LIQUIDATION_PATH,
1249 page(vec![
1250 row(2, "SELL", "0.5", now - TimeDelta::hours(1)),
1251 row(2, "SELL", "0.5", now - TimeDelta::hours(1)),
1252 no_order_id,
1253 liquidation_row(8, "SELL", "0.5", now, true),
1255 row(1, "BUY", "0", now - TimeDelta::hours(2)),
1256 row(0, "BUY", "0.5", now - TimeDelta::days(3)),
1257 ]),
1258 )
1259 .await;
1260 let client = margin_client_at(&server, is_isolated);
1261
1262 let liquidations = client
1263 .fetch_forced_liquidations(now - TimeDelta::days(1), now)
1264 .await
1265 .unwrap();
1266
1267 let (unfilled, filled) = match (is_isolated, liquidations.as_slice()) {
1268 (false, [unfilled, filled]) => (unfilled, filled),
1269 (true, [unfilled, filled, marked]) => {
1270 assert_eq!(marked.order_id, OrderId::new("8"));
1271 (unfilled, filled)
1272 }
1273 _ => panic!("is_isolated = {is_isolated}: {liquidations:?}"),
1274 };
1275 assert_eq!(unfilled.order_id, OrderId::new("1"));
1276 assert_eq!(unfilled.side, Side::Buy);
1277 assert_eq!(unfilled.avg_price, None, "nothing filled");
1278 assert_eq!(
1279 *filled,
1280 MarginForcedLiquidation {
1281 order_id: OrderId::new("2"),
1282 instrument: InstrumentNameExchange::new("BTCUSDT"),
1283 side: Side::Sell,
1284 price: Some(dec!(59000)),
1285 quantity: Some(dec!(0.5)),
1286 filled_quantity: dec!(0.5),
1287 avg_price: Some(dec!(60000)),
1288 time_exchange: Utc
1289 .timestamp_millis_opt((now - TimeDelta::hours(1)).timestamp_millis())
1290 .unwrap(),
1291 }
1292 );
1293
1294 let queries = queries(&server).await;
1295 let [query] = queries.as_slice() else {
1296 panic!("one page: {queries:?}");
1297 };
1298 assert_eq!(query["size"], "100");
1299 assert_eq!(
1300 query.get("isolatedSymbol").map(String::as_str),
1301 is_isolated.then_some("BTCUSDT"),
1302 );
1303 }
1304 }
1305
1306 #[tokio::test]
1307 async fn a_full_page_of_forced_liquidations_reads_the_next() {
1308 let server = MockServer::start().await;
1309 let now = Utc::now();
1310 let time = now - TimeDelta::hours(1);
1311 let full = (0..100)
1312 .map(|id| liquidation_row(id, "SELL", "0.5", time, false))
1313 .collect();
1314 mount_page(&server, LIQUIDATION_PATH, "1", page(full)).await;
1315 mount_page(
1316 &server,
1317 LIQUIDATION_PATH,
1318 "2",
1319 page(vec![liquidation_row(100, "SELL", "0.5", time, false)]),
1320 )
1321 .await;
1322 let client = margin_client_at(&server, false);
1323
1324 let liquidations = client
1325 .fetch_forced_liquidations(now - TimeDelta::days(1), now)
1326 .await
1327 .unwrap();
1328
1329 assert_eq!(liquidations.len(), 101);
1330 assert_eq!(queries(&server).await.len(), 2);
1331 }
1332}