brokapi_runtime/
history.rs

1//! Reading a long candle history in windows that the broker accepts, within the rate limit.
2//!
3//! One `candles` request covers a limited span (BCS: 1440 bars, T-Bank: from a day for minute bars to
4//! ten years for monthly ones). A year of hourly bars is therefore several requests, and each one
5//! spends quota. [`CandlePages`] cuts the range into windows, reads them one after another, waits out
6//! [`Error::RateLimited`], repeats a read that failed on the network, joins the edges of neighbouring
7//! windows without duplicates and always knows where to start again ([`resume_from`](CandlePages::resume_from)).
8//!
9//! Only reads are repeated here. A candle read changes nothing at the broker, so a repeat is safe.
10//! Commands are never repeated anywhere in the library.
11//!
12//! The helper is written against [`MarketDataSource`], so it works with every connector. The span of
13//! a window comes from [`MarketDataSource::candles_max_span`] unless [`HistoryOpts::window`] says
14//! otherwise.
15//!
16//! # Examples
17//!
18//! ```
19//! # use brokapi_core::*;
20//! # use brokapi_runtime::history::{CandlePages, HistoryOpts};
21//! # async fn demo(md: &impl MarketDataSource, sber: &InstrumentId) -> Result<(), Box<dyn std::error::Error>> {
22//! let year = TimeRange::new(
23//!     UnixNanos::from_nanos(1_767_225_600_000_000_000), // 2026-01-01
24//!     UnixNanos::from_nanos(1_798_761_600_000_000_000), // 2027-01-01
25//! );
26//! let mut pages = CandlePages::new(md, sber, Timeframe::H1, year, HistoryOpts::default());
27//! while let Some(page) = pages.next_page().await? {
28//!     // Write the page to disk here. After a crash start again from `pages.resume_from()`.
29//!     println!("{} bars, next window starts at {:?}", page.len(), pages.resume_from());
30//! }
31//! # Ok(()) }
32//! ```
33
34use std::fmt;
35use std::time::Duration;
36
37use brokapi_core::{
38    Candle, Error, InstrumentId, MarketDataSource, TimeRange, Timeframe, UnixNanos,
39};
40
41/// How [`CandlePages`] cuts the range and how patient it is.
42///
43/// # Examples
44///
45/// ```
46/// use std::time::Duration;
47/// use brokapi_runtime::history::HistoryOpts;
48///
49/// let opts = HistoryOpts { max_rate_waits: 20, ..HistoryOpts::default() };
50/// assert_eq!(opts.window, None);
51/// assert_eq!(opts.max_rate_wait, Duration::from_secs(60));
52/// ```
53#[derive(Debug, Clone, Copy, PartialEq, Eq)]
54pub struct HistoryOpts {
55    /// The span of one request. `None`: take [`MarketDataSource::candles_max_span`], and when the
56    /// connector does not know it, read the whole range in one request.
57    pub window: Option<Duration>,
58    /// How many times one window waits out [`Error::RateLimited`] before the error is returned.
59    /// Default 10.
60    pub max_rate_waits: u32,
61    /// The longest single pause on [`Error::RateLimited`]. The pause is the `retry_after` of the
62    /// error, cut to this value. Default 60 seconds.
63    pub max_rate_wait: Duration,
64    /// How many times one window is read again after [`Error::Transport`]. Default 2.
65    pub transport_retries: u32,
66    /// The pause before the first repeat after [`Error::Transport`]. It doubles for each next
67    /// repeat. Default 1 second.
68    pub transport_backoff: Duration,
69}
70
71impl Default for HistoryOpts {
72    fn default() -> Self {
73        Self {
74            window: None,
75            max_rate_waits: 10,
76            max_rate_wait: Duration::from_secs(60),
77            transport_retries: 2,
78            transport_backoff: Duration::from_secs(1),
79        }
80    }
81}
82
83/// A history read stopped. Carries the error and the place to start from again.
84///
85/// # Examples
86///
87/// ```
88/// use brokapi_core::{Error, UnixNanos};
89/// use brokapi_runtime::history::HistoryError;
90///
91/// let e = HistoryError { error: Error::Unknown, resume_from: UnixNanos::ZERO };
92/// assert!(e.to_string().contains("history read stopped"));
93/// ```
94#[derive(Debug, Clone)]
95pub struct HistoryError {
96    /// What stopped the read. [`Error::class`] says whether it makes sense to try again later.
97    pub error: Error,
98    /// The start of the window that failed. Everything before it was delivered. Start a new
99    /// [`CandlePages`] with a range from here and nothing is read twice.
100    pub resume_from: UnixNanos,
101}
102
103impl fmt::Display for HistoryError {
104    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
105        write!(
106            f,
107            "history read stopped at {} ns: {}",
108            self.resume_from.as_nanos(),
109            self.error
110        )
111    }
112}
113
114impl std::error::Error for HistoryError {
115    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
116        Some(&self.error)
117    }
118}
119
120/// A reader of candle history, one window per [`next_page`](Self::next_page) call.
121///
122/// Pages come in time order. Inside the whole read every bar is delivered once: a bar that two
123/// neighbouring windows both return (brokers differ on whether the end of a range is included) is
124/// kept in the first. A page can be empty (a weekend, a halt): that is not the end, the end is `None`.
125///
126/// The type borrows the connector and the instrument and holds no network state of its own, so a
127/// crash loses only the progress you did not save. Save [`resume_from`](Self::resume_from) after
128/// each page you wrote.
129///
130/// # Examples
131///
132/// See the module [`history`](crate::history).
133pub struct CandlePages<'a, S: MarketDataSource + ?Sized> {
134    src: &'a S,
135    id: &'a InstrumentId,
136    tf: Timeframe,
137    next_from: UnixNanos,
138    to: UnixNanos,
139    window: Option<Duration>,
140    opts: HistoryOpts,
141    last_open: Option<UnixNanos>,
142    requests: u32,
143    rate_waits: u32,
144}
145
146impl<'a, S: MarketDataSource + ?Sized> CandlePages<'a, S> {
147    /// Prepares a read of `tf` bars of `id` over `range`. Nothing is sent yet.
148    pub fn new(
149        src: &'a S,
150        id: &'a InstrumentId,
151        tf: Timeframe,
152        range: TimeRange,
153        opts: HistoryOpts,
154    ) -> Self {
155        let window = opts.window.or_else(|| src.candles_max_span(tf));
156        Self {
157            src,
158            id,
159            tf,
160            next_from: range.from,
161            to: range.to,
162            window,
163            opts,
164            last_open: None,
165            requests: 0,
166            rate_waits: 0,
167        }
168    }
169
170    /// The start of the next window: where to begin again after a stop. Equals the end of the range
171    /// when everything is read.
172    pub fn resume_from(&self) -> UnixNanos {
173        self.next_from
174    }
175
176    /// How many `candles` calls were made, repeats included. A call that the local rate limiter refused is counted
177    /// too, although it never reached the broker: count the requests that went out on the broker's side if you
178    /// need them.
179    pub fn requests(&self) -> u32 {
180        self.requests
181    }
182
183    /// How many times a rate limit was waited out inside this reader (a wait of your own loop is not counted).
184    pub fn rate_waits(&self) -> u32 {
185        self.rate_waits
186    }
187
188    /// Reads the next window. `Ok(None)` when the range is finished.
189    ///
190    /// # Errors
191    ///
192    /// [`HistoryError`] with the error that stopped the read and `resume_from`:
193    /// - [`Error::RateLimited`] after [`HistoryOpts::max_rate_waits`] pauses: the quota is spent by
194    ///   others too. Wait longer and start again from `resume_from`.
195    /// - [`Error::Transport`] after [`HistoryOpts::transport_retries`] repeats: the broker is
196    ///   unreachable. Start again later.
197    /// - [`Error::Rejected`], [`Error::Config`], [`Error::Unsupported`]: the same read fails again.
198    ///   Check the instrument, the timeframe and the window. Do not loop.
199    pub async fn next_page(&mut self) -> Result<Option<Vec<Candle>>, HistoryError> {
200        if self.next_from >= self.to {
201            return Ok(None);
202        }
203        let from = self.next_from;
204        let end = match self.window {
205            Some(w) => {
206                let step = i64::try_from(w.as_nanos()).unwrap_or(i64::MAX).max(1);
207                UnixNanos::from_nanos(from.as_nanos().saturating_add(step).min(self.to.as_nanos()))
208            }
209            None => self.to,
210        };
211        let mut bars = self.read_window(TimeRange::new(from, end)).await?;
212        bars.sort_by_key(|c| c.open_ts);
213        if let Some(last) = self.last_open {
214            bars.retain(|c| c.open_ts > last);
215        }
216        bars.dedup_by_key(|c| c.open_ts);
217        if let Some(c) = bars.last() {
218            self.last_open = Some(c.open_ts);
219        }
220        self.next_from = end;
221        Ok(Some(bars))
222    }
223
224    /// One window with its patience: pauses on a rate limit, repeats on a network error.
225    async fn read_window(&mut self, range: TimeRange) -> Result<Vec<Candle>, HistoryError> {
226        let mut waits = 0u32;
227        let mut retries = 0u32;
228        loop {
229            self.requests += 1;
230            match self.src.candles(self.id, self.tf, range).await {
231                Ok(bars) => return Ok(bars),
232                Err(Error::RateLimited { retry_after }) if waits < self.opts.max_rate_waits => {
233                    waits += 1;
234                    self.rate_waits += 1;
235                    tokio::time::sleep(retry_after.min(self.opts.max_rate_wait)).await;
236                }
237                Err(Error::Transport(_)) if retries < self.opts.transport_retries => {
238                    let pause = self.opts.transport_backoff * 2u32.saturating_pow(retries);
239                    retries += 1;
240                    tokio::time::sleep(pause).await;
241                }
242                Err(error) => {
243                    return Err(HistoryError {
244                        error,
245                        resume_from: range.from,
246                    });
247                }
248            }
249        }
250    }
251}
252
253/// Reads the whole range into memory: [`CandlePages`] until the end.
254///
255/// Suits a few thousand bars. For a year of minute bars of many instruments write page by page
256/// instead, so memory does not grow.
257///
258/// # Errors
259///
260/// As [`CandlePages::next_page`]. Bars read before the error are lost with it: use
261/// [`CandlePages`] when you need them.
262///
263/// # Examples
264///
265/// ```
266/// # use brokapi_core::*;
267/// # use brokapi_runtime::history::{load_candles, HistoryOpts};
268/// # async fn demo(md: &impl MarketDataSource, sber: &InstrumentId, range: TimeRange) -> Result<(), Box<dyn std::error::Error>> {
269/// let bars = load_candles(md, sber, Timeframe::D1, range, HistoryOpts::default()).await?;
270/// assert!(bars.windows(2).all(|w| w[0].open_ts < w[1].open_ts));
271/// # Ok(()) }
272/// ```
273pub async fn load_candles<S: MarketDataSource + ?Sized>(
274    src: &S,
275    id: &InstrumentId,
276    tf: Timeframe,
277    range: TimeRange,
278    opts: HistoryOpts,
279) -> Result<Vec<Candle>, HistoryError> {
280    let mut pages = CandlePages::new(src, id, tf, range, opts);
281    let mut all = Vec::new();
282    while let Some(mut page) = pages.next_page().await? {
283        all.append(&mut page);
284    }
285    Ok(all)
286}
287
288#[cfg(test)]
289mod tests {
290    use std::sync::Mutex;
291    use std::sync::atomic::{AtomicU32, Ordering};
292
293    use async_trait::async_trait;
294    use brokapi_core::{
295        InstrumentKey, MarketTrade, NativeId, Price, Qty, SubscribeOpts, SubscriptionId, Timestamps,
296    };
297
298    use super::*;
299
300    /// A source with one bar per hour, a rate limit on the first calls and an optional window cap.
301    struct Fake {
302        calls: Mutex<Vec<TimeRange>>,
303        rate_limited_first: AtomicU32,
304        transport_first: AtomicU32,
305        max_span: Option<Duration>,
306        /// A bar at the very end of a range is returned too, like a broker that includes the end.
307        inclusive_end: bool,
308    }
309
310    impl Fake {
311        fn new(max_span: Option<Duration>) -> Self {
312            Self {
313                calls: Mutex::new(Vec::new()),
314                rate_limited_first: AtomicU32::new(0),
315                transport_first: AtomicU32::new(0),
316                max_span,
317                inclusive_end: false,
318            }
319        }
320    }
321
322    fn id() -> InstrumentId {
323        InstrumentId::new(
324            InstrumentKey::new("SBER", "TQBR"),
325            NativeId::TickerClass {
326                ticker: "SBER".into(),
327                class_code: "TQBR".into(),
328            },
329        )
330    }
331
332    const HOUR: i64 = 3_600_000_000_000;
333
334    fn bar(open: i64) -> Candle {
335        Candle {
336            instrument: id(),
337            timeframe: Timeframe::H1,
338            open: Price::from(1),
339            high: Price::from(2),
340            low: Price::from(1),
341            close: Price::from(2),
342            volume: Qty::from(1),
343            open_ts: UnixNanos::from_nanos(open),
344            close_ts: UnixNanos::from_nanos(open + HOUR),
345            ts: Timestamps::new(UnixNanos::ZERO, std::time::Instant::now()),
346        }
347    }
348
349    #[async_trait]
350    impl MarketDataSource for Fake {
351        async fn last_trades(&self, _: &InstrumentId, _: u32) -> Result<Vec<MarketTrade>, Error> {
352            unimplemented!()
353        }
354        async fn subscribe_market_trades(
355            &self,
356            _: &InstrumentId,
357            _: SubscribeOpts,
358        ) -> Result<SubscriptionId, Error> {
359            unimplemented!()
360        }
361        async fn subscribe_quotes(
362            &self,
363            _: &InstrumentId,
364            _: SubscribeOpts,
365        ) -> Result<SubscriptionId, Error> {
366            unimplemented!()
367        }
368        async fn candles(
369            &self,
370            _: &InstrumentId,
371            _: Timeframe,
372            range: TimeRange,
373        ) -> Result<Vec<Candle>, Error> {
374            if let Some(max) = self.max_span {
375                let span = (range.to.as_nanos() - range.from.as_nanos()) as u128;
376                if span > max.as_nanos() {
377                    return Err(Error::Rejected {
378                        code: "span".into(),
379                        message: "window too long".into(),
380                    });
381                }
382            }
383            if self
384                .rate_limited_first
385                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1))
386                .is_ok()
387            {
388                return Err(Error::RateLimited {
389                    retry_after: Duration::from_millis(5),
390                });
391            }
392            if self
393                .transport_first
394                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1))
395                .is_ok()
396            {
397                return Err(Error::Transport(brokapi_core::TransportError::Message(
398                    "reset".into(),
399                )));
400            }
401            self.calls.lock().unwrap().push(range);
402            let mut open = (range.from.as_nanos() + HOUR - 1) / HOUR * HOUR;
403            let mut out = Vec::new();
404            while open < range.to.as_nanos() || (self.inclusive_end && open == range.to.as_nanos())
405            {
406                out.push(bar(open));
407                open += HOUR;
408            }
409            Ok(out)
410        }
411        async fn subscribe_candles(
412            &self,
413            _: &InstrumentId,
414            _: Timeframe,
415            _: SubscribeOpts,
416        ) -> Result<SubscriptionId, Error> {
417            unimplemented!()
418        }
419        fn candles_max_span(&self, _: Timeframe) -> Option<Duration> {
420            self.max_span
421        }
422    }
423
424    fn hours(n: i64) -> TimeRange {
425        TimeRange::new(UnixNanos::ZERO, UnixNanos::from_nanos(n * HOUR))
426    }
427
428    fn fast() -> HistoryOpts {
429        HistoryOpts {
430            transport_backoff: Duration::from_millis(1),
431            ..HistoryOpts::default()
432        }
433    }
434
435    #[tokio::test]
436    async fn cuts_the_range_into_windows_the_source_accepts() {
437        let fake = Fake::new(Some(Duration::from_secs(10 * 3600)));
438        let bars = load_candles(&fake, &id(), Timeframe::H1, hours(35), fast())
439            .await
440            .unwrap();
441        assert_eq!(bars.len(), 35);
442        assert!(bars.windows(2).all(|w| w[0].open_ts < w[1].open_ts));
443        // 35 hours in windows of 10: 10 + 10 + 10 + 5
444        assert_eq!(fake.calls.lock().unwrap().len(), 4);
445    }
446
447    #[tokio::test]
448    async fn without_a_known_span_the_range_goes_in_one_request() {
449        let fake = Fake::new(None);
450        let bars = load_candles(&fake, &id(), Timeframe::H1, hours(35), fast())
451            .await
452            .unwrap();
453        assert_eq!(bars.len(), 35);
454        assert_eq!(fake.calls.lock().unwrap().len(), 1);
455    }
456
457    #[tokio::test]
458    async fn the_window_option_overrides_the_source() {
459        let fake = Fake::new(Some(Duration::from_secs(100 * 3600)));
460        let opts = HistoryOpts {
461            window: Some(Duration::from_secs(12 * 3600)),
462            ..fast()
463        };
464        load_candles(&fake, &id(), Timeframe::H1, hours(36), opts)
465            .await
466            .unwrap();
467        assert_eq!(fake.calls.lock().unwrap().len(), 3);
468    }
469
470    #[tokio::test]
471    async fn a_bar_on_the_edge_of_two_windows_is_delivered_once() {
472        let mut fake = Fake::new(Some(Duration::from_secs(10 * 3600)));
473        fake.inclusive_end = true;
474        let bars = load_candles(&fake, &id(), Timeframe::H1, hours(30), fast())
475            .await
476            .unwrap();
477        let mut opens: Vec<i64> = bars.iter().map(|b| b.open_ts.as_nanos()).collect();
478        let before = opens.len();
479        opens.dedup();
480        assert_eq!(before, opens.len(), "no bar twice");
481        // 0..=30 hours: the last window includes its end
482        assert_eq!(opens.len(), 31);
483    }
484
485    #[tokio::test]
486    async fn a_rate_limit_is_waited_out_and_counted() {
487        let fake = Fake::new(Some(Duration::from_secs(10 * 3600)));
488        fake.rate_limited_first.store(3, Ordering::SeqCst);
489        let inst = id();
490        let mut pages = CandlePages::new(&fake, &inst, Timeframe::H1, hours(20), fast());
491        while pages.next_page().await.unwrap().is_some() {}
492        assert_eq!(pages.rate_waits(), 3);
493        assert_eq!(pages.requests(), 5, "2 windows and 3 refused attempts");
494        assert_eq!(pages.resume_from(), hours(20).to);
495    }
496
497    #[tokio::test]
498    async fn too_many_rate_limits_stop_the_read_at_the_failed_window() {
499        let fake = Fake::new(Some(Duration::from_secs(10 * 3600)));
500        fake.rate_limited_first.store(100, Ordering::SeqCst);
501        let opts = HistoryOpts {
502            max_rate_waits: 2,
503            ..fast()
504        };
505        let err = load_candles(&fake, &id(), Timeframe::H1, hours(20), opts)
506            .await
507            .unwrap_err();
508        assert!(matches!(err.error, Error::RateLimited { .. }));
509        assert_eq!(err.resume_from, UnixNanos::ZERO);
510    }
511
512    #[tokio::test]
513    async fn a_network_error_is_repeated_a_few_times_then_reported() {
514        let fake = Fake::new(None);
515        fake.transport_first.store(2, Ordering::SeqCst);
516        let bars = load_candles(&fake, &id(), Timeframe::H1, hours(5), fast())
517            .await
518            .unwrap();
519        assert_eq!(bars.len(), 5);
520
521        let fake = Fake::new(None);
522        fake.transport_first.store(3, Ordering::SeqCst);
523        let err = load_candles(&fake, &id(), Timeframe::H1, hours(5), fast())
524            .await
525            .unwrap_err();
526        assert!(matches!(err.error, Error::Transport(_)));
527    }
528
529    #[tokio::test]
530    async fn a_refusal_is_not_repeated_and_tells_where_to_resume() {
531        // The window option is larger than the source accepts, so every request is refused.
532        let fake = Fake::new(Some(Duration::from_secs(5 * 3600)));
533        let opts = HistoryOpts {
534            window: Some(Duration::from_secs(10 * 3600)),
535            ..fast()
536        };
537        let inst = id();
538        let mut pages = CandlePages::new(&fake, &inst, Timeframe::H1, hours(20), opts);
539        let err = pages.next_page().await.unwrap_err();
540        assert!(matches!(err.error, Error::Rejected { .. }));
541        assert_eq!(pages.requests(), 1, "a refusal is final");
542        assert_eq!(err.resume_from, UnixNanos::ZERO);
543    }
544
545    #[tokio::test]
546    async fn an_empty_range_is_finished_at_once() {
547        let fake = Fake::new(None);
548        let inst = id();
549        let mut pages = CandlePages::new(&fake, &inst, Timeframe::H1, hours(0), fast());
550        assert!(pages.next_page().await.unwrap().is_none());
551        assert_eq!(pages.requests(), 0);
552    }
553}
brokapi_runtime/history.rs

brokapi_runtime/
history.rs
