quantwave_core/
streaming.rs

1//! Streaming readiness tracking for any `Next<T>` indicator.
2//!
3//! Live systems need to know when warmup is complete before acting on signals.
4//! Rather than modifying every indicator struct, wrap any `Next` implementor in
5//! `TrackedNext` to get `bars_consumed`, `warmup_bars`, and `is_ready`.
6//!
7//! Python DX: `quantwave.wrap_streaming()` mirrors this at the Python layer;
8//! Rust consumers use `track()` or `TrackedNext::new()` directly.
9
10use crate::traits::Next;
11
12/// Readiness state for streaming indicators.
13pub trait StreamingReadiness {
14    /// Bars processed since construction (or last reset).
15    fn bars_consumed(&self) -> usize;
16    /// Bars required before the indicator is considered ready.
17    fn warmup_bars(&self) -> usize;
18    /// True when `bars_consumed >= warmup_bars` (or heuristic for warmup_bars=0).
19    fn is_ready(&self) -> bool {
20        let warmup = self.warmup_bars();
21        if warmup == 0 {
22            self.bars_consumed() > 0
23        } else {
24            self.bars_consumed() >= warmup
25        }
26    }
27}
28
29/// Wraps any `Next<Input>` indicator with bar-count readiness tracking.
30#[derive(Debug, Clone)]
31pub struct TrackedNext<I> {
32    inner: I,
33    bars_consumed: usize,
34    warmup_bars: usize,
35}
36
37impl<I> TrackedNext<I> {
38    pub fn new(inner: I, warmup_bars: usize) -> Self {
39        Self {
40            inner,
41            bars_consumed: 0,
42            warmup_bars,
43        }
44    }
45
46    pub fn into_inner(self) -> I {
47        self.inner
48    }
49
50    pub fn inner(&self) -> &I {
51        &self.inner
52    }
53
54    pub fn inner_mut(&mut self) -> &mut I {
55        &mut self.inner
56    }
57
58    pub fn reset(&mut self) {
59        self.bars_consumed = 0;
60    }
61}
62
63impl<I, Input> Next<Input> for TrackedNext<I>
64where
65    I: Next<Input>,
66{
67    type Output = I::Output;
68
69    fn next(&mut self, input: Input) -> Self::Output {
70        self.bars_consumed += 1;
71        self.inner.next(input)
72    }
73}
74
75impl<I> StreamingReadiness for TrackedNext<I> {
76    fn bars_consumed(&self) -> usize {
77        self.bars_consumed
78    }
79
80    fn warmup_bars(&self) -> usize {
81        self.warmup_bars
82    }
83}
84
85/// Convenience: wrap an indicator with explicit warmup bar count.
86pub fn track<I>(inner: I, warmup_bars: usize) -> TrackedNext<I> {
87    TrackedNext::new(inner, warmup_bars)
88}
89
90/// Derive warmup from the largest numeric period-like parameter.
91pub fn warmup_from_params(params: &[(&str, usize)]) -> usize {
92    params.iter().map(|(_, v)| *v).max().unwrap_or(0)
93}
94
95#[cfg(test)]
96mod tests {
97    use super::*;
98    use crate::indicators::smoothing::EMA;
99
100    #[test]
101    fn test_tracked_next_readiness() {
102        let mut tracked = track(EMA::new(5), 5);
103        assert!(!tracked.is_ready());
104        assert_eq!(tracked.bars_consumed(), 0);
105
106        for i in 1..=4 {
107            tracked.next(i as f64);
108            assert_eq!(tracked.bars_consumed(), i);
109            assert!(!tracked.is_ready());
110        }
111        tracked.next(5.0);
112        assert_eq!(tracked.bars_consumed(), 5);
113        assert!(tracked.is_ready());
114    }
115
116    #[test]
117    fn test_zero_warmup_uses_heuristic() {
118        let mut tracked = track(EMA::new(3), 0);
119        assert!(!tracked.is_ready());
120        tracked.next(1.0);
121        assert!(tracked.is_ready());
122    }
123
124    #[test]
125    fn test_reset_clears_readiness() {
126        let mut tracked = track(EMA::new(3), 2);
127        tracked.next(1.0);
128        tracked.next(2.0);
129        assert!(tracked.is_ready());
130        tracked.reset();
131        assert!(!tracked.is_ready());
132        assert_eq!(tracked.bars_consumed(), 0);
133    }
134
135    #[test]
136    fn test_warmup_from_params() {
137        assert_eq!(warmup_from_params(&[("fast", 12), ("slow", 26)]), 26);
138        assert_eq!(warmup_from_params(&[]), 0);
139    }
140}
quantwave_core/streaming.rs

quantwave_core/
streaming.rs
