sklears_utils/
time_series.rs

1//! Time Series Utilities
2//!
3//! This module provides comprehensive time series data structures and utilities
4//! for temporal data processing in machine learning applications.
5
6use crate::UtilsError;
7use chrono::{DateTime, Local, TimeZone, Utc};
8use scirs2_core::ndarray::{Array1, Array2, Axis};
9use std::collections::{BTreeMap, HashMap, VecDeque};
10use std::time::{Duration, SystemTime, UNIX_EPOCH};
11
12/// Timestamp representation for time series data
13#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
14pub struct Timestamp {
15    pub timestamp: i64, // Unix timestamp in milliseconds
16}
17
18impl Timestamp {
19    /// Create a new timestamp from milliseconds since Unix epoch
20    pub fn from_millis(millis: i64) -> Self {
21        Self { timestamp: millis }
22    }
23
24    /// Create a new timestamp from seconds since Unix epoch
25    pub fn from_secs(secs: i64) -> Self {
26        Self {
27            timestamp: secs * 1000,
28        }
29    }
30
31    /// Create a timestamp from the current time
32    pub fn now() -> Self {
33        let duration = SystemTime::now()
34            .duration_since(UNIX_EPOCH)
35            .unwrap_or(Duration::from_secs(0));
36        Self::from_millis(duration.as_millis() as i64)
37    }
38
39    /// Convert to milliseconds
40    pub fn as_millis(&self) -> i64 {
41        self.timestamp
42    }
43
44    /// Convert to seconds
45    pub fn as_secs(&self) -> i64 {
46        self.timestamp / 1000
47    }
48
49    /// Convert to `DateTime<Utc>`
50    pub fn to_datetime_utc(&self) -> DateTime<Utc> {
51        let naive = DateTime::from_timestamp_millis(self.timestamp)
52            .unwrap_or_default()
53            .naive_utc();
54        DateTime::from_naive_utc_and_offset(naive, Utc)
55    }
56
57    /// Add duration to timestamp
58    pub fn add_duration(&self, duration: Duration) -> Self {
59        Self::from_millis(self.timestamp + duration.as_millis() as i64)
60    }
61
62    /// Subtract duration from timestamp
63    pub fn sub_duration(&self, duration: Duration) -> Self {
64        Self::from_millis(self.timestamp - duration.as_millis() as i64)
65    }
66}
67
68/// Time series data point with timestamp and value
69#[derive(Debug, Clone)]
70pub struct TimeSeriesPoint<T> {
71    pub timestamp: Timestamp,
72    pub value: T,
73}
74
75impl<T> TimeSeriesPoint<T> {
76    pub fn new(timestamp: Timestamp, value: T) -> Self {
77        Self { timestamp, value }
78    }
79}
80
81/// Time series data structure with efficient temporal indexing
82#[derive(Debug, Clone)]
83pub struct TimeSeries<T> {
84    data: BTreeMap<Timestamp, T>,
85    metadata: HashMap<String, String>,
86}
87
88impl<T: Clone> Default for TimeSeries<T> {
89    fn default() -> Self {
90        Self::new()
91    }
92}
93
94impl<T: Clone> TimeSeries<T> {
95    /// Create a new empty time series
96    pub fn new() -> Self {
97        Self {
98            data: BTreeMap::new(),
99            metadata: HashMap::new(),
100        }
101    }
102
103    /// Create time series from vectors of timestamps and values
104    pub fn from_vecs(timestamps: Vec<Timestamp>, values: Vec<T>) -> Result<Self, UtilsError> {
105        if timestamps.len() != values.len() {
106            return Err(UtilsError::ShapeMismatch {
107                expected: vec![timestamps.len()],
108                actual: vec![values.len()],
109            });
110        }
111
112        let mut ts = Self::new();
113        for (timestamp, value) in timestamps.into_iter().zip(values) {
114            ts.insert(timestamp, value);
115        }
116        Ok(ts)
117    }
118
119    /// Insert a new data point
120    pub fn insert(&mut self, timestamp: Timestamp, value: T) {
121        self.data.insert(timestamp, value);
122    }
123
124    /// Get value at specific timestamp
125    pub fn get(&self, timestamp: &Timestamp) -> Option<&T> {
126        self.data.get(timestamp)
127    }
128
129    /// Get the number of data points
130    pub fn len(&self) -> usize {
131        self.data.len()
132    }
133
134    /// Check if time series is empty
135    pub fn is_empty(&self) -> bool {
136        self.data.is_empty()
137    }
138
139    /// Get the first timestamp
140    pub fn first_timestamp(&self) -> Option<Timestamp> {
141        self.data.keys().next().copied()
142    }
143
144    /// Get the last timestamp
145    pub fn last_timestamp(&self) -> Option<Timestamp> {
146        self.data.keys().next_back().copied()
147    }
148
149    /// Get data points in a time range
150    pub fn range(&self, start: Timestamp, end: Timestamp) -> Vec<TimeSeriesPoint<T>> {
151        self.data
152            .range(start..=end)
153            .map(|(&timestamp, value)| TimeSeriesPoint::new(timestamp, value.clone()))
154            .collect()
155    }
156
157    /// Get all timestamps
158    pub fn timestamps(&self) -> Vec<Timestamp> {
159        self.data.keys().copied().collect()
160    }
161
162    /// Get all values
163    pub fn values(&self) -> Vec<T> {
164        self.data.values().cloned().collect()
165    }
166
167    /// Set metadata
168    pub fn set_metadata(&mut self, key: String, value: String) {
169        self.metadata.insert(key, value);
170    }
171
172    /// Get metadata
173    pub fn get_metadata(&self, key: &str) -> Option<&String> {
174        self.metadata.get(key)
175    }
176
177    /// Resample time series to regular intervals
178    pub fn resample(
179        &self,
180        interval: Duration,
181        aggregation: AggregationMethod,
182    ) -> Result<TimeSeries<f64>, UtilsError>
183    where
184        T: Into<f64> + Copy,
185    {
186        if self.is_empty() {
187            return Ok(TimeSeries::new());
188        }
189
190        let start = self.first_timestamp().expect("operation should succeed");
191        let end = self.last_timestamp().expect("operation should succeed");
192        let mut resampled = TimeSeries::new();
193
194        let mut current = start;
195        while current.timestamp <= end.timestamp {
196            let window_end = current.add_duration(interval);
197            let window_data: Vec<f64> = self
198                .range(current, window_end)
199                .into_iter()
200                .map(|point| point.value.into())
201                .collect();
202
203            if !window_data.is_empty() {
204                let aggregated = match aggregation {
205                    AggregationMethod::Mean => {
206                        window_data.iter().sum::<f64>() / window_data.len() as f64
207                    }
208                    AggregationMethod::Sum => window_data.iter().sum(),
209                    AggregationMethod::Min => {
210                        window_data.iter().fold(f64::INFINITY, |a, &b| a.min(b))
211                    }
212                    AggregationMethod::Max => {
213                        window_data.iter().fold(f64::NEG_INFINITY, |a, &b| a.max(b))
214                    }
215                    AggregationMethod::First => window_data[0],
216                    AggregationMethod::Last => window_data[window_data.len() - 1],
217                };
218                resampled.insert(current, aggregated);
219            }
220
221            current = current.add_duration(interval);
222        }
223
224        Ok(resampled)
225    }
226}
227
228/// Aggregation methods for resampling
229#[derive(Debug, Clone, Copy)]
230pub enum AggregationMethod {
231    Mean,
232    Sum,
233    Min,
234    Max,
235    First,
236    Last,
237}
238
239/// Sliding window for time series analysis
240#[derive(Debug, Clone)]
241pub struct SlidingWindow<T> {
242    window_size: Duration,
243    data: VecDeque<TimeSeriesPoint<T>>,
244}
245
246impl<T: Clone> SlidingWindow<T> {
247    /// Create a new sliding window with specified size
248    pub fn new(window_size: Duration) -> Self {
249        Self {
250            window_size,
251            data: VecDeque::new(),
252        }
253    }
254
255    /// Add a new data point and maintain window size
256    pub fn add(&mut self, point: TimeSeriesPoint<T>) {
257        self.data.push_back(point.clone());
258
259        // Remove old data points outside the window
260        let cutoff = point.timestamp.sub_duration(self.window_size);
261        while let Some(front) = self.data.front() {
262            if front.timestamp < cutoff {
263                self.data.pop_front();
264            } else {
265                break;
266            }
267        }
268    }
269
270    /// Get current window data
271    pub fn current_window(&self) -> Vec<TimeSeriesPoint<T>> {
272        self.data.iter().cloned().collect()
273    }
274
275    /// Get window size
276    pub fn len(&self) -> usize {
277        self.data.len()
278    }
279
280    /// Check if window is empty
281    pub fn is_empty(&self) -> bool {
282        self.data.is_empty()
283    }
284
285    /// Compute statistics for numeric data in the window
286    pub fn compute_stats(&self) -> WindowStats
287    where
288        T: Into<f64> + Copy,
289    {
290        if self.is_empty() {
291            return WindowStats::default();
292        }
293
294        let values: Vec<f64> = self.data.iter().map(|p| p.value.into()).collect();
295        let n = values.len() as f64;
296        let sum = values.iter().sum::<f64>();
297        let mean = sum / n;
298        let variance = values.iter().map(|x| (x - mean).powi(2)).sum::<f64>() / n;
299        let std_dev = variance.sqrt();
300        let min = values.iter().fold(f64::INFINITY, |a, &b| a.min(b));
301        let max = values.iter().fold(f64::NEG_INFINITY, |a, &b| a.max(b));
302
303        WindowStats {
304            count: values.len(),
305            mean,
306            std_dev,
307            min,
308            max,
309            sum,
310        }
311    }
312}
313
314/// Statistics for sliding window
315#[derive(Debug, Clone, Default)]
316pub struct WindowStats {
317    pub count: usize,
318    pub mean: f64,
319    pub std_dev: f64,
320    pub min: f64,
321    pub max: f64,
322    pub sum: f64,
323}
324
325/// Temporal indexing utilities
326pub struct TemporalIndex {
327    index: BTreeMap<Timestamp, Vec<usize>>,
328}
329
330impl Default for TemporalIndex {
331    fn default() -> Self {
332        Self::new()
333    }
334}
335
336impl TemporalIndex {
337    /// Create a new temporal index
338    pub fn new() -> Self {
339        Self {
340            index: BTreeMap::new(),
341        }
342    }
343
344    /// Add an entry to the index
345    pub fn add_entry(&mut self, timestamp: Timestamp, id: usize) {
346        self.index.entry(timestamp).or_default().push(id);
347    }
348
349    /// Find entries in time range
350    pub fn find_range(&self, start: Timestamp, end: Timestamp) -> Vec<usize> {
351        self.index
352            .range(start..=end)
353            .flat_map(|(_, ids)| ids.iter().copied())
354            .collect()
355    }
356
357    /// Find entries before timestamp
358    pub fn find_before(&self, timestamp: Timestamp) -> Vec<usize> {
359        self.index
360            .range(..timestamp)
361            .flat_map(|(_, ids)| ids.iter().copied())
362            .collect()
363    }
364
365    /// Find entries after timestamp
366    pub fn find_after(&self, timestamp: Timestamp) -> Vec<usize> {
367        self.index
368            .range((
369                std::ops::Bound::Excluded(timestamp),
370                std::ops::Bound::Unbounded,
371            ))
372            .flat_map(|(_, ids)| ids.iter().copied())
373            .collect()
374    }
375}
376
377/// Time zone utilities
378pub struct TimeZoneUtils;
379
380impl TimeZoneUtils {
381    /// Convert timestamp from one timezone to another
382    pub fn convert_timezone(
383        timestamp: Timestamp,
384        from_tz: &str,
385        to_tz: &str,
386    ) -> Result<Timestamp, UtilsError> {
387        // Simplified timezone conversion - in practice would use proper timezone libraries
388        let datetime_utc = timestamp.to_datetime_utc();
389
390        // This is a simplified implementation
391        // In practice, you'd want to use proper timezone handling
392        match (from_tz, to_tz) {
393            ("UTC", "Local") => {
394                let local = Local.from_utc_datetime(&datetime_utc.naive_utc());
395                Ok(Timestamp::from_millis(local.timestamp_millis()))
396            }
397            ("Local", "UTC") => {
398                // Assume input is local time, convert to UTC
399                Ok(timestamp) // Simplified
400            }
401            _ => Ok(timestamp), // No conversion for unsupported timezones
402        }
403    }
404
405    /// Get timestamp for start of day
406    pub fn start_of_day(timestamp: Timestamp) -> Timestamp {
407        let datetime = timestamp.to_datetime_utc();
408        let start_of_day = datetime
409            .date_naive()
410            .and_hms_opt(0, 0, 0)
411            .expect("operation should succeed");
412        let start_of_day_utc: DateTime<Utc> =
413            DateTime::from_naive_utc_and_offset(start_of_day, Utc);
414        Timestamp::from_millis(start_of_day_utc.timestamp_millis())
415    }
416
417    /// Get timestamp for end of day
418    pub fn end_of_day(timestamp: Timestamp) -> Timestamp {
419        let datetime = timestamp.to_datetime_utc();
420        let end_of_day = datetime
421            .date_naive()
422            .and_hms_opt(23, 59, 59)
423            .expect("operation should succeed");
424        let end_of_day_utc: DateTime<Utc> = DateTime::from_naive_utc_and_offset(end_of_day, Utc);
425        Timestamp::from_millis(end_of_day_utc.timestamp_millis())
426    }
427}
428
429/// Temporal aggregation utilities
430pub struct TemporalAggregator;
431
432impl TemporalAggregator {
433    /// Aggregate time series data by time periods
434    pub fn aggregate_by_period<T>(
435        time_series: &TimeSeries<T>,
436        period: Duration,
437        aggregation: AggregationMethod,
438    ) -> Result<TimeSeries<f64>, UtilsError>
439    where
440        T: Into<f64> + Copy,
441    {
442        time_series.resample(period, aggregation)
443    }
444
445    /// Compute rolling statistics
446    pub fn rolling_statistics<T>(
447        data: &[TimeSeriesPoint<T>],
448        window_size: Duration,
449    ) -> Vec<WindowStats>
450    where
451        T: Into<f64> + Copy + Clone,
452    {
453        let mut results = Vec::new();
454        let mut window = SlidingWindow::new(window_size);
455
456        for point in data {
457            window.add(point.clone());
458            results.push(window.compute_stats());
459        }
460
461        results
462    }
463
464    /// Detect trends in time series data
465    pub fn detect_trend<T>(data: &[TimeSeriesPoint<T>], window_size: usize) -> Vec<TrendDirection>
466    where
467        T: Into<f64> + Copy,
468    {
469        if data.len() < window_size * 2 {
470            return vec![TrendDirection::Stable; data.len()];
471        }
472
473        let mut trends = Vec::new();
474        let values: Vec<f64> = data.iter().map(|p| p.value.into()).collect();
475
476        for i in window_size..(values.len() - window_size) {
477            let before: f64 = values[(i - window_size)..i].iter().sum::<f64>() / window_size as f64;
478            let after: f64 =
479                values[(i + 1)..(i + 1 + window_size)].iter().sum::<f64>() / window_size as f64;
480
481            let trend = if after > before * 1.05 {
482                TrendDirection::Increasing
483            } else if after < before * 0.95 {
484                TrendDirection::Decreasing
485            } else {
486                TrendDirection::Stable
487            };
488
489            trends.push(trend);
490        }
491
492        // Pad with stable for edge cases
493        let mut result = vec![TrendDirection::Stable; window_size];
494        result.extend(trends);
495        result.extend(vec![TrendDirection::Stable; window_size]);
496        result
497    }
498}
499
500/// Trend direction enumeration
501#[derive(Debug, Clone, Copy, PartialEq)]
502pub enum TrendDirection {
503    Increasing,
504    Decreasing,
505    Stable,
506}
507
508/// Lag features generator for time series
509pub struct LagFeatureGenerator;
510
511impl LagFeatureGenerator {
512    /// Generate lag features for time series data
513    pub fn generate_lag_features(
514        data: &Array1<f64>,
515        lags: &[usize],
516    ) -> Result<Array2<f64>, UtilsError> {
517        if data.is_empty() || lags.is_empty() {
518            return Err(UtilsError::EmptyInput);
519        }
520
521        let max_lag = *lags.iter().max().expect("operation should succeed");
522        if data.len() <= max_lag {
523            return Err(UtilsError::InsufficientData {
524                min: max_lag + 1,
525                actual: data.len(),
526            });
527        }
528
529        let n_samples = data.len() - max_lag;
530        let n_features = lags.len() + 1; // +1 for original feature
531        let mut features = Array2::zeros((n_samples, n_features));
532
533        for (i, mut row) in features.axis_iter_mut(Axis(0)).enumerate() {
534            let idx = i + max_lag;
535
536            // Original value
537            row[0] = data[idx];
538
539            // Lag features
540            for (j, &lag) in lags.iter().enumerate() {
541                row[j + 1] = data[idx - lag];
542            }
543        }
544
545        Ok(features)
546    }
547
548    /// Generate differencing features
549    pub fn generate_diff_features(
550        data: &Array1<f64>,
551        orders: &[usize],
552    ) -> Result<Array2<f64>, UtilsError> {
553        if data.is_empty() || orders.is_empty() {
554            return Err(UtilsError::EmptyInput);
555        }
556
557        let max_order = *orders.iter().max().expect("operation should succeed");
558        if data.len() <= max_order {
559            return Err(UtilsError::InsufficientData {
560                min: max_order + 1,
561                actual: data.len(),
562            });
563        }
564
565        let n_samples = data.len() - max_order;
566        let n_features = orders.len();
567        let mut features = Array2::zeros((n_samples, n_features));
568
569        for (j, &order) in orders.iter().enumerate() {
570            let mut diff_data = data.to_owned();
571
572            // Apply differencing 'order' times
573            for _ in 0..order {
574                let mut new_diff = Array1::zeros(diff_data.len() - 1);
575                for i in 0..new_diff.len() {
576                    new_diff[i] = diff_data[i + 1] - diff_data[i];
577                }
578                diff_data = new_diff;
579            }
580
581            // Copy to features matrix
582            for i in 0..n_samples {
583                features[(i, j)] = diff_data[i];
584            }
585        }
586
587        Ok(features)
588    }
589}
590
591#[allow(non_snake_case)]
592#[cfg(test)]
593mod tests {
594    use super::*;
595    use std::time::Duration;
596
597    #[test]
598    fn test_timestamp_creation() {
599        let ts1 = Timestamp::from_secs(1000);
600        let ts2 = Timestamp::from_millis(1_000_000);
601
602        assert_eq!(ts1.as_secs(), 1000);
603        assert_eq!(ts2.as_millis(), 1_000_000);
604        assert_eq!(ts1, ts2);
605    }
606
607    #[test]
608    fn test_time_series_basic_operations() {
609        let mut ts = TimeSeries::new();
610        let ts1 = Timestamp::from_secs(100);
611        let ts2 = Timestamp::from_secs(200);
612
613        ts.insert(ts1, 10.0);
614        ts.insert(ts2, 20.0);
615
616        assert_eq!(ts.len(), 2);
617        assert_eq!(ts.get(&ts1), Some(&10.0));
618        assert_eq!(ts.first_timestamp(), Some(ts1));
619        assert_eq!(ts.last_timestamp(), Some(ts2));
620    }
621
622    #[test]
623    fn test_sliding_window() {
624        let mut window = SlidingWindow::new(Duration::from_secs(10));
625        let base_time = Timestamp::from_secs(100);
626
627        window.add(TimeSeriesPoint::new(base_time, 1.0));
628        window.add(TimeSeriesPoint::new(
629            base_time.add_duration(Duration::from_secs(5)),
630            2.0,
631        ));
632        window.add(TimeSeriesPoint::new(
633            base_time.add_duration(Duration::from_secs(15)),
634            3.0,
635        ));
636
637        assert_eq!(window.len(), 2); // First point should be evicted
638        let stats = window.compute_stats();
639        assert_eq!(stats.count, 2);
640        assert_eq!(stats.mean, 2.5);
641    }
642
643    #[test]
644    fn test_temporal_index() {
645        let mut index = TemporalIndex::new();
646        let ts1 = Timestamp::from_secs(100);
647        let ts2 = Timestamp::from_secs(200);
648        let ts3 = Timestamp::from_secs(300);
649
650        index.add_entry(ts1, 1);
651        index.add_entry(ts2, 2);
652        index.add_entry(ts3, 3);
653
654        let range_results = index.find_range(ts1, ts2);
655        assert_eq!(range_results, vec![1, 2]);
656
657        let before_results = index.find_before(ts2);
658        assert_eq!(before_results, vec![1]);
659    }
660
661    #[test]
662    fn test_lag_feature_generation() {
663        let data = Array1::from(vec![1.0, 2.0, 3.0, 4.0, 5.0]);
664        let lags = vec![1, 2];
665
666        let features = LagFeatureGenerator::generate_lag_features(&data, &lags)
667            .expect("operation should succeed");
668
669        assert_eq!(features.shape(), &[3, 3]); // 3 samples, 3 features (original + 2 lags)
670        assert_eq!(features[(0, 0)], 3.0); // Original value at index 2
671        assert_eq!(features[(0, 1)], 2.0); // Lag 1
672        assert_eq!(features[(0, 2)], 1.0); // Lag 2
673    }
674
675    #[test]
676    fn test_diff_features() {
677        let data = Array1::from(vec![1.0, 3.0, 6.0, 10.0, 15.0]);
678        let orders = vec![1, 2];
679
680        let features = LagFeatureGenerator::generate_diff_features(&data, &orders)
681            .expect("operation should succeed");
682
683        assert_eq!(features.shape(), &[3, 2]); // 3 samples, 2 features (1st and 2nd diff)
684        assert_eq!(features[(0, 0)], 2.0); // First difference: 3-1
685        assert_eq!(features[(0, 1)], 1.0); // Second difference: (3-1) - (3-1) = 0, but this is computed differently
686    }
687
688    #[test]
689    fn test_time_series_resampling() {
690        let timestamps = vec![
691            Timestamp::from_secs(0),
692            Timestamp::from_secs(1),
693            Timestamp::from_secs(2),
694            Timestamp::from_secs(3),
695        ];
696        let values = vec![1.0, 2.0, 3.0, 4.0];
697
698        let ts = TimeSeries::from_vecs(timestamps, values).expect("operation should succeed");
699        let resampled = ts
700            .resample(Duration::from_secs(2), AggregationMethod::Mean)
701            .expect("operation should succeed");
702
703        assert!(!resampled.is_empty());
704    }
705}
sklears_utils/time_series.rs

sklears_utils/
time_series.rs
