fermentation/aggregate/
basic.rs

1use std::marker::PhantomData;
2use std::time::Instant;
3use crate::{ForwardDecay, Item};
4use crate::aggregate::Aggregator;
5use crate::g::{Exponential, Function};
6
7/// Decayed aggregate sum, count and average over a stream of items.
8///
9/// ## Examples
10/// ### Basic Aggregation
11/// ```rust
12/// use std::time::{Duration, Instant};
13/// use fermentation::{ForwardDecay, g};
14/// use fermentation::aggregate::{BasicAggregator, Aggregator};
15///
16/// let decay = ForwardDecay::new(Instant::now(), g::Polynomial::new(2));
17/// let landmark = decay.landmark();
18/// let now = landmark + Duration::from_secs(10);
19/// let stream = vec![
20///     (landmark + Duration::from_secs(5), 4.0),
21///     (landmark + Duration::from_secs(7), 8.0),
22///     (landmark + Duration::from_secs(3), 3.0),
23///     (landmark + Duration::from_secs(8), 6.0),
24///     (landmark + Duration::from_secs(4), 4.0),
25/// ];
26///
27/// let mut aggregator = BasicAggregator::new(decay);
28///
29/// for item in stream {
30///     aggregator.update(item);
31/// }
32///
33/// let epsilon = 0.01;
34///
35/// assert_eq!(aggregator.sum(now), 9.67);
36/// assert_eq!(aggregator.count(now), 1.63);
37/// assert!(aggregator.average() >= (5.93 - epsilon) && aggregator.average() <= (5.93 + epsilon));
38/// ```
39///
40/// ### Update Landmark
41/// ```rust
42/// use std::time::{Duration, Instant};
43/// use fermentation::{ForwardDecay, g};
44/// use fermentation::aggregate::{BasicAggregator, Aggregator};
45///
46/// let decay = ForwardDecay::new(Instant::now(), g::Exponential::new(0.2));
47/// let landmark = decay.landmark();
48/// let new_landmark = landmark + Duration::from_secs(1);
49/// let now = landmark + Duration::from_secs(10);
50/// let stream = vec![
51///     (landmark + Duration::from_secs(5), 4.0),
52///     (landmark + Duration::from_secs(7), 8.0),
53///     (landmark + Duration::from_secs(3), 3.0),
54///     (landmark + Duration::from_secs(8), 6.0),
55///     (landmark + Duration::from_secs(4), 4.0),
56/// ];
57///
58/// let mut aggregator = BasicAggregator::new(decay);
59/// let mut clone = aggregator.clone();
60///
61/// clone.reset(new_landmark);
62///
63/// for item in stream {
64///     aggregator.update(item);
65///     clone.update(item);
66/// }
67///
68/// aggregator.update_landmark(new_landmark);
69///
70/// let epsilon = 0.0001;
71///
72/// assert!((aggregator.sum(now) - clone.sum(now)).abs() < epsilon);
73/// assert!((aggregator.count(now) - clone.count(now)).abs() < epsilon);
74/// assert!((aggregator.average() - clone.average()).abs() < epsilon);
75/// ```
76#[derive(Copy, Clone)]
77pub struct BasicAggregator<G, I> {
78    decay: ForwardDecay<G>,
79    sum: f64,
80    count: f64,
81    _phantom_data: PhantomData<I>
82}
83
84impl<G, I> Aggregator for BasicAggregator<G, I> where G: Function, I: Item {
85    type Item = I;
86
87    fn update(&mut self, item: I) {
88        let static_weight = self.decay.static_weight(&item);
89
90        self.sum += static_weight * item.value();
91        self.count += static_weight;
92    }
93
94    fn reset(&mut self, landmark: Instant) {
95        self.decay.set_landmark(landmark);
96        self.sum = 0.0;
97        self.count = 0.0;
98    }
99}
100
101impl<I> BasicAggregator<Exponential, I>
102where
103    I: Item,
104{
105    pub fn update_landmark(&mut self, landmark: Instant) {
106        let age = self.decay.set_landmark(landmark);
107        let factor = self.decay.g().invoke(age);
108
109        self.sum /= factor;
110        self.count /= factor;
111    }
112}
113
114impl<G, I> BasicAggregator<G, I>
115where
116    G: Function,
117    I: Item,
118{
119    pub fn new(decay: ForwardDecay<G>) -> Self {
120        Self {
121            decay,
122            sum: 0.0,
123            count: 0.0,
124            _phantom_data: Default::default()
125        }
126    }
127
128    pub fn sum(&self, timestamp: Instant) -> f64 {
129        self.sum / self.decay.normalizing_factor(timestamp)
130    }
131
132    pub fn static_sum(&self) -> f64 {
133        self.sum
134    }
135
136    pub fn count(&self, timestamp: Instant) -> f64 {
137        self.count / self.decay.normalizing_factor(timestamp)
138    }
139
140    pub fn static_count(&self) -> f64 {
141        self.count
142    }
143
144    pub fn average(&self) -> f64 {
145        self.sum / self.count
146    }
147
148    pub fn decay(&mut self) -> &ForwardDecay<G> {
149        &self.decay
150    }
151}
152
153#[cfg(test)]
154mod tests {
155    use std::ops::Add;
156    use std::time::{Duration, Instant};
157    use crate::g;
158    use super::*;
159
160    #[test]
161    fn example() {
162        let landmark = Instant::now();
163        let now = landmark + Duration::from_secs(10);
164        let stream = vec![
165            (landmark.add(Duration::from_secs(5)), 4.0),
166            (landmark.add(Duration::from_secs(7)), 8.0),
167            (landmark.add(Duration::from_secs(3)), 3.0),
168            (landmark.add(Duration::from_secs(8)), 6.0),
169            (landmark.add(Duration::from_secs(4)), 4.0),
170        ];
171
172        let fd = ForwardDecay::new(landmark, g::Polynomial::new(2));
173        let mut aggregator = BasicAggregator::new(fd);
174
175        for item in stream {
176            aggregator.update(item);
177        }
178
179        let epsilon = 0.01;
180
181        assert_eq!(aggregator.sum(now), 9.67);
182        assert_eq!(aggregator.static_sum(), 967.0);
183        assert_eq!(aggregator.count(now), 1.63);
184        assert_eq!(aggregator.static_count(), 163.0);
185        assert!(aggregator.average() >= (5.93 - epsilon) && aggregator.average() <= (5.93 + epsilon));
186    }
187}
fermentation/aggregate/basic.rs

fermentation/aggregate/
basic.rs
