sklears_decomposition/
streaming.rs

1//! Real-time and streaming decomposition algorithms
2//!
3//! This module provides streaming implementations of decomposition algorithms
4//! for real-time processing of data streams.
5
6use scirs2_core::ndarray::{Array1, Array2, Axis, Zip};
7use scirs2_linalg::compat::{Eigh, SVD};
8use sklears_core::{
9    error::{Result, SklearsError},
10    types::Float,
11};
12use std::collections::VecDeque;
13
14/// Configuration for streaming decomposition algorithms
15#[derive(Debug, Clone)]
16pub struct StreamingConfig {
17    /// Window size for streaming algorithms
18    pub window_size: usize,
19    /// Forgetting factor for exponential weighting
20    pub forgetting_factor: Float,
21    /// Update frequency (number of samples before update)
22    pub update_frequency: usize,
23    /// Minimum number of samples before first update
24    pub min_samples: usize,
25    /// Enable concept drift detection
26    pub detect_drift: bool,
27    /// Drift detection threshold
28    pub drift_threshold: Float,
29}
30
31impl Default for StreamingConfig {
32    fn default() -> Self {
33        Self {
34            window_size: 1000,
35            forgetting_factor: 0.95,
36            update_frequency: 10,
37            min_samples: 50,
38            detect_drift: true,
39            drift_threshold: 0.1,
40        }
41    }
42}
43
44/// Streaming PCA for real-time dimensionality reduction
45#[derive(Debug, Clone)]
46pub struct StreamingPCA {
47    config: StreamingConfig,
48    n_components: usize,
49    // Current state
50    components: Option<Array2<Float>>,
51    explained_variance: Option<Array1<Float>>,
52    mean: Option<Array1<Float>>,
53    n_samples_seen: usize,
54    // Streaming state
55    data_buffer: VecDeque<Array1<Float>>,
56    covariance_matrix: Option<Array2<Float>>,
57    sum_weights: Float,
58    // Drift detection
59    previous_components: Option<Array2<Float>>,
60    drift_detected: bool,
61}
62
63impl StreamingPCA {
64    /// Create a new streaming PCA
65    pub fn new(n_components: usize) -> Self {
66        Self {
67            config: StreamingConfig::default(),
68            n_components,
69            components: None,
70            explained_variance: None,
71            mean: None,
72            n_samples_seen: 0,
73            data_buffer: VecDeque::new(),
74            covariance_matrix: None,
75            sum_weights: 0.0,
76            previous_components: None,
77            drift_detected: false,
78        }
79    }
80
81    /// Configure the streaming PCA
82    pub fn with_config(mut self, config: StreamingConfig) -> Self {
83        self.config = config;
84        self
85    }
86
87    /// Update the PCA with a new data sample
88    pub fn partial_fit(&mut self, sample: &Array1<Float>) -> Result<()> {
89        let n_features = sample.len();
90
91        // Initialize if first sample
92        if self.mean.is_none() {
93            self.mean = Some(Array1::zeros(n_features));
94            self.covariance_matrix = Some(Array2::zeros((n_features, n_features)));
95        }
96
97        // Add to buffer
98        self.data_buffer.push_back(sample.clone());
99        if self.data_buffer.len() > self.config.window_size {
100            self.data_buffer.pop_front();
101        }
102
103        self.n_samples_seen += 1;
104
105        // Update running statistics
106        self.update_statistics(sample)?;
107
108        // Update decomposition if enough samples
109        if self.n_samples_seen >= self.config.min_samples
110            && self
111                .n_samples_seen
112                .is_multiple_of(self.config.update_frequency)
113        {
114            self.update_decomposition()?;
115        }
116
117        Ok(())
118    }
119
120    /// Update running mean and covariance
121    fn update_statistics(&mut self, sample: &Array1<Float>) -> Result<()> {
122        let weight = 1.0 / (self.n_samples_seen as Float);
123        let alpha = self.config.forgetting_factor;
124
125        if let Some(ref mut mean) = self.mean {
126            let delta = sample - &*mean;
127            mean.zip_mut_with(&delta, |m, d| *m += d * weight);
128
129            if let Some(ref mut cov) = self.covariance_matrix {
130                let delta_mean = sample - &*mean;
131                for i in 0..sample.len() {
132                    for j in 0..sample.len() {
133                        cov[[i, j]] =
134                            alpha * cov[[i, j]] + (1.0 - alpha) * delta_mean[i] * delta_mean[j];
135                    }
136                }
137            }
138        }
139
140        Ok(())
141    }
142
143    /// Update the PCA decomposition
144    fn update_decomposition(&mut self) -> Result<()> {
145        if let Some(ref cov) = self.covariance_matrix {
146            // Symmetrize covariance matrix for numerical stability
147            let symmetric_cov = (cov + &cov.t()) / 2.0;
148
149            // Eigendecomposition of covariance matrix
150            let (eigenvalues, eigenvectors) = symmetric_cov
151                .eigh(scirs2_linalg::compat::UPLO::Upper)
152                .map_err(|e| {
153                    SklearsError::InvalidOperation(format!("Eigendecomposition failed: {:?}", e))
154                })?;
155
156            // Sort by eigenvalue (descending)
157            let mut indices: Vec<usize> = (0..eigenvalues.len()).collect();
158            indices.sort_by(|&i, &j| {
159                eigenvalues[j]
160                    .partial_cmp(&eigenvalues[i])
161                    .expect("operation should succeed")
162            });
163
164            // Select top components
165            let n_components = self.n_components.min(eigenvalues.len());
166            let mut components = Array2::<Float>::zeros((n_components, cov.nrows()));
167            let mut explained_variance = Array1::<Float>::zeros(n_components);
168
169            for (i, &idx) in indices.iter().take(n_components).enumerate() {
170                components.row_mut(i).assign(&eigenvectors.column(idx));
171                explained_variance[i] = eigenvalues[idx].max(0.0);
172            }
173
174            // Detect concept drift
175            if self.config.detect_drift {
176                self.detect_concept_drift(&components)?;
177            }
178
179            self.components = Some(components);
180            self.explained_variance = Some(explained_variance);
181        }
182
183        Ok(())
184    }
185
186    /// Detect concept drift in the data
187    fn detect_concept_drift(&mut self, new_components: &Array2<Float>) -> Result<()> {
188        if let Some(ref prev_components) = self.previous_components {
189            let similarity = self.compute_subspace_similarity(prev_components, new_components)?;
190
191            if similarity < (1.0 - self.config.drift_threshold) {
192                self.drift_detected = true;
193                println!("Concept drift detected! Similarity: {similarity:.4}");
194            } else {
195                self.drift_detected = false;
196            }
197        }
198
199        self.previous_components = Some(new_components.clone());
200        Ok(())
201    }
202
203    /// Compute similarity between two subspaces
204    fn compute_subspace_similarity(&self, a: &Array2<Float>, b: &Array2<Float>) -> Result<Float> {
205        // Compute the principal angles between subspaces
206        let ab = a.dot(&b.t());
207        let (_u, s, _vt) = ab
208            .svd(true)
209            .map_err(|e| SklearsError::InvalidOperation(format!("SVD failed: {:?}", e)))?;
210
211        // Average cosine of principal angles
212        let similarity = s.iter().map(|&x| x.powi(2)).sum::<Float>() / s.len() as Float;
213        Ok(similarity)
214    }
215
216    /// Transform new data using current components
217    pub fn transform(&self, x: &Array2<Float>) -> Result<Array2<Float>> {
218        if let (Some(ref components), Some(ref mean)) = (&self.components, &self.mean) {
219            let centered = x - mean;
220            Ok(centered.dot(&components.t()))
221        } else {
222            Err(SklearsError::NotFitted {
223                operation: "transform".to_string(),
224            })
225        }
226    }
227
228    /// Get the current components
229    pub fn components(&self) -> Option<&Array2<Float>> {
230        self.components.as_ref()
231    }
232
233    /// Get the explained variance
234    pub fn explained_variance(&self) -> Option<&Array1<Float>> {
235        self.explained_variance.as_ref()
236    }
237
238    /// Check if concept drift was detected
239    pub fn is_drift_detected(&self) -> bool {
240        self.drift_detected
241    }
242
243    /// Get number of samples seen
244    pub fn n_samples_seen(&self) -> usize {
245        self.n_samples_seen
246    }
247
248    /// Force an immediate update of the decomposition (useful for real-time scenarios)
249    pub fn force_update(&mut self) -> Result<()> {
250        if self.n_samples_seen >= self.config.min_samples {
251            self.update_decomposition()
252        } else {
253            Err(SklearsError::InvalidParameter {
254                name: "min_samples".to_string(),
255                reason: "Not enough samples for decomposition update".to_string(),
256            })
257        }
258    }
259
260    /// Reset the streaming PCA (useful when drift is detected)
261    pub fn reset(&mut self) {
262        self.components = None;
263        self.explained_variance = None;
264        self.mean = None;
265        self.n_samples_seen = 0;
266        self.data_buffer.clear();
267        self.covariance_matrix = None;
268        self.sum_weights = 0.0;
269        self.previous_components = None;
270        self.drift_detected = false;
271    }
272
273    /// Transform a single sample for real-time processing
274    pub fn transform_sample(&self, sample: &Array1<Float>) -> Result<Array1<Float>> {
275        if let (Some(ref components), Some(ref mean)) = (&self.components, &self.mean) {
276            let centered = sample - mean;
277            Ok(components.dot(&centered))
278        } else {
279            Err(SklearsError::NotFitted {
280                operation: "transform_sample".to_string(),
281            })
282        }
283    }
284
285    /// Get reconstruction error for monitoring model quality in real-time
286    pub fn reconstruction_error(&self, sample: &Array1<Float>) -> Result<Float> {
287        if let (Some(ref components), Some(ref mean)) = (&self.components, &self.mean) {
288            let centered = sample - mean;
289            let transformed = components.dot(&centered);
290            let reconstructed = components.t().dot(&transformed);
291            let error = (&centered - &reconstructed)
292                .mapv(|x| x.powi(2))
293                .sum()
294                .sqrt();
295            Ok(error)
296        } else {
297            Err(SklearsError::NotFitted {
298                operation: "reconstruction_error".to_string(),
299            })
300        }
301    }
302}
303
304/// Streaming ICA for real-time blind source separation
305#[derive(Debug, Clone)]
306pub struct StreamingICA {
307    config: StreamingConfig,
308    n_components: usize,
309    // Current state
310    components: Option<Array2<Float>>,
311    mixing_matrix: Option<Array2<Float>>,
312    mean: Option<Array1<Float>>,
313    n_samples_seen: usize,
314    // Streaming state
315    data_buffer: VecDeque<Array1<Float>>,
316    learning_rate: Float,
317    // Drift detection
318    previous_components: Option<Array2<Float>>,
319    drift_detected: bool,
320}
321
322impl StreamingICA {
323    /// Create a new streaming ICA
324    pub fn new(n_components: usize) -> Self {
325        Self {
326            config: StreamingConfig::default(),
327            n_components,
328            components: None,
329            mixing_matrix: None,
330            mean: None,
331            n_samples_seen: 0,
332            data_buffer: VecDeque::new(),
333            learning_rate: 0.01,
334            previous_components: None,
335            drift_detected: false,
336        }
337    }
338
339    /// Configure the streaming ICA
340    pub fn with_config(mut self, config: StreamingConfig) -> Self {
341        self.config = config;
342        self
343    }
344
345    /// Update the ICA with a new data sample
346    pub fn partial_fit(&mut self, sample: &Array1<Float>) -> Result<()> {
347        let n_features = sample.len();
348
349        // Initialize if first sample
350        if self.mean.is_none() {
351            self.mean = Some(Array1::zeros(n_features));
352            self.components = Some(Array2::eye(self.n_components));
353        }
354
355        // Add to buffer
356        self.data_buffer.push_back(sample.clone());
357        if self.data_buffer.len() > self.config.window_size {
358            self.data_buffer.pop_front();
359        }
360
361        self.n_samples_seen += 1;
362
363        // Update running mean
364        self.update_mean(sample)?;
365
366        // Update ICA decomposition
367        if self.n_samples_seen >= self.config.min_samples
368            && self
369                .n_samples_seen
370                .is_multiple_of(self.config.update_frequency)
371        {
372            self.update_decomposition()?;
373        }
374
375        Ok(())
376    }
377
378    /// Update running mean
379    fn update_mean(&mut self, sample: &Array1<Float>) -> Result<()> {
380        let weight = 1.0 / (self.n_samples_seen as Float);
381
382        if let Some(ref mut mean) = self.mean {
383            let delta = sample - &*mean;
384            mean.zip_mut_with(&delta, |m, d| *m += d * weight);
385        }
386
387        Ok(())
388    }
389
390    /// Update the ICA decomposition using online learning
391    fn update_decomposition(&mut self) -> Result<()> {
392        if let (Some(ref mean), Some(_)) = (&self.mean, &self.components) {
393            // Process recent samples
394            let recent_samples: Vec<_> = self
395                .data_buffer
396                .iter()
397                .rev()
398                .take(self.config.update_frequency)
399                .cloned()
400                .collect();
401
402            for sample in recent_samples {
403                let centered = &sample - mean;
404                // Inline the update to avoid borrow checker issues
405                if let Some(ref mut components) = self.components {
406                    let y = components.dot(&centered);
407                    let g: Array1<Float> = y.mapv(|val| val.tanh());
408                    let g_prime: Array1<Float> = y.mapv(|val| 1.0 - val.tanh().powi(2));
409
410                    let outer_product = g
411                        .insert_axis(Axis(1))
412                        .dot(&centered.clone().insert_axis(Axis(0)));
413                    let diagonal = Array2::from_diag(&g_prime);
414
415                    let gradient = outer_product - diagonal.dot(&*components);
416                    *components += &(gradient * self.learning_rate);
417                }
418            }
419
420            // Orthogonalize components
421            if let Some(ref mut components) = self.components {
422                Self::orthogonalize_components_static(components)?;
423            }
424
425            // Detect concept drift
426            if self.config.detect_drift {
427                if let Some(ref components) = self.components {
428                    let components_clone = components.clone();
429                    self.detect_concept_drift(&components_clone)?;
430                }
431            }
432        }
433
434        Ok(())
435    }
436
437    /// Online ICA update using natural gradient
438    #[allow(dead_code)]
439    fn online_ica_update(
440        &mut self,
441        components: &mut Array2<Float>,
442        x: &Array1<Float>,
443    ) -> Result<()> {
444        // Project data
445        let y = components.dot(x);
446
447        // Compute nonlinearity (tanh)
448        let g: Array1<Float> = y.mapv(|val| val.tanh());
449        let g_prime: Array1<Float> = y.mapv(|val| 1.0 - val.tanh().powi(2));
450
451        // Natural gradient update
452        let outer_product = g.insert_axis(Axis(1)).dot(&x.clone().insert_axis(Axis(0)));
453        let _identity = Array2::<Float>::eye(self.n_components);
454        let diagonal = Array2::from_diag(&g_prime);
455
456        let gradient = outer_product - diagonal.dot(components);
457
458        // Update components with learning rate
459        *components += &(gradient * self.learning_rate);
460
461        // Orthogonalize components
462        self.orthogonalize_components(components)?;
463
464        Ok(())
465    }
466
467    /// Orthogonalize components using Gram-Schmidt
468    #[allow(dead_code)]
469    fn orthogonalize_components(&self, components: &mut Array2<Float>) -> Result<()> {
470        Self::orthogonalize_components_static(components)
471    }
472
473    /// Static version of orthogonalize_components for borrow checker
474    fn orthogonalize_components_static(components: &mut Array2<Float>) -> Result<()> {
475        for i in 0..components.nrows() {
476            // Normalize current component
477            let norm = components.row(i).dot(&components.row(i)).sqrt();
478            if norm > 1e-10 {
479                components.row_mut(i).mapv_inplace(|x| x / norm);
480            }
481
482            // Orthogonalize against previous components
483            for j in 0..i {
484                let projection = components.row(i).dot(&components.row(j));
485                let row_j = components.row(j).to_owned();
486                Zip::from(components.row_mut(i))
487                    .and(&row_j)
488                    .for_each(|a, &b| *a -= projection * b);
489            }
490        }
491
492        Ok(())
493    }
494
495    /// Detect concept drift in the data
496    fn detect_concept_drift(&mut self, new_components: &Array2<Float>) -> Result<()> {
497        if let Some(ref prev_components) = self.previous_components {
498            let similarity = self.compute_subspace_similarity(prev_components, new_components)?;
499
500            if similarity < (1.0 - self.config.drift_threshold) {
501                self.drift_detected = true;
502                println!("Concept drift detected in ICA! Similarity: {similarity:.4}");
503            } else {
504                self.drift_detected = false;
505            }
506        }
507
508        self.previous_components = Some(new_components.clone());
509        Ok(())
510    }
511
512    /// Compute similarity between two subspaces
513    fn compute_subspace_similarity(&self, a: &Array2<Float>, b: &Array2<Float>) -> Result<Float> {
514        let ab = a.dot(&b.t());
515        let (_u, s, _vt) = ab
516            .svd(true)
517            .map_err(|e| SklearsError::InvalidOperation(format!("SVD failed: {:?}", e)))?;
518
519        let similarity = s.iter().map(|&x| x.powi(2)).sum::<Float>() / s.len() as Float;
520        Ok(similarity)
521    }
522
523    /// Transform new data using current components
524    pub fn transform(&self, x: &Array2<Float>) -> Result<Array2<Float>> {
525        if let (Some(ref components), Some(ref mean)) = (&self.components, &self.mean) {
526            let centered = x - mean;
527            Ok(components.dot(&centered.t()).t().to_owned())
528        } else {
529            Err(SklearsError::NotFitted {
530                operation: "transform".to_string(),
531            })
532        }
533    }
534
535    /// Get the current components
536    pub fn components(&self) -> Option<&Array2<Float>> {
537        self.components.as_ref()
538    }
539
540    /// Check if concept drift was detected
541    pub fn is_drift_detected(&self) -> bool {
542        self.drift_detected
543    }
544
545    /// Get number of samples seen
546    pub fn n_samples_seen(&self) -> usize {
547        self.n_samples_seen
548    }
549
550    /// Transform a single sample for real-time signal processing
551    pub fn transform_sample(&self, sample: &Array1<Float>) -> Result<Array1<Float>> {
552        if let (Some(ref components), Some(ref mean)) = (&self.components, &self.mean) {
553            let centered = sample - mean;
554            Ok(components.dot(&centered))
555        } else {
556            Err(SklearsError::NotFitted {
557                operation: "transform_sample".to_string(),
558            })
559        }
560    }
561
562    /// Get the mixing matrix for source reconstruction
563    pub fn mixing_matrix(&self) -> Option<&Array2<Float>> {
564        self.mixing_matrix.as_ref()
565    }
566
567    /// Reconstruct the original sources from mixed signal
568    pub fn separate_sources(&self, mixed_sample: &Array1<Float>) -> Result<Array1<Float>> {
569        if let (Some(ref components), Some(ref mean)) = (&self.components, &self.mean) {
570            let centered = mixed_sample - mean;
571            Ok(components.dot(&centered))
572        } else {
573            Err(SklearsError::NotFitted {
574                operation: "separate_sources".to_string(),
575            })
576        }
577    }
578
579    /// Set learning rate for adaptive tuning
580    pub fn set_learning_rate(&mut self, learning_rate: Float) {
581        self.learning_rate = learning_rate;
582    }
583
584    /// Get current learning rate
585    pub fn learning_rate(&self) -> Float {
586        self.learning_rate
587    }
588
589    /// Force immediate update for real-time scenarios
590    pub fn force_update(&mut self) -> Result<()> {
591        if self.n_samples_seen >= self.config.min_samples {
592            self.update_decomposition()
593        } else {
594            Err(SklearsError::InvalidParameter {
595                name: "min_samples".to_string(),
596                reason: "Not enough samples for decomposition update".to_string(),
597            })
598        }
599    }
600
601    /// Reset for new signal processing session
602    pub fn reset(&mut self) {
603        self.components = None;
604        self.mixing_matrix = None;
605        self.mean = None;
606        self.n_samples_seen = 0;
607        self.data_buffer.clear();
608        self.previous_components = None;
609        self.drift_detected = false;
610    }
611}
612
613/// Adaptive decomposition method that can switch between algorithms
614#[derive(Debug, Clone)]
615pub struct AdaptiveDecomposition {
616    #[allow(dead_code)]
617    config: StreamingConfig,
618    // Current active algorithm
619    active_algorithm: String,
620    // Algorithm instances
621    streaming_pca: Option<StreamingPCA>,
622    streaming_ica: Option<StreamingICA>,
623    // Performance metrics
624    reconstruction_errors: VecDeque<Float>,
625    adaptation_threshold: Float,
626}
627
628impl AdaptiveDecomposition {
629    /// Create a new adaptive decomposition
630    pub fn new(n_components: usize) -> Self {
631        Self {
632            config: StreamingConfig::default(),
633            active_algorithm: "PCA".to_string(),
634            streaming_pca: Some(StreamingPCA::new(n_components)),
635            streaming_ica: Some(StreamingICA::new(n_components)),
636            reconstruction_errors: VecDeque::new(),
637            adaptation_threshold: 0.05,
638        }
639    }
640
641    /// Update with new data sample
642    pub fn partial_fit(&mut self, sample: &Array1<Float>) -> Result<()> {
643        // Update both algorithms
644        if let Some(ref mut pca) = self.streaming_pca {
645            pca.partial_fit(sample)?;
646        }
647        if let Some(ref mut ica) = self.streaming_ica {
648            ica.partial_fit(sample)?;
649        }
650
651        // Evaluate performance and adapt if needed
652        self.evaluate_and_adapt(sample)?;
653
654        Ok(())
655    }
656
657    /// Evaluate performance and adapt algorithm if needed
658    fn evaluate_and_adapt(&mut self, sample: &Array1<Float>) -> Result<()> {
659        // Compute reconstruction errors for both algorithms
660        let pca_error = self.compute_reconstruction_error_pca(sample)?;
661        let ica_error = self.compute_reconstruction_error_ica(sample)?;
662
663        // Store errors
664        self.reconstruction_errors
665            .push_back(pca_error.min(ica_error));
666        if self.reconstruction_errors.len() > 100 {
667            self.reconstruction_errors.pop_front();
668        }
669
670        // Decide which algorithm to use
671        if pca_error < ica_error - self.adaptation_threshold {
672            if self.active_algorithm != "PCA" {
673                println!("Switching to PCA (error: {pca_error:.4})");
674                self.active_algorithm = "PCA".to_string();
675            }
676        } else if ica_error < pca_error - self.adaptation_threshold
677            && self.active_algorithm != "ICA"
678        {
679            println!("Switching to ICA (error: {ica_error:.4})");
680            self.active_algorithm = "ICA".to_string();
681        }
682
683        Ok(())
684    }
685
686    /// Compute reconstruction error for PCA
687    fn compute_reconstruction_error_pca(&self, sample: &Array1<Float>) -> Result<Float> {
688        if let Some(ref pca) = self.streaming_pca {
689            if let (Some(components), Some(mean)) = (pca.components(), pca.mean.as_ref()) {
690                let centered = sample - mean;
691                let transformed = components.dot(&centered);
692                let reconstructed = components.t().dot(&transformed) + mean;
693                let error = (sample - &reconstructed).mapv(|x| x.powi(2)).sum().sqrt();
694                return Ok(error);
695            }
696        }
697        Ok(Float::INFINITY)
698    }
699
700    /// Compute reconstruction error for ICA
701    fn compute_reconstruction_error_ica(&self, sample: &Array1<Float>) -> Result<Float> {
702        if let Some(ref ica) = self.streaming_ica {
703            if let (Some(components), Some(mean)) = (ica.components(), ica.mean.as_ref()) {
704                let centered = sample - mean;
705                let transformed = components.dot(&centered);
706                let reconstructed = components.t().dot(&transformed) + mean;
707                let error = (sample - &reconstructed).mapv(|x| x.powi(2)).sum().sqrt();
708                return Ok(error);
709            }
710        }
711        Ok(Float::INFINITY)
712    }
713
714    /// Get the active algorithm name
715    pub fn active_algorithm(&self) -> &str {
716        &self.active_algorithm
717    }
718
719    /// Get average reconstruction error
720    pub fn average_reconstruction_error(&self) -> Float {
721        if self.reconstruction_errors.is_empty() {
722            0.0
723        } else {
724            self.reconstruction_errors.iter().sum::<Float>()
725                / self.reconstruction_errors.len() as Float
726        }
727    }
728}
729
730#[allow(non_snake_case)]
731#[cfg(test)]
732mod tests {
733    use super::*;
734
735    #[test]
736    fn test_streaming_pca_basic() {
737        let mut pca = StreamingPCA::new(2);
738
739        // Add some samples
740        for i in 0..100 {
741            let sample = Array1::from_vec(vec![i as Float, (i * 2) as Float, (i * 3) as Float]);
742            pca.partial_fit(&sample).expect("sampling should succeed");
743        }
744
745        assert!(pca.n_samples_seen() == 100);
746        assert!(pca.components().is_some());
747    }
748
749    #[test]
750    fn test_streaming_ica_basic() {
751        let mut ica = StreamingICA::new(2);
752
753        // Add some samples
754        for i in 0..100 {
755            let sample = Array1::from_vec(vec![i as Float, (i * 2) as Float]);
756            ica.partial_fit(&sample).expect("sampling should succeed");
757        }
758
759        assert!(ica.n_samples_seen() == 100);
760        assert!(ica.components().is_some());
761    }
762
763    #[test]
764    fn test_adaptive_decomposition() {
765        let mut adaptive = AdaptiveDecomposition::new(2);
766
767        // Add some samples
768        for i in 0..100 {
769            let sample = Array1::from_vec(vec![i as Float, (i * 2) as Float]);
770            adaptive
771                .partial_fit(&sample)
772                .expect("sampling should succeed");
773        }
774
775        assert!(adaptive.active_algorithm() == "PCA" || adaptive.active_algorithm() == "ICA");
776    }
777}
sklears_decomposition/streaming.rs

sklears_decomposition/
streaming.rs
