kestrel_chartkit/engine/
multi_series.rs

1use crate::model::{Bar, BarQuality};
2use std::collections::HashMap;
3
4/// Synchronized bar pair for multi-series computations (e.g. Asset vs Benchmark).
5#[derive(Debug, Clone, PartialEq)]
6pub struct SynchronizedBarPair {
7    pub timestamp: i64,
8    pub primary: Bar,
9    pub benchmark: Bar,
10    /// Original timestamp of the benchmark sample, retained when forward-filled.
11    pub benchmark_source_timestamp: i64,
12}
13
14#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
15pub enum MissingSeriesPolicy {
16    Skip,
17    #[default]
18    ForwardFill,
19}
20
21/// Synchronizer for aligning multiple price series with different timestamps or missing bars.
22#[derive(Debug, Clone, Default)]
23pub struct MultiSeriesSync {
24    last_benchmark: Option<Bar>,
25    missing_policy: MissingSeriesPolicy,
26}
27
28impl MultiSeriesSync {
29    pub fn new() -> Self {
30        Self {
31            last_benchmark: None,
32            missing_policy: MissingSeriesPolicy::ForwardFill,
33        }
34    }
35
36    pub fn with_policy(missing_policy: MissingSeriesPolicy) -> Self {
37        Self {
38            last_benchmark: None,
39            missing_policy,
40        }
41    }
42
43    pub fn reset(&mut self) {
44        self.last_benchmark = None;
45    }
46
47    /// Aligns incoming primary asset bars with benchmark bars.
48    /// A missing benchmark is either skipped or forward-filled according to the configured policy.
49    pub fn align_step(
50        &mut self,
51        primary: &Bar,
52        benchmark: Option<&Bar>,
53    ) -> Option<SynchronizedBarPair> {
54        let bench_bar = match benchmark {
55            Some(b) => {
56                self.last_benchmark = Some(b.clone());
57                b.clone()
58            }
59            None if self.missing_policy == MissingSeriesPolicy::ForwardFill => {
60                self.last_benchmark.clone()?
61            }
62            None => return None,
63        };
64
65        let benchmark_source_timestamp = bench_bar.timestamp;
66
67        Some(SynchronizedBarPair {
68            timestamp: primary.timestamp,
69            primary: primary.clone(),
70            benchmark: bench_bar,
71            benchmark_source_timestamp,
72        })
73    }
74
75    /// Batch aligns two full series of bars by timestamp.
76    pub fn align_series(
77        primary_series: &[Bar],
78        benchmark_series: &[Bar],
79    ) -> Vec<SynchronizedBarPair> {
80        let mut bench_map: HashMap<i64, Bar> = HashMap::new();
81        for b in benchmark_series {
82            bench_map.insert(b.timestamp, b.clone());
83        }
84
85        let mut sync = MultiSeriesSync::new();
86        let mut result = Vec::with_capacity(primary_series.len());
87
88        for p_bar in primary_series {
89            let b_bar = bench_map.get(&p_bar.timestamp);
90            if let Some(pair) = sync.align_step(p_bar, b_bar) {
91                result.push(pair);
92            }
93        }
94
95        result
96    }
97}
98
99/// One series' bar within an [`AlignedStep`], tagged with explicit [`BarQuality`] (observed vs.
100/// forward-filled, and whether a gap preceded it).
101#[derive(Debug, Clone, PartialEq)]
102pub struct AlignedSeriesBar {
103    pub bar: Bar,
104    pub quality: BarQuality,
105}
106
107/// One synchronized alignment step across an arbitrary number of named series.
108#[derive(Debug, Clone, PartialEq, Default)]
109pub struct AlignedStep {
110    pub timestamp: i64,
111    /// Present entries are keyed by series ID. A series absent from this map means it was
112    /// missing this step and either had no prior bar to forward-fill from, or its last bar aged
113    /// out under [`MultiSeriesAligner::with_max_forward_fill_age`] — i.e. absence itself is the
114    /// explicit gap signal.
115    pub series: HashMap<String, AlignedSeriesBar>,
116}
117
118/// Generalizes [`MultiSeriesSync`] from exactly two series (primary/benchmark) to an arbitrary,
119/// named set: N-way synchronization, an optional maximum forward-fill staleness age, and explicit
120/// per-series [`BarQuality`] (observed vs. forward-filled) instead of a single implicit benchmark
121/// slot.
122#[derive(Debug, Clone)]
123pub struct MultiSeriesAligner {
124    series_ids: Vec<String>,
125    last_seen: HashMap<String, (Bar, i64)>,
126    missing_policy: MissingSeriesPolicy,
127    max_forward_fill_age_seconds: Option<i64>,
128}
129
130impl MultiSeriesAligner {
131    pub fn new(series_ids: impl IntoIterator<Item = impl Into<String>>) -> Self {
132        Self {
133            series_ids: series_ids.into_iter().map(Into::into).collect(),
134            last_seen: HashMap::new(),
135            missing_policy: MissingSeriesPolicy::ForwardFill,
136            max_forward_fill_age_seconds: None,
137        }
138    }
139
140    pub fn with_policy(mut self, policy: MissingSeriesPolicy) -> Self {
141        self.missing_policy = policy;
142        self
143    }
144
145    /// Caps how old (in seconds) a forward-filled bar may be before it is dropped instead of
146    /// reused, so a series that has gone permanently silent does not keep forward-filling
147    /// forever.
148    pub fn with_max_forward_fill_age(mut self, seconds: i64) -> Self {
149        self.max_forward_fill_age_seconds = Some(seconds);
150        self
151    }
152
153    pub fn reset(&mut self) {
154        self.last_seen.clear();
155    }
156
157    /// Aligns one step at `timestamp`. `incoming` holds bars for whichever configured series
158    /// delivered one for exactly this timestamp; missing series are forward-filled (subject to
159    /// [`MultiSeriesAligner::with_max_forward_fill_age`]) or skipped per the configured
160    /// [`MissingSeriesPolicy`].
161    pub fn align_step(&mut self, timestamp: i64, incoming: &HashMap<String, Bar>) -> AlignedStep {
162        let mut series = HashMap::with_capacity(self.series_ids.len());
163
164        for id in &self.series_ids {
165            if let Some(bar) = incoming.get(id) {
166                self.last_seen.insert(id.clone(), (bar.clone(), timestamp));
167                series.insert(
168                    id.clone(),
169                    AlignedSeriesBar {
170                        bar: bar.clone(),
171                        quality: BarQuality::observed(),
172                    },
173                );
174                continue;
175            }
176
177            if self.missing_policy != MissingSeriesPolicy::ForwardFill {
178                continue;
179            }
180
181            let Some((last_bar, last_ts)) = self.last_seen.get(id) else {
182                continue;
183            };
184            let age = timestamp - last_ts;
185            let within_age = self
186                .max_forward_fill_age_seconds
187                .map(|max| age <= max)
188                .unwrap_or(true);
189            if !within_age {
190                continue;
191            }
192
193            let mut filled = last_bar.clone();
194            filled.timestamp = timestamp;
195            series.insert(
196                id.clone(),
197                AlignedSeriesBar {
198                    bar: filled,
199                    quality: BarQuality {
200                        volume_available: false,
201                        is_synthetic: false,
202                        is_forward_filled: true,
203                        has_gap: age > 0,
204                    },
205                },
206            );
207        }
208
209        AlignedStep { timestamp, series }
210    }
211}
212
213#[cfg(test)]
214mod tests {
215    use super::*;
216
217    #[test]
218    fn test_multi_series_sync_forward_fill() {
219        let mut sync = MultiSeriesSync::new();
220        let b1 = Bar::new(1000, 10.0, 11.0, 9.0, 10.5, 500.0);
221        let p1 = Bar::new(1000, 100.0, 105.0, 95.0, 102.0, 1000.0);
222        let p2 = Bar::new(2000, 102.0, 106.0, 101.0, 105.0, 1000.0);
223
224        let pair1 = sync.align_step(&p1, Some(&b1)).unwrap();
225        assert_eq!(pair1.benchmark.timestamp, 1000);
226
227        // p2 arrives, benchmark is missing for 2000 -> forward fills b1
228        let pair2 = sync.align_step(&p2, None).unwrap();
229        assert_eq!(pair2.benchmark.timestamp, 1000);
230        assert_eq!(pair2.benchmark_source_timestamp, 1000);
231        assert_eq!(pair2.benchmark.close, 10.5);
232    }
233
234    #[test]
235    fn never_substitutes_the_primary_for_a_missing_benchmark() {
236        let primary = Bar::new(1_000, 100.0, 105.0, 95.0, 102.0, 1_000.0);
237        assert!(MultiSeriesSync::new().align_step(&primary, None).is_none());
238        assert!(MultiSeriesSync::with_policy(MissingSeriesPolicy::Skip)
239            .align_step(&primary, None)
240            .is_none());
241    }
242
243    #[test]
244    fn test_multi_series_aligner_aligns_more_than_two_series() {
245        let mut aligner =
246            MultiSeriesAligner::new(["a", "b", "c"]).with_policy(MissingSeriesPolicy::ForwardFill);
247
248        let mut incoming = HashMap::new();
249        incoming.insert("a".to_string(), Bar::new(0, 1.0, 2.0, 0.5, 1.5, 10.0));
250        incoming.insert("b".to_string(), Bar::new(0, 2.0, 3.0, 1.5, 2.5, 10.0));
251        incoming.insert("c".to_string(), Bar::new(0, 3.0, 4.0, 2.5, 3.5, 10.0));
252        let step0 = aligner.align_step(0, &incoming);
253        assert_eq!(step0.series.len(), 3);
254        for bar in step0.series.values() {
255            assert_eq!(bar.quality, BarQuality::observed());
256        }
257
258        // Only "a" delivers at t=60; "b" and "c" forward-fill with explicit quality flags.
259        let mut incoming = HashMap::new();
260        incoming.insert("a".to_string(), Bar::new(60, 1.1, 2.1, 0.6, 1.6, 10.0));
261        let step1 = aligner.align_step(60, &incoming);
262        assert_eq!(step1.series.len(), 3);
263        assert_eq!(step1.series["a"].quality, BarQuality::observed());
264        assert!(step1.series["b"].quality.is_forward_filled);
265        assert!(step1.series["b"].quality.has_gap);
266        assert_eq!(step1.series["b"].bar.close, 2.5);
267        assert_eq!(step1.series["b"].bar.timestamp, 60);
268    }
269
270    #[test]
271    fn test_multi_series_aligner_drops_stale_forward_fills() {
272        let mut aligner = MultiSeriesAligner::new(["a", "b"])
273            .with_policy(MissingSeriesPolicy::ForwardFill)
274            .with_max_forward_fill_age(90);
275
276        let mut incoming = HashMap::new();
277        incoming.insert("a".to_string(), Bar::new(0, 1.0, 2.0, 0.5, 1.5, 10.0));
278        incoming.insert("b".to_string(), Bar::new(0, 2.0, 3.0, 1.5, 2.5, 10.0));
279        aligner.align_step(0, &incoming);
280
281        // "b" stays silent past the 90s max forward-fill age -> dropped, not stale-filled.
282        let step = aligner.align_step(
283            200,
284            &HashMap::from([("a".to_string(), Bar::new(200, 1.2, 2.2, 0.7, 1.7, 10.0))]),
285        );
286        assert!(step.series.contains_key("a"));
287        assert!(
288            !step.series.contains_key("b"),
289            "stale forward-fill beyond max age must be omitted, not silently reused"
290        );
291    }
292
293    #[test]
294    fn test_multi_series_aligner_skip_policy_never_forward_fills() {
295        let mut aligner =
296            MultiSeriesAligner::new(["a", "b"]).with_policy(MissingSeriesPolicy::Skip);
297        aligner.align_step(
298            0,
299            &HashMap::from([
300                ("a".to_string(), Bar::new(0, 1.0, 2.0, 0.5, 1.5, 10.0)),
301                ("b".to_string(), Bar::new(0, 2.0, 3.0, 1.5, 2.5, 10.0)),
302            ]),
303        );
304
305        let step = aligner.align_step(
306            60,
307            &HashMap::from([("a".to_string(), Bar::new(60, 1.1, 2.1, 0.6, 1.6, 10.0))]),
308        );
309        assert!(step.series.contains_key("a"));
310        assert!(!step.series.contains_key("b"));
311    }
312}
kestrel_chartkit/engine/multi_series.rs

kestrel_chartkit/engine/
multi_series.rs
