hyper_agent_core/
candle_buffer.rs

1use hyper_ta::Candle;
2use std::collections::VecDeque;
3
4const DEFAULT_CAPACITY: usize = 200;
5
6pub struct CandleBuffer {
7    candles: VecDeque<Candle>,
8    capacity: usize,
9    symbol: String,
10}
11
12impl CandleBuffer {
13    pub fn new(symbol: String) -> Self {
14        Self::with_capacity(symbol, DEFAULT_CAPACITY)
15    }
16
17    pub fn with_capacity(symbol: String, capacity: usize) -> Self {
18        Self {
19            candles: VecDeque::with_capacity(capacity),
20            capacity,
21            symbol,
22        }
23    }
24
25    pub fn symbol(&self) -> &str {
26        &self.symbol
27    }
28
29    pub fn len(&self) -> usize {
30        self.candles.len()
31    }
32
33    pub fn is_empty(&self) -> bool {
34        self.candles.is_empty()
35    }
36
37    /// Push a new candle. If at capacity, the oldest candle is evicted.
38    pub fn push(&mut self, candle: Candle) {
39        if self.candles.len() >= self.capacity {
40            self.candles.pop_front();
41        }
42        self.candles.push_back(candle);
43    }
44
45    /// Get the most recent candle.
46    pub fn last(&self) -> Option<&Candle> {
47        self.candles.back()
48    }
49
50    /// Return all candles as a slice (for indicator calculation).
51    pub fn as_slice(&self) -> Vec<Candle> {
52        self.candles.iter().cloned().collect()
53    }
54
55    /// Backfill candles from REST API response.
56    /// Merges by timestamp — existing candles are not duplicated.
57    /// Candles are inserted in time order, evicting oldest if over capacity.
58    pub fn backfill(&mut self, candles: Vec<Candle>) {
59        // Collect existing timestamps for dedup
60        let existing_times: std::collections::HashSet<u64> =
61            self.candles.iter().map(|c| c.time).collect();
62
63        let mut new_candles: Vec<Candle> = candles
64            .into_iter()
65            .filter(|c| !existing_times.contains(&c.time))
66            .collect();
67
68        // Sort by time ascending
69        new_candles.sort_by_key(|c| c.time);
70
71        // Merge: rebuild the buffer with all candles sorted
72        let mut all: Vec<Candle> = self.candles.drain(..).collect();
73        all.extend(new_candles);
74        all.sort_by_key(|c| c.time);
75
76        // Keep only the most recent `capacity` candles
77        if all.len() > self.capacity {
78            all = all.split_off(all.len() - self.capacity);
79        }
80
81        self.candles = all.into_iter().collect();
82    }
83}
84
85#[cfg(test)]
86mod tests {
87    use super::*;
88
89    fn make_candle(time: u64, close: f64) -> Candle {
90        Candle {
91            time,
92            open: close - 1.0,
93            high: close + 1.0,
94            low: close - 2.0,
95            close,
96            volume: 100.0,
97        }
98    }
99
100    #[test]
101    fn push_and_len() {
102        let mut buf = CandleBuffer::new("BTC-PERP".into());
103        assert!(buf.is_empty());
104        buf.push(make_candle(1000, 50000.0));
105        assert_eq!(buf.len(), 1);
106        assert_eq!(buf.symbol(), "BTC-PERP");
107    }
108
109    #[test]
110    fn capacity_eviction() {
111        let mut buf = CandleBuffer::with_capacity("ETH".into(), 3);
112        buf.push(make_candle(1, 100.0));
113        buf.push(make_candle(2, 200.0));
114        buf.push(make_candle(3, 300.0));
115        buf.push(make_candle(4, 400.0));
116        assert_eq!(buf.len(), 3);
117        assert_eq!(buf.as_slice()[0].time, 2); // oldest evicted
118        assert_eq!(buf.last().unwrap().time, 4);
119    }
120
121    #[test]
122    fn backfill_dedup() {
123        let mut buf = CandleBuffer::with_capacity("BTC".into(), 5);
124        buf.push(make_candle(3, 300.0));
125        buf.push(make_candle(5, 500.0));
126
127        // Backfill with some overlapping timestamps
128        buf.backfill(vec![
129            make_candle(1, 100.0),
130            make_candle(3, 300.0), // duplicate
131            make_candle(4, 400.0),
132        ]);
133
134        assert_eq!(buf.len(), 4); // 1, 3, 4, 5
135        let times: Vec<u64> = buf.as_slice().iter().map(|c| c.time).collect();
136        assert_eq!(times, vec![1, 3, 4, 5]);
137    }
138
139    #[test]
140    fn backfill_respects_capacity() {
141        let mut buf = CandleBuffer::with_capacity("BTC".into(), 3);
142        buf.backfill(vec![
143            make_candle(1, 100.0),
144            make_candle(2, 200.0),
145            make_candle(3, 300.0),
146            make_candle(4, 400.0),
147            make_candle(5, 500.0),
148        ]);
149        assert_eq!(buf.len(), 3);
150        assert_eq!(buf.as_slice()[0].time, 3); // keeps most recent 3
151    }
152
153    #[test]
154    fn last_returns_most_recent() {
155        let mut buf = CandleBuffer::new("SOL".into());
156        assert!(buf.last().is_none());
157        buf.push(make_candle(1, 10.0));
158        buf.push(make_candle(2, 20.0));
159        assert_eq!(buf.last().unwrap().close, 20.0);
160    }
161}
hyper_agent_core/candle_buffer.rs

hyper_agent_core/
candle_buffer.rs
