chronix_analytics/multivariate/
context.rs

1//! Multi-series context: time-aligned columnar matrix of multiple series.
2
3use crate::multivariate::error::MultivariateError;
4
5/// Column-major matrix for aligned multi-series data.
6///
7/// Layout: `data[series_idx][time_idx]` — each inner Vec is one series.
8#[derive(Debug, Clone)]
9pub struct ColumnarMatrix {
10    /// One row per series, contiguous for SIMD-friendly access.
11    pub data: Vec<Vec<f64>>,
12    /// Common aligned timestamps.
13    pub timestamps: Vec<i64>,
14    /// Series identifiers (measurement + tags).
15    pub series_ids: Vec<String>,
16}
17
18impl ColumnarMatrix {
19    /// Number of series (columns in the conceptual layout).
20    pub fn n_series(&self) -> usize {
21        self.data.len()
22    }
23
24    /// Number of time points.
25    pub fn n_timestamps(&self) -> usize {
26        self.timestamps.len()
27    }
28
29    /// Get a single series by index.
30    pub fn series(&self, idx: usize) -> Option<&[f64]> {
31        self.data.get(idx).map(std::vec::Vec::as_slice)
32    }
33
34    /// Get a single series by name.
35    pub fn series_by_name(&self, name: &str) -> Option<&[f64]> {
36        self.series_ids
37            .iter()
38            .position(|s| s == name)
39            .and_then(|idx| self.series(idx))
40    }
41}
42
43/// Time-aligned context for multi-series analysis.
44///
45/// Constructed by providing raw per-series data possibly at different sampling
46/// rates; the context resamples and aligns them to a common timeline.
47#[derive(Debug, Clone)]
48pub struct MultiSeriesContext {
49    /// Aligned columnar data matrix.
50    pub matrix: ColumnarMatrix,
51}
52
53impl MultiSeriesContext {
54    /// Builds a context from individual series data.
55    ///
56    /// Each entry in `series` is (id, timestamps, values).
57    /// All series will be resampled to the finest common interval via linear
58    /// interpolation.
59    ///
60    /// `max_points` overrides the default resampling cap (1 000 000). Pass
61    /// `None` to use the default.
62    pub fn build(
63        series: Vec<(String, Vec<i64>, Vec<f64>)>,
64        max_series: Option<usize>,
65    ) -> Result<Self, MultivariateError> {
66        Self::build_with_max_points(series, max_series, None)
67    }
68
69    /// Like [`build`](Self::build) but with a configurable resampling cap.
70    #[allow(clippy::needless_pass_by_value)] // series vector is consumed conceptually (aligned copies)
71    pub fn build_with_max_points(
72        series: Vec<(String, Vec<i64>, Vec<f64>)>,
73        max_series: Option<usize>,
74        max_points: Option<usize>,
75    ) -> Result<Self, MultivariateError> {
76        if series.is_empty() {
77            return Err(MultivariateError::InsufficientData { min: 1, got: 0 });
78        }
79        let max = max_series.unwrap_or(100);
80        if series.len() > max {
81            return Err(MultivariateError::InvalidParameter(format!(
82                "too many series: {} > max {}",
83                series.len(),
84                max
85            )));
86        }
87
88        // Find global time range and finest interval
89        let mut global_min = i64::MAX;
90        let mut global_max = i64::MIN;
91        let mut finest_interval = i64::MAX;
92
93        for (_, ts, _) in &series {
94            if ts.is_empty() {
95                continue;
96            }
97            global_min = global_min.min(ts[0]);
98            if let Some(&last) = ts.last() {
99                global_max = global_max.max(last);
100            }
101            for w in ts.windows(2) {
102                let delta = w[1] - w[0];
103                if delta > 0 {
104                    finest_interval = finest_interval.min(delta);
105                }
106            }
107        }
108
109        // If we never found a valid range (all series have 0 or 1 points), use single-point fallback
110        if global_min > global_max {
111            // Gather single-point timestamps
112            let mut only_ts = i64::MIN;
113            for (_, ts, _) in &series {
114                if let Some(&t) = ts.first() {
115                    only_ts = only_ts.max(t);
116                }
117            }
118            if only_ts == i64::MIN {
119                return Err(MultivariateError::InsufficientData { min: 1, got: 0 });
120            }
121            global_min = only_ts;
122            global_max = only_ts;
123        }
124
125        if finest_interval == i64::MAX || finest_interval <= 0 {
126            finest_interval = 1_000_000_000; // fallback 1s
127        }
128
129        // Build common timeline
130        let n_points = ((global_max - global_min) / finest_interval + 1) as usize;
131        // Reject instead of silently truncating at the safety cap.
132        // The cap is configurable; defaults to 1M if not specified.
133        const DEFAULT_MAX_POINTS: usize = 1_000_000;
134        let cap = max_points.unwrap_or(DEFAULT_MAX_POINTS);
135        if n_points > cap {
136            return Err(MultivariateError::InvalidParameter(format!(
137                "resampled timeline would produce {n_points} points (max {cap}); \
138                 reduce the time range or increase the sampling interval"
139            )));
140        }
141        let timestamps: Vec<i64> = (0..n_points)
142            .map(|i| global_min + i as i64 * finest_interval)
143            .collect();
144
145        // Resample each series onto the common timeline via linear interpolation
146        let mut data = Vec::with_capacity(series.len());
147        let mut ids = Vec::with_capacity(series.len());
148
149        for (id, ts, vals) in &series {
150            ids.push(id.clone());
151            let aligned = linear_interpolate(ts, vals, &timestamps);
152            data.push(aligned);
153        }
154
155        Ok(Self {
156            matrix: ColumnarMatrix {
157                data,
158                timestamps,
159                series_ids: ids,
160            },
161        })
162    }
163}
164
165/// Linear interpolation of (ts, vals) onto `target_ts`.
166///
167/// Returns `NaN` for target timestamps outside the input range
168/// to prevent false correlations from constant extrapolation.
169fn linear_interpolate(ts: &[i64], vals: &[f64], target_ts: &[i64]) -> Vec<f64> {
170    let mut result = Vec::with_capacity(target_ts.len());
171    if ts.is_empty() || vals.is_empty() {
172        result.resize(target_ts.len(), f64::NAN);
173        return result;
174    }
175    let mut j = 0usize;
176    for &t in target_ts {
177        // Advance j until ts[j] <= t < ts[j+1]
178        while j + 1 < ts.len() && ts[j + 1] <= t {
179            j += 1;
180        }
181        if t < ts[0] {
182            // Before first timestamp — return NaN.
183            result.push(f64::NAN);
184        } else if j + 1 >= ts.len() {
185            // At or past the last timestamp.
186            if t == ts[ts.len() - 1] {
187                // Exact match at the last point — return its value.
188                result.push(vals[ts.len() - 1]);
189            } else {
190                // Beyond range — return NaN instead of extrapolating.
191                result.push(f64::NAN);
192            }
193        } else {
194            let dt = (ts[j + 1] - ts[j]) as f64;
195            let frac = if dt > 0.0 {
196                (t - ts[j]) as f64 / dt
197            } else {
198                0.0
199            };
200            result.push(vals[j] + frac * (vals[j + 1] - vals[j]));
201        }
202    }
203    result
204}
205
206#[cfg(test)]
207mod tests {
208    use super::*;
209
210    #[test]
211    fn build_aligned_context() {
212        let s1 = (
213            "cpu".to_string(),
214            vec![0, 1_000_000_000, 2_000_000_000],
215            vec![10.0, 20.0, 30.0],
216        );
217        let s2 = (
218            "mem".to_string(),
219            vec![0, 2_000_000_000],
220            vec![100.0, 200.0],
221        );
222        let ctx = MultiSeriesContext::build(vec![s1, s2], None).unwrap();
223        assert_eq!(ctx.matrix.n_series(), 2);
224        assert!(ctx.matrix.n_timestamps() >= 2);
225    }
226
227    #[test]
228    fn series_by_name() {
229        let s1 = ("cpu".to_string(), vec![0, 1_000_000_000], vec![10.0, 20.0]);
230        let ctx = MultiSeriesContext::build(vec![s1], None).unwrap();
231        assert!(ctx.matrix.series_by_name("cpu").is_some());
232        assert!(ctx.matrix.series_by_name("missing").is_none());
233    }
234
235    #[test]
236    fn too_many_series() {
237        let series: Vec<_> = (0..200)
238            .map(|i| (format!("s{i}"), vec![0i64, 1], vec![0.0, 1.0]))
239            .collect();
240        assert!(MultiSeriesContext::build(series, Some(100)).is_err());
241    }
242
243    #[test]
244    fn interpolation_correctness() {
245        let ts = vec![0i64, 10];
246        let vals = vec![0.0, 100.0];
247        let target = vec![0, 5, 10];
248        let result = linear_interpolate(&ts, &vals, &target);
249        assert!((result[0] - 0.0).abs() < 1e-6);
250        assert!((result[1] - 50.0).abs() < 1e-6);
251        assert!((result[2] - 100.0).abs() < 1e-6);
252    }
253}
chronix_analytics/multivariate/context.rs

chronix_analytics/multivariate/
context.rs
