llm_optimizer_processor/aggregation/
minmax.rs

1use super::trait_::{Aggregator, ToF64};
2use anyhow::anyhow;
3use serde::{Deserialize, Serialize};
4
5/// Accumulator for min aggregation
6#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
7pub struct MinAccumulator {
8    min: Option<f64>,
9    count: u64,
10}
11
12/// Accumulator for max aggregation
13#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
14pub struct MaxAccumulator {
15    max: Option<f64>,
16    count: u64,
17}
18
19/// Min aggregator - tracks the minimum value seen
20///
21/// This aggregator tracks the minimum value across all values processed.
22/// Useful for tracking best-case latency, minimum cost, etc.
23///
24/// # Examples
25///
26/// ```
27/// use processor::aggregation::{Aggregator, MinAggregator};
28///
29/// let mut agg = MinAggregator::new();
30/// agg.update(30.0).unwrap();
31/// agg.update(10.0).unwrap();
32/// agg.update(20.0).unwrap();
33///
34/// assert_eq!(agg.finalize().unwrap(), 10.0);
35/// ```
36#[derive(Debug, Clone)]
37pub struct MinAggregator<T> {
38    min: Option<f64>,
39    count: u64,
40    _phantom: std::marker::PhantomData<T>,
41}
42
43impl<T: ToF64 + Clone> MinAggregator<T> {
44    /// Create a new min aggregator
45    pub fn new() -> Self {
46        Self {
47            min: None,
48            count: 0,
49            _phantom: std::marker::PhantomData,
50        }
51    }
52
53    /// Get the current minimum value
54    pub fn min(&self) -> Option<f64> {
55        self.min
56    }
57}
58
59impl<T: ToF64 + Clone> Default for MinAggregator<T> {
60    fn default() -> Self {
61        Self::new()
62    }
63}
64
65impl<T: ToF64 + Clone + Send + Sync + std::fmt::Debug> Aggregator for MinAggregator<T> {
66    type Input = T;
67    type Output = f64;
68    type Accumulator = MinAccumulator;
69
70    fn new() -> Self {
71        Self::new()
72    }
73
74    fn update(&mut self, value: <Self as Aggregator>::Input) -> anyhow::Result<()> {
75        let val = value.to_f64();
76        self.min = Some(match self.min {
77            Some(current) => current.min(val),
78            None => val,
79        });
80        self.count += 1;
81        Ok(())
82    }
83
84    fn finalize(&self) -> anyhow::Result<<Self as Aggregator>::Output> {
85        self.min.ok_or_else(|| anyhow!("Cannot compute minimum of zero values"))
86    }
87
88    fn accumulator(&self) -> <Self as Aggregator>::Accumulator {
89        MinAccumulator {
90            min: self.min,
91            count: self.count,
92        }
93    }
94
95    fn merge(&mut self, other: <Self as Aggregator>::Accumulator) -> anyhow::Result<()> {
96        self.min = match (self.min, other.min) {
97            (Some(a), Some(b)) => Some(a.min(b)),
98            (Some(a), None) => Some(a),
99            (None, Some(b)) => Some(b),
100            (None, None) => None,
101        };
102        self.count += other.count;
103        Ok(())
104    }
105
106    fn reset(&mut self) {
107        self.min = None;
108        self.count = 0;
109    }
110
111    fn count(&self) -> u64 {
112        self.count
113    }
114}
115
116/// Max aggregator - tracks the maximum value seen
117///
118/// This aggregator tracks the maximum value across all values processed.
119/// Useful for tracking worst-case latency, maximum cost, peak usage, etc.
120///
121/// # Examples
122///
123/// ```
124/// use processor::aggregation::{Aggregator, MaxAggregator};
125///
126/// let mut agg = MaxAggregator::new();
127/// agg.update(10.0).unwrap();
128/// agg.update(30.0).unwrap();
129/// agg.update(20.0).unwrap();
130///
131/// assert_eq!(agg.finalize().unwrap(), 30.0);
132/// ```
133#[derive(Debug, Clone)]
134pub struct MaxAggregator<T> {
135    max: Option<f64>,
136    count: u64,
137    _phantom: std::marker::PhantomData<T>,
138}
139
140impl<T: ToF64 + Clone> MaxAggregator<T> {
141    /// Create a new max aggregator
142    pub fn new() -> Self {
143        Self {
144            max: None,
145            count: 0,
146            _phantom: std::marker::PhantomData,
147        }
148    }
149
150    /// Get the current maximum value
151    pub fn max(&self) -> Option<f64> {
152        self.max
153    }
154}
155
156impl<T: ToF64 + Clone> Default for MaxAggregator<T> {
157    fn default() -> Self {
158        Self::new()
159    }
160}
161
162impl<T: ToF64 + Clone + Send + Sync + std::fmt::Debug> Aggregator for MaxAggregator<T> {
163    type Input = T;
164    type Output = f64;
165    type Accumulator = MaxAccumulator;
166
167    fn new() -> Self {
168        Self::new()
169    }
170
171    fn update(&mut self, value: <Self as Aggregator>::Input) -> anyhow::Result<()> {
172        let val = value.to_f64();
173        self.max = Some(match self.max {
174            Some(current) => current.max(val),
175            None => val,
176        });
177        self.count += 1;
178        Ok(())
179    }
180
181    fn finalize(&self) -> anyhow::Result<<Self as Aggregator>::Output> {
182        self.max.ok_or_else(|| anyhow!("Cannot compute maximum of zero values"))
183    }
184
185    fn accumulator(&self) -> <Self as Aggregator>::Accumulator {
186        MaxAccumulator {
187            max: self.max,
188            count: self.count,
189        }
190    }
191
192    fn merge(&mut self, other: <Self as Aggregator>::Accumulator) -> anyhow::Result<()> {
193        self.max = match (self.max, other.max) {
194            (Some(a), Some(b)) => Some(a.max(b)),
195            (Some(a), None) => Some(a),
196            (None, Some(b)) => Some(b),
197            (None, None) => None,
198        };
199        self.count += other.count;
200        Ok(())
201    }
202
203    fn reset(&mut self) {
204        self.max = None;
205        self.count = 0;
206    }
207
208    fn count(&self) -> u64 {
209        self.count
210    }
211}
212
213#[cfg(test)]
214mod tests {
215    use super::*;
216
217    #[test]
218    fn test_min_basic() {
219        let mut agg = MinAggregator::<f64>::new();
220        assert!(agg.is_empty());
221
222        agg.update(30.0).unwrap();
223        agg.update(10.0).unwrap();
224        agg.update(20.0).unwrap();
225
226        assert_eq!(agg.count(), 3);
227        assert_eq!(agg.finalize().unwrap(), 10.0);
228        assert_eq!(agg.min().unwrap(), 10.0);
229    }
230
231    #[test]
232    fn test_max_basic() {
233        let mut agg = MaxAggregator::<f64>::new();
234        assert!(agg.is_empty());
235
236        agg.update(10.0).unwrap();
237        agg.update(30.0).unwrap();
238        agg.update(20.0).unwrap();
239
240        assert_eq!(agg.count(), 3);
241        assert_eq!(agg.finalize().unwrap(), 30.0);
242        assert_eq!(agg.max().unwrap(), 30.0);
243    }
244
245    #[test]
246    fn test_min_batch() {
247        let mut agg = MinAggregator::<f64>::new();
248        agg.update_batch(&[50.0, 10.0, 30.0, 20.0, 40.0]).unwrap();
249
250        assert_eq!(agg.finalize().unwrap(), 10.0);
251    }
252
253    #[test]
254    fn test_max_batch() {
255        let mut agg = MaxAggregator::<f64>::new();
256        agg.update_batch(&[10.0, 50.0, 30.0, 20.0, 40.0]).unwrap();
257
258        assert_eq!(agg.finalize().unwrap(), 50.0);
259    }
260
261    #[test]
262    fn test_min_merge() {
263        let mut agg1 = MinAggregator::<f64>::new();
264        agg1.update(30.0).unwrap();
265        agg1.update(20.0).unwrap();
266
267        let mut agg2 = MinAggregator::<f64>::new();
268        agg2.update(10.0).unwrap();
269        agg2.update(40.0).unwrap();
270
271        let acc2 = agg2.accumulator();
272        agg1.merge(acc2).unwrap();
273
274        assert_eq!(agg1.finalize().unwrap(), 10.0);
275        assert_eq!(agg1.count(), 4);
276    }
277
278    #[test]
279    fn test_max_merge() {
280        let mut agg1 = MaxAggregator::<f64>::new();
281        agg1.update(20.0).unwrap();
282        agg1.update(30.0).unwrap();
283
284        let mut agg2 = MaxAggregator::<f64>::new();
285        agg2.update(50.0).unwrap();
286        agg2.update(10.0).unwrap();
287
288        let acc2 = agg2.accumulator();
289        agg1.merge(acc2).unwrap();
290
291        assert_eq!(agg1.finalize().unwrap(), 50.0);
292        assert_eq!(agg1.count(), 4);
293    }
294
295    #[test]
296    fn test_min_reset() {
297        let mut agg = MinAggregator::<f64>::new();
298        agg.update(10.0).unwrap();
299        agg.update(20.0).unwrap();
300
301        assert_eq!(agg.finalize().unwrap(), 10.0);
302
303        agg.reset();
304        assert_eq!(agg.count(), 0);
305        assert!(agg.min().is_none());
306    }
307
308    #[test]
309    fn test_max_reset() {
310        let mut agg = MaxAggregator::<f64>::new();
311        agg.update(10.0).unwrap();
312        agg.update(20.0).unwrap();
313
314        assert_eq!(agg.finalize().unwrap(), 20.0);
315
316        agg.reset();
317        assert_eq!(agg.count(), 0);
318        assert!(agg.max().is_none());
319    }
320
321    #[test]
322    fn test_min_serialization() {
323        let mut agg = MinAggregator::<f64>::new();
324        agg.update(10.0).unwrap();
325        agg.update(20.0).unwrap();
326
327        let acc = agg.accumulator();
328        let serialized = serde_json::to_string(&acc).unwrap();
329        let deserialized: MinAccumulator = serde_json::from_str(&serialized).unwrap();
330
331        assert_eq!(acc, deserialized);
332        assert_eq!(deserialized.min.unwrap(), 10.0);
333    }
334
335    #[test]
336    fn test_max_serialization() {
337        let mut agg = MaxAggregator::<f64>::new();
338        agg.update(10.0).unwrap();
339        agg.update(20.0).unwrap();
340
341        let acc = agg.accumulator();
342        let serialized = serde_json::to_string(&acc).unwrap();
343        let deserialized: MaxAccumulator = serde_json::from_str(&serialized).unwrap();
344
345        assert_eq!(acc, deserialized);
346        assert_eq!(deserialized.max.unwrap(), 20.0);
347    }
348
349    #[test]
350    fn test_min_empty() {
351        let agg = MinAggregator::<f64>::new();
352        assert!(agg.finalize().is_err());
353        assert!(agg.min().is_none());
354    }
355
356    #[test]
357    fn test_max_empty() {
358        let agg = MaxAggregator::<f64>::new();
359        assert!(agg.finalize().is_err());
360        assert!(agg.max().is_none());
361    }
362
363    #[test]
364    fn test_min_negative() {
365        let mut agg = MinAggregator::<f64>::new();
366        agg.update(-10.0).unwrap();
367        agg.update(5.0).unwrap();
368        agg.update(-20.0).unwrap();
369
370        assert_eq!(agg.finalize().unwrap(), -20.0);
371    }
372
373    #[test]
374    fn test_max_negative() {
375        let mut agg = MaxAggregator::<f64>::new();
376        agg.update(-10.0).unwrap();
377        agg.update(-5.0).unwrap();
378        agg.update(-20.0).unwrap();
379
380        assert_eq!(agg.finalize().unwrap(), -5.0);
381    }
382
383    #[test]
384    fn test_min_with_different_types() {
385        let mut agg = MinAggregator::<i32>::new();
386        agg.update(30).unwrap();
387        agg.update(10).unwrap();
388        agg.update(20).unwrap();
389        assert_eq!(agg.finalize().unwrap(), 10.0);
390    }
391
392    #[test]
393    fn test_max_with_different_types() {
394        let mut agg = MaxAggregator::<i32>::new();
395        agg.update(10).unwrap();
396        agg.update(30).unwrap();
397        agg.update(20).unwrap();
398        assert_eq!(agg.finalize().unwrap(), 30.0);
399    }
400}
llm_optimizer_processor/aggregation/minmax.rs

llm_optimizer_processor/aggregation/
minmax.rs
