brokapi_finam/map/
m10.rs

1//! 10-minute candles built from 5-minute ones: Finam has no `M10` timeframe.
2//!
3//! The grid is the epoch: a 10-minute candle opens at a multiple of 600 s. Exchange time is UTC+3
4//! (a whole number of 10 minute steps), so these are the usual :00, :10, :20 ... boundaries.
5//! A candle is built from the M5 candles that fall in it: the first opens it, the last closes it,
6//! high and low are taken over the pair, volume is the sum. A half with no trades has no M5 candle at
7//! all, so a lone M5 candle is a legitimate 10-minute candle (and the last pair of a range is often
8//! incomplete): it is emitted as it is, not dropped.
9
10use brokapi_core::{Candle, Timeframe, UnixNanos};
11
12const SEC: i64 = 1_000_000_000;
13const M10_NANOS: i64 = 600 * SEC;
14
15/// Start of the 10-minute interval that contains `ts`.
16pub fn floor_m10(ts: UnixNanos) -> UnixNanos {
17    UnixNanos::from_nanos(ts.as_nanos().div_euclid(M10_NANOS) * M10_NANOS)
18}
19
20/// Merges the M5 candles of one 10-minute interval into an M10 candle. `parts` must be non-empty,
21/// belong to one instrument and one interval; the order does not matter.
22fn merge(parts: &[&Candle]) -> Candle {
23    let mut sorted: Vec<&Candle> = parts.to_vec();
24    sorted.sort_by_key(|c| c.open_ts);
25    let first = sorted[0];
26    let last = sorted[sorted.len() - 1];
27    let open_ts = floor_m10(first.open_ts);
28    Candle {
29        instrument: first.instrument.clone(),
30        timeframe: Timeframe::M10,
31        open: first.open,
32        high: sorted.iter().map(|c| c.high).max().unwrap_or(first.high),
33        low: sorted.iter().map(|c| c.low).min().unwrap_or(first.low),
34        close: last.close,
35        volume: sorted.iter().map(|c| c.volume).sum(),
36        open_ts,
37        close_ts: UnixNanos::from_nanos(open_ts.as_nanos() + M10_NANOS),
38        ts: last.ts,
39    }
40}
41
42/// History: M5 candles (in any order) -> M10 candles, oldest first.
43pub fn merge_m5(m5: &[Candle]) -> Vec<Candle> {
44    let mut groups: std::collections::BTreeMap<UnixNanos, Vec<&Candle>> = Default::default();
45    for c in m5 {
46        groups.entry(floor_m10(c.open_ts)).or_default().push(c);
47    }
48    groups.values().map(|g| merge(g)).collect()
49}
50
51/// Stream: follows M5 candle updates and keeps the M10 candle that is forming.
52///
53/// Every M5 update (a forming bar sent again, or a closed one) gives the current M10 candle. A
54/// bar of an older interval than the current one is a replay and is ignored.
55///
56/// # Examples
57///
58/// ```
59/// use std::time::Instant;
60/// use brokapi_core::{Candle, Price, Qty, Timeframe, Timestamps, UnixNanos};
61/// use brokapi_finam::map::m10::M10Builder;
62/// use brokapi_finam::util::instrument_from_symbol;
63///
64/// let m5 = |open_secs: i64, close: i64, volume: i64| Candle {
65///     instrument: instrument_from_symbol("SBER@MISX"),
66///     timeframe: Timeframe::M5,
67///     open: Price::from(100),
68///     high: Price::from(close.max(100)),
69///     low: Price::from(close.min(100)),
70///     close: Price::from(close),
71///     volume: Qty::from(volume),
72///     open_ts: UnixNanos::from_nanos(open_secs * 1_000_000_000),
73///     close_ts: UnixNanos::from_nanos((open_secs + 300) * 1_000_000_000),
74///     ts: Timestamps::new(UnixNanos::from_nanos(0), Instant::now()),
75/// };
76///
77/// let mut builder = M10Builder::new();
78/// // The first half of the interval 0..600 s.
79/// let part = builder.on_m5(&m5(0, 101, 10)).unwrap();
80/// assert_eq!(part.timeframe, Timeframe::M10);
81/// // The second half arrives: the same M10 candle, now with both halves.
82/// let whole = builder.on_m5(&m5(300, 103, 5)).unwrap();
83/// assert_eq!(whole.volume, Qty::from(15));
84/// assert_eq!(whole.close, Price::from(103));
85/// // A bar of an older interval is a replay and is ignored.
86/// let _next = builder.on_m5(&m5(600, 104, 1)).unwrap();
87/// assert!(builder.on_m5(&m5(0, 101, 10)).is_none());
88/// ```
89#[derive(Debug, Default)]
90pub struct M10Builder {
91    current: Option<UnixNanos>,
92    /// The latest version of each M5 candle of the current interval, by open time.
93    parts: Vec<Candle>,
94}
95
96impl M10Builder {
97    /// An empty builder.
98    pub fn new() -> Self {
99        Self::default()
100    }
101
102    /// Feeds one M5 update. Returns the current M10 candle, or `None` when the bar belongs to an interval
103    /// older than the current one (a replay after a reconnect).
104    ///
105    /// Call it for every M5 message, including the repeats of a forming bar: each one gives the
106    /// up-to-date M10 candle. Prices and volume of the candle are not final until the interval ends.
107    pub fn on_m5(&mut self, bar: &Candle) -> Option<Candle> {
108        let group = floor_m10(bar.open_ts);
109        match self.current {
110            Some(cur) if group < cur => return None,
111            Some(cur) if group == cur => {}
112            _ => {
113                self.current = Some(group);
114                self.parts.clear();
115            }
116        }
117        match self.parts.iter_mut().find(|p| p.open_ts == bar.open_ts) {
118            Some(p) => *p = bar.clone(),
119            None => self.parts.push(bar.clone()),
120        }
121        let refs: Vec<&Candle> = self.parts.iter().collect();
122        Some(merge(&refs))
123    }
124}
125
126#[cfg(test)]
127mod tests {
128    use super::*;
129    use brokapi_core::{Decimal, dec};
130    use brokapi_core::{InstrumentId, InstrumentKey, NativeId, Timestamps};
131    use std::time::Instant;
132
133    fn id() -> InstrumentId {
134        InstrumentId::new(
135            InstrumentKey::new("SBER", "MISX"),
136            NativeId::SymbolMic {
137                symbol: "SBER".into(),
138                mic: "MISX".into(),
139            },
140        )
141    }
142
143    /// An M5 candle opening `minute` minutes after a 10-minute boundary (the boundary is 1_800_000 s).
144    fn m5(minute: i64, o: &str, h: &str, l: &str, c: &str, v: &str) -> Candle {
145        let open_ts = UnixNanos::from_nanos(1_800_000 * SEC + minute * 60 * SEC);
146        Candle {
147            instrument: id(),
148            timeframe: Timeframe::M5,
149            open: o.parse().unwrap(),
150            high: h.parse().unwrap(),
151            low: l.parse().unwrap(),
152            close: c.parse().unwrap(),
153            volume: v.parse().unwrap(),
154            open_ts,
155            close_ts: UnixNanos::from_nanos(open_ts.as_nanos() + 300 * SEC),
156            ts: Timestamps::new(UnixNanos::ZERO, Instant::now()),
157        }
158    }
159
160    #[test]
161    fn floor_is_a_multiple_of_ten_minutes() {
162        let t = UnixNanos::from_nanos(1_800_000 * SEC + 7 * 60 * SEC + 13);
163        assert_eq!(floor_m10(t).as_nanos(), 1_800_000 * SEC);
164        assert_eq!(floor_m10(floor_m10(t)), floor_m10(t));
165    }
166
167    #[test]
168    fn a_pair_becomes_one_candle() {
169        let out = merge_m5(&[
170            m5(0, "10.0", "12.0", "9.0", "11.0", "100.0"),
171            m5(5, "11.0", "15.0", "10.5", "14.0", "50.0"),
172        ]);
173        assert_eq!(out.len(), 1);
174        let c = &out[0];
175        assert_eq!(c.timeframe, Timeframe::M10);
176        assert_eq!(
177            (
178                c.open.decimal(),
179                c.high.decimal(),
180                c.low.decimal(),
181                c.close.decimal()
182            ),
183            (dec!(10.0), dec!(15.0), dec!(9.0), dec!(14.0))
184        );
185        assert_eq!(c.volume.decimal(), dec!(150.0));
186        assert_eq!(c.open_ts.as_nanos(), 1_800_000 * SEC);
187        assert_eq!(c.close_ts.as_nanos(), 1_800_000 * SEC + 600 * SEC);
188    }
189
190    #[test]
191    fn pairs_are_cut_on_ten_minute_boundaries_not_by_position() {
192        // 00:05 is the second half of the first interval, 00:10 opens the next one.
193        let out = merge_m5(&[
194            m5(5, "11.0", "15.0", "10.5", "14.0", "50.0"),
195            m5(10, "20.0", "21.0", "19.0", "20.5", "7.0"),
196            m5(15, "20.5", "23.0", "20.0", "22.0", "3.0"),
197        ]);
198        assert_eq!(out.len(), 2);
199        // The first interval has only its second half (no trades in the first): it is kept as it is.
200        assert_eq!(
201            (
202                out[0].open.decimal(),
203                out[0].close.decimal(),
204                out[0].volume.decimal()
205            ),
206            (dec!(11.0), dec!(14.0), dec!(50.0))
207        );
208        assert_eq!(out[0].open_ts.as_nanos(), 1_800_000 * SEC);
209        assert_eq!(
210            (
211                out[1].open.decimal(),
212                out[1].high.decimal(),
213                out[1].close.decimal()
214            ),
215            (dec!(20.0), dec!(23.0), dec!(22.0))
216        );
217        assert_eq!(out[1].open_ts.as_nanos(), 1_800_000 * SEC + 600 * SEC);
218    }
219
220    #[test]
221    fn incomplete_last_pair_is_emitted() {
222        let out = merge_m5(&[
223            m5(0, "1.0", "2.0", "0.5", "1.5", "10.0"),
224            m5(5, "1.5", "3.0", "1.0", "2.5", "10.0"),
225            m5(10, "2.5", "2.6", "2.4", "2.5", "4.0"),
226        ]);
227        assert_eq!(out.len(), 2);
228        assert_eq!(out[1].volume.decimal(), dec!(4.0));
229        assert_eq!(out[1].close.decimal(), dec!(2.5));
230        assert_eq!(out[1].open_ts.as_nanos(), 1_800_000 * SEC + 600 * SEC);
231    }
232
233    #[test]
234    fn input_order_does_not_matter_and_empty_is_empty() {
235        let a = merge_m5(&[
236            m5(5, "11.0", "15.0", "10.5", "14.0", "50.0"),
237            m5(0, "10.0", "12.0", "9.0", "11.0", "100.0"),
238        ]);
239        assert_eq!(
240            (a[0].open.decimal(), a[0].close.decimal()),
241            (dec!(10.0), dec!(14.0))
242        );
243        assert!(merge_m5(&[]).is_empty());
244    }
245
246    #[test]
247    fn builder_follows_the_forming_candle() {
248        let mut b = M10Builder::new();
249        // First half forms, then updates.
250        let c = b
251            .on_m5(&m5(0, "10.0", "11.0", "9.5", "10.5", "5.0"))
252            .unwrap();
253        assert_eq!(
254            (c.open.decimal(), c.close.decimal(), c.volume.decimal()),
255            (dec!(10.0), dec!(10.5), dec!(5.0))
256        );
257        let c = b
258            .on_m5(&m5(0, "10.0", "12.0", "9.5", "11.0", "8.0"))
259            .unwrap();
260        assert_eq!(
261            (c.high.decimal(), c.close.decimal(), c.volume.decimal()),
262            (dec!(12.0), dec!(11.0), dec!(8.0)),
263            "an update replaces, it does not add"
264        );
265        // Second half joins.
266        let c = b
267            .on_m5(&m5(5, "11.0", "13.0", "10.0", "12.5", "4.0"))
268            .unwrap();
269        assert_eq!(
270            (
271                c.open.decimal(),
272                c.high.decimal(),
273                c.low.decimal(),
274                c.close.decimal()
275            ),
276            (dec!(10.0), dec!(13.0), dec!(9.5), dec!(12.5))
277        );
278        assert_eq!(c.volume.decimal(), dec!(12.0));
279        // The next interval starts fresh.
280        let c = b
281            .on_m5(&m5(10, "12.5", "12.6", "12.4", "12.5", "1.0"))
282            .unwrap();
283        assert_eq!(
284            (c.open.decimal(), c.volume.decimal()),
285            (dec!(12.5), dec!(1.0))
286        );
287        assert_eq!(c.open_ts.as_nanos(), 1_800_000 * SEC + 600 * SEC);
288        // A replay of the old interval is ignored.
289        assert!(
290            b.on_m5(&m5(5, "11.0", "13.0", "10.0", "12.5", "4.0"))
291                .is_none()
292        );
293    }
294}
brokapi_finam/map/m10.rs

brokapi_finam/map/
m10.rs
