chronix-analytics 0.7.0

Analytics engine for Chronix — preprocessing, forecasting, anomaly detection, multivariate analysis, model lifecycle, and streaming analytics on a SIMD/rayon CPU compute core
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
//! Champion/Challenger A/B testing for model evaluation.
//!
//! During prediction, traffic is probabilistically routed between the
//! champion and a challenger model.  Both predictions are logged against
//! actuals, and an evaluator determines when to promote the challenger.

use std::collections::VecDeque;

use parking_lot::Mutex;

use crate::lifecycle::registry::VersionAccuracy;

/// Relative+absolute tolerance comparison for float metrics.
///
/// Uses `1e-9` relative tolerance with a `1e-12` absolute floor,
/// correctly handling values across all magnitudes (MAPE 0.01–100+,
/// RMSE 0.001–1000+).
#[inline]
fn approx_eq(a: f64, b: f64) -> bool {
    let diff = (a - b).abs();
    let magnitude = a.abs().max(b.abs());
    diff <= 1e-9 * magnitude.max(1e-3)
}

/// Criterion for promoting a challenger to champion.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum PromotionCriteria {
    /// Promote when challenger's MAPE is lower.
    LowerMape,
    /// Promote when challenger's RMSE is lower.
    LowerRmse,
    /// Promote when both MAPE and RMSE are lower.
    LowerBoth,
}

/// Configuration for an A/B test.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ABTestConfig {
    /// Fraction of traffic routed to challenger `[0.0, 1.0]`. Default: `0.10`.
    pub challenger_traffic_pct: f64,
    /// Number of evaluation windows before a decision is made. Default: 10.
    pub evaluation_windows: usize,
    /// Promotion criterion.
    pub criteria: PromotionCriteria,
}

impl Default for ABTestConfig {
    fn default() -> Self {
        Self {
            challenger_traffic_pct: 0.10,
            evaluation_windows: 10,
            criteria: PromotionCriteria::LowerMape,
        }
    }
}

/// Result of an A/B test evaluation round.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ABTestResult {
    /// Promote the challenger to champion.
    Promote,
    /// Keep the current champion; retire the challenger.
    Keep,
    /// Not enough data for a decision yet.
    Inconclusive,
}

/// Logged prediction record.
#[derive(Debug, Clone)]
struct PredictionRecord {
    actual: f64,
    champion_prediction: f64,
    challenger_prediction: f64,
}

/// Evaluator that collects prediction pairs and judges the test.
///
/// Thread-safe — mutable state is wrapped in `Mutex`.
#[derive(Debug)]
pub struct ABTestEvaluator {
    config: ABTestConfig,
    inner: Mutex<ABTestInner>,
}

#[derive(Debug)]
struct ABTestInner {
    records: VecDeque<PredictionRecord>,
    consecutive_wins: usize,
    last_eval_count: usize,
}

impl ABTestEvaluator {
    /// Creates a new A/B test evaluator with the given configuration.
    pub fn new(config: ABTestConfig) -> Self {
        Self {
            config,
            inner: Mutex::new(ABTestInner {
                records: VecDeque::new(),
                consecutive_wins: 0,
                last_eval_count: 0,
            }),
        }
    }

    /// Decide whether a specific request should be routed to the challenger.
    ///
    /// Uses a simple deterministic hash on the `request_id` for consistent
    /// routing (no randomness in tight loops).
    pub fn should_use_challenger(&self, request_id: u64) -> bool {
        let frac = (request_id % 10000) as f64 / 10000.0;
        frac < self.config.challenger_traffic_pct
    }

    /// Log a prediction pair (both champion and challenger predictions along
    /// with the actual observed value).
    pub fn log_prediction(&self, actual: f64, champion_pred: f64, challenger_pred: f64) {
        let mut inner = self.inner.lock();
        inner.records.push_back(PredictionRecord {
            actual,
            champion_prediction: champion_pred,
            challenger_prediction: challenger_pred,
        });

        // Keep bounded to 10× evaluation_windows
        let max_records = self.config.evaluation_windows * 100;
        while inner.records.len() > max_records {
            inner.records.pop_front();
        }
    }

    /// Evaluate the A/B test based on logged records.
    ///
    /// Computes metrics over the most recent `evaluation_windows` records
    /// to avoid stale early predictions diluting accuracy.
    pub fn evaluate(&self) -> ABTestResult {
        let mut inner = self.inner.lock();
        if inner.records.len() < self.config.evaluation_windows {
            return ABTestResult::Inconclusive;
        }

        // Only count as a new evaluation window when fresh data has arrived.
        let current_count = inner.records.len();
        if current_count == inner.last_eval_count {
            // No new data — return current state without inflating counters.
            return if inner.consecutive_wins >= 3 {
                ABTestResult::Promote
            } else if current_count >= self.config.evaluation_windows * 5
                && inner.consecutive_wins == 0
            {
                ABTestResult::Keep
            } else {
                ABTestResult::Inconclusive
            };
        }
        inner.last_eval_count = current_count;

        // Use only the most recent window of records for metric computation
        let window = self.config.evaluation_windows;
        let recent = inner.records.iter().rev().take(window);
        let actuals: Vec<f64> = recent.clone().map(|r| r.actual).collect();
        let champ_preds: Vec<f64> = recent.clone().map(|r| r.champion_prediction).collect();
        let chal_preds: Vec<f64> = recent.map(|r| r.challenger_prediction).collect();

        let champ_metrics = VersionAccuracy::compute(&actuals, &champ_preds);
        let chal_metrics = VersionAccuracy::compute(&actuals, &chal_preds);

        metrics::gauge!("chronix_model_ab_test_champion_mape").set(champ_metrics.mape);
        metrics::gauge!("chronix_model_ab_test_challenger_mape").set(chal_metrics.mape);

        let challenger_wins = match self.config.criteria {
            PromotionCriteria::LowerMape => {
                !chal_metrics.mape.is_nan()
                    && !champ_metrics.mape.is_nan()
                    && chal_metrics.mape < champ_metrics.mape
            }
            PromotionCriteria::LowerRmse => {
                !chal_metrics.rmse.is_nan()
                    && !champ_metrics.rmse.is_nan()
                    && chal_metrics.rmse < champ_metrics.rmse
            }
            PromotionCriteria::LowerBoth => {
                chal_metrics.rmse < champ_metrics.rmse
                    && !chal_metrics.mape.is_nan()
                    && !champ_metrics.mape.is_nan()
                    && chal_metrics.mape < champ_metrics.mape
            }
        };

        // Tie detection: when challenger and champion produce exactly
        // equal metrics under the active criterion. On a tie the
        // `consecutive_wins` counter is left unchanged — the streak is
        // neither extended nor broken.  This prevents a lucky sequence
        // of ties from inflating the counter, while also avoiding
        // penalising the challenger for matching the champion exactly.
        // Use relative tolerance (1e-9) for tie detection
        // instead of f64::EPSILON (~2.2e-16) which is meaningless for
        // MAPE/RMSE values in the range 0.01-100+.
        let is_tie = match self.config.criteria {
            PromotionCriteria::LowerMape => {
                !chal_metrics.mape.is_nan()
                    && !champ_metrics.mape.is_nan()
                    && approx_eq(chal_metrics.mape, champ_metrics.mape)
            }
            PromotionCriteria::LowerRmse => {
                !chal_metrics.rmse.is_nan()
                    && !champ_metrics.rmse.is_nan()
                    && approx_eq(chal_metrics.rmse, champ_metrics.rmse)
            }
            PromotionCriteria::LowerBoth => {
                !chal_metrics.mape.is_nan()
                    && !champ_metrics.mape.is_nan()
                    && approx_eq(chal_metrics.mape, champ_metrics.mape)
                    && !chal_metrics.rmse.is_nan()
                    && !champ_metrics.rmse.is_nan()
                    && approx_eq(chal_metrics.rmse, champ_metrics.rmse)
            }
        };

        if challenger_wins {
            inner.consecutive_wins += 1;
        } else if is_tie {
            // Tie: preserve existing streak — the challenger matched
            // the champion exactly, so no reason to penalise it.
        } else {
            inner.consecutive_wins = 0;
        }

        // Require sustained wins across multiple evaluations
        if inner.consecutive_wins >= 3 {
            ABTestResult::Promote
        } else if inner.records.len() >= self.config.evaluation_windows * 5
            && inner.consecutive_wins == 0
        {
            ABTestResult::Keep
        } else {
            ABTestResult::Inconclusive
        }
    }

    /// Champion accuracy from logged data.
    pub fn champion_metrics(&self) -> Option<VersionAccuracy> {
        let inner = self.inner.lock();
        if inner.records.is_empty() {
            return None;
        }
        let actuals: Vec<f64> = inner.records.iter().map(|r| r.actual).collect();
        let preds: Vec<f64> = inner
            .records
            .iter()
            .map(|r| r.champion_prediction)
            .collect();
        Some(VersionAccuracy::compute(&actuals, &preds))
    }

    /// Challenger accuracy from logged data.
    pub fn challenger_metrics(&self) -> Option<VersionAccuracy> {
        let inner = self.inner.lock();
        if inner.records.is_empty() {
            return None;
        }
        let actuals: Vec<f64> = inner.records.iter().map(|r| r.actual).collect();
        let preds: Vec<f64> = inner
            .records
            .iter()
            .map(|r| r.challenger_prediction)
            .collect();
        Some(VersionAccuracy::compute(&actuals, &preds))
    }

    /// Number of logged records.
    pub fn record_count(&self) -> usize {
        self.inner.lock().records.len()
    }
}

// -----------------------------------------------------------------------
// Tests
// -----------------------------------------------------------------------
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_routing_fraction() {
        let eval = ABTestEvaluator::new(ABTestConfig {
            challenger_traffic_pct: 0.10,
            ..Default::default()
        });

        let mut challenger_count = 0;
        for id in 0..10000 {
            if eval.should_use_challenger(id) {
                challenger_count += 1;
            }
        }
        // ~10% of 10000 = ~1000
        assert!((900..=1100).contains(&challenger_count));
    }

    #[test]
    fn test_inconclusive_with_few_records() {
        let eval = ABTestEvaluator::new(ABTestConfig {
            evaluation_windows: 10,
            ..Default::default()
        });

        for i in 0..5 {
            eval.log_prediction(100.0, 110.0 + i as f64, 105.0 + i as f64);
        }

        assert_eq!(eval.evaluate(), ABTestResult::Inconclusive);
    }

    #[test]
    fn test_promote_when_challenger_consistently_better() {
        let eval = ABTestEvaluator::new(ABTestConfig {
            evaluation_windows: 5,
            criteria: PromotionCriteria::LowerMape,
            ..Default::default()
        });

        // Challenger is consistently better
        for i in 0..50 {
            let actual = 100.0 + (i as f64 * 0.1).sin();
            let champ_pred = actual + 10.0; // champion off by 10
            let chal_pred = actual + 1.0; // challenger off by 1
            eval.log_prediction(actual, champ_pred, chal_pred);
        }

        // Need 3 consecutive evaluations with challenger winning.
        // Each evaluation requires new data to avoid no-op re-evaluation.
        for batch in 0..3 {
            for i in 0..5 {
                let actual = 100.0 + ((50 + batch * 5 + i) as f64 * 0.1).sin();
                let champ_pred = actual + 10.0;
                let chal_pred = actual + 1.0;
                eval.log_prediction(actual, champ_pred, chal_pred);
            }
            eval.evaluate();
        }
        let result = eval.evaluate();
        assert_eq!(result, ABTestResult::Promote);
    }

    #[test]
    fn test_keep_when_champion_wins() {
        let eval = ABTestEvaluator::new(ABTestConfig {
            evaluation_windows: 5,
            criteria: PromotionCriteria::LowerRmse,
            ..Default::default()
        });

        // Champion is better
        for i in 0..50 {
            let actual = 100.0 + (i as f64 * 0.1).sin();
            let champ_pred = actual + 0.5;
            let chal_pred = actual + 20.0;
            eval.log_prediction(actual, champ_pred, chal_pred);
        }

        let result = eval.evaluate();
        assert_eq!(result, ABTestResult::Keep);
    }

    #[test]
    fn test_metrics_available() {
        let eval = ABTestEvaluator::new(Default::default());
        eval.log_prediction(100.0, 110.0, 105.0);
        eval.log_prediction(200.0, 190.0, 195.0);

        let cm = eval.champion_metrics().unwrap();
        let clm = eval.challenger_metrics().unwrap();
        assert!(cm.mae > 0.0);
        assert!(clm.mae > 0.0);
        assert!(clm.mae < cm.mae); // challenger closer
    }

    #[test]
    fn test_record_count() {
        let eval = ABTestEvaluator::new(Default::default());
        assert_eq!(eval.record_count(), 0);
        eval.log_prediction(1.0, 1.0, 1.0);
        assert_eq!(eval.record_count(), 1);
    }

    /// When challenger MAPE exactly equals champion MAPE (tie),
    /// `consecutive_wins` must be preserved — not reset to zero.
    #[test]
    fn test_ab_test_tie_preserves_streak() {
        let eval = ABTestEvaluator::new(ABTestConfig {
            evaluation_windows: 5,
            criteria: PromotionCriteria::LowerMape,
            ..Default::default()
        });

        // ── Round 1: challenger wins (lower MAPE) ──────────────
        for _ in 0..10 {
            let actual = 100.0;
            let champ_pred = 110.0; // 10% error
            let chal_pred = 101.0; // 1% error
            eval.log_prediction(actual, champ_pred, chal_pred);
        }
        let r1 = eval.evaluate();
        assert_eq!(r1, ABTestResult::Inconclusive); // 1 consecutive win

        // ── Round 2: challenger wins again ─────────────────────
        for _ in 0..5 {
            eval.log_prediction(100.0, 110.0, 101.0);
        }
        let r2 = eval.evaluate();
        assert_eq!(r2, ABTestResult::Inconclusive); // 2 consecutive wins

        // ── Round 3: TIE — identical predictions ──────────────
        // Both champion and challenger predict with the same error.
        for _ in 0..5 {
            eval.log_prediction(100.0, 105.0, 105.0);
        }
        let r3 = eval.evaluate();
        // Streak should NOT be reset; still at 2 consecutive wins
        assert_eq!(r3, ABTestResult::Inconclusive);

        // ── Round 4: challenger wins again ─────────────────────
        for _ in 0..5 {
            eval.log_prediction(100.0, 110.0, 101.0);
        }
        let r4 = eval.evaluate();
        // This should be the 3rd win (2 before tie + 1 now) → Promote
        assert_eq!(r4, ABTestResult::Promote);
    }
}