sklears_multioutput/
streaming.rs

1//! Streaming and Incremental Learning for Multi-Output Prediction
2//!
3//! This module provides algorithms for learning from streaming data with multiple outputs,
4//! including incremental learning, online learning, and concept drift detection.
5#![allow(non_snake_case)] // Standard ML notation: X for feature matrices, K for kernels
6
7// Use SciRS2-Core for arrays and random number generation (SciRS2 Policy)
8use scirs2_core::ndarray::{Array1, Array2, ArrayView2, Axis};
9use sklears_core::{
10    error::{Result as SklResult, SklearsError},
11    traits::{Estimator, Fit, Predict, Untrained},
12    types::Float,
13};
14use std::collections::VecDeque;
15
16// ============================================================================
17// Incremental Multi-Output Regression
18// ============================================================================
19
20/// Configuration for incremental multi-output regression
21#[derive(Debug, Clone)]
22pub struct IncrementalMultiOutputRegressionConfig {
23    /// Learning rate for gradient updates
24    pub learning_rate: Float,
25    /// L2 regularization parameter
26    pub alpha: Float,
27    /// Whether to fit intercept
28    pub fit_intercept: bool,
29    /// Maximum number of samples to keep in memory (for computing statistics)
30    pub max_samples: usize,
31    /// Whether to use adaptive learning rate
32    pub adaptive_learning_rate: bool,
33    /// Decay factor for learning rate
34    pub learning_rate_decay: Float,
35}
36
37impl Default for IncrementalMultiOutputRegressionConfig {
38    fn default() -> Self {
39        Self {
40            learning_rate: 0.01,
41            alpha: 0.0001,
42            fit_intercept: true,
43            max_samples: 10000,
44            adaptive_learning_rate: true,
45            learning_rate_decay: 0.999,
46        }
47    }
48}
49
50/// Incremental Multi-Output Regressor
51///
52/// Online learning algorithm that can learn from data streams with multiple outputs.
53/// Uses stochastic gradient descent with optional adaptive learning rates.
54///
55/// # Examples
56///
57/// ```rust
58/// use sklears_multioutput::streaming::{IncrementalMultiOutputRegression, IncrementalMultiOutputRegressionConfig};
59/// // Use SciRS2-Core for arrays and random number generation (SciRS2 Policy)
60/// use scirs2_core::ndarray::array;
61/// use sklears_core::traits::{Fit, Predict};
62///
63/// let X = array![[1.0, 2.0], [2.0, 3.0], [3.0, 4.0]];
64/// let y = array![[1.0, 2.0], [2.0, 3.0], [3.0, 4.0]];
65///
66/// let mut model = IncrementalMultiOutputRegression::new();
67/// let trained = model.fit(&X.view(), &y.view()).unwrap();
68///
69/// // Continue learning with new data
70/// let X_new = array![[4.0, 5.0]];
71/// let y_new = array![[4.0, 5.0]];
72/// let updated = trained.partial_fit(&X_new.view(), &y_new.view()).unwrap();
73///
74/// let predictions = updated.predict(&X.view()).unwrap();
75/// assert_eq!(predictions.dim(), (3, 2));
76/// ```
77#[derive(Debug, Clone)]
78pub struct IncrementalMultiOutputRegression<S = Untrained> {
79    state: S,
80    config: IncrementalMultiOutputRegressionConfig,
81}
82
83/// Trained state for Incremental Multi-Output Regression
84#[derive(Debug, Clone)]
85pub struct IncrementalMultiOutputRegressionTrained {
86    /// Coefficient matrix (n_features x n_outputs)
87    pub coef: Array2<Float>,
88    /// Intercept vector (n_outputs)
89    pub intercept: Array1<Float>,
90    /// Number of features
91    pub n_features: usize,
92    /// Number of outputs
93    pub n_outputs: usize,
94    /// Number of samples seen so far
95    pub n_samples_seen: usize,
96    /// Current learning rate
97    pub current_learning_rate: Float,
98    /// Running mean of features (for normalization)
99    pub feature_mean: Array1<Float>,
100    /// Running std of features (for normalization)
101    pub feature_std: Array1<Float>,
102    /// Configuration
103    pub config: IncrementalMultiOutputRegressionConfig,
104}
105
106impl IncrementalMultiOutputRegression<Untrained> {
107    /// Create a new incremental multi-output regressor
108    pub fn new() -> Self {
109        Self {
110            state: Untrained,
111            config: IncrementalMultiOutputRegressionConfig::default(),
112        }
113    }
114
115    /// Set the configuration
116    pub fn config(mut self, config: IncrementalMultiOutputRegressionConfig) -> Self {
117        self.config = config;
118        self
119    }
120
121    /// Set the learning rate
122    pub fn learning_rate(mut self, lr: Float) -> Self {
123        self.config.learning_rate = lr;
124        self
125    }
126
127    /// Set the regularization parameter
128    pub fn alpha(mut self, alpha: Float) -> Self {
129        self.config.alpha = alpha;
130        self
131    }
132
133    /// Set whether to fit intercept
134    pub fn fit_intercept(mut self, fit_intercept: bool) -> Self {
135        self.config.fit_intercept = fit_intercept;
136        self
137    }
138}
139
140impl Default for IncrementalMultiOutputRegression<Untrained> {
141    fn default() -> Self {
142        Self::new()
143    }
144}
145
146impl Fit<ArrayView2<'_, Float>, ArrayView2<'_, Float>>
147    for IncrementalMultiOutputRegression<Untrained>
148{
149    type Fitted = IncrementalMultiOutputRegression<IncrementalMultiOutputRegressionTrained>;
150
151    fn fit(self, X: &ArrayView2<Float>, y: &ArrayView2<Float>) -> SklResult<Self::Fitted> {
152        if X.nrows() != y.nrows() {
153            return Err(SklearsError::InvalidInput(
154                "Number of samples in X and y must match".to_string(),
155            ));
156        }
157
158        let n_samples = X.nrows();
159        let n_features = X.ncols();
160        let n_outputs = y.ncols();
161
162        // Initialize coefficients
163        let mut coef = Array2::zeros((n_features, n_outputs));
164        let mut intercept = Array1::zeros(n_outputs);
165
166        // Compute feature statistics
167        let feature_mean = X
168            .mean_axis(Axis(0))
169            .expect("array should have elements for mean computation");
170        let feature_std = X.std_axis(Axis(0), 0.0);
171
172        let mut current_learning_rate = self.config.learning_rate;
173
174        // Perform initial gradient descent over the batch
175        for _ in 0..10 {
176            // Mini-batch iterations
177            for i in 0..n_samples {
178                let x_i = X.row(i);
179                let y_i = y.row(i);
180
181                // Prediction
182                let pred = coef.t().dot(&x_i) + &intercept;
183
184                // Error
185                let error = &y_i - &pred;
186
187                // Update coefficients using gradient descent
188                for j in 0..n_features {
189                    for k in 0..n_outputs {
190                        let gradient = -error[k] * x_i[j] + self.config.alpha * coef[[j, k]];
191                        coef[[j, k]] -= current_learning_rate * gradient;
192                    }
193                }
194
195                // Update intercept
196                if self.config.fit_intercept {
197                    for k in 0..n_outputs {
198                        intercept[k] += current_learning_rate * error[k];
199                    }
200                }
201            }
202
203            // Decay learning rate
204            if self.config.adaptive_learning_rate {
205                current_learning_rate *= self.config.learning_rate_decay;
206            }
207        }
208
209        Ok(IncrementalMultiOutputRegression {
210            state: IncrementalMultiOutputRegressionTrained {
211                coef,
212                intercept,
213                n_features,
214                n_outputs,
215                n_samples_seen: n_samples,
216                current_learning_rate,
217                feature_mean,
218                feature_std,
219                config: self.config,
220            },
221            config: IncrementalMultiOutputRegressionConfig::default(),
222        })
223    }
224}
225
226impl IncrementalMultiOutputRegression<IncrementalMultiOutputRegressionTrained> {
227    /// Partial fit on new data (incremental learning)
228    pub fn partial_fit(mut self, X: &ArrayView2<Float>, y: &ArrayView2<Float>) -> SklResult<Self> {
229        if X.nrows() != y.nrows() {
230            return Err(SklearsError::InvalidInput(
231                "Number of samples in X and y must match".to_string(),
232            ));
233        }
234
235        if X.ncols() != self.state.n_features {
236            return Err(SklearsError::InvalidInput(format!(
237                "Expected {} features, got {}",
238                self.state.n_features,
239                X.ncols()
240            )));
241        }
242
243        if y.ncols() != self.state.n_outputs {
244            return Err(SklearsError::InvalidInput(format!(
245                "Expected {} outputs, got {}",
246                self.state.n_outputs,
247                y.ncols()
248            )));
249        }
250
251        let n_samples = X.nrows();
252
253        // Update feature statistics (running average)
254        let n_old = self.state.n_samples_seen as Float;
255        let n_new = n_samples as Float;
256        let n_total = n_old + n_new;
257
258        let new_mean = X
259            .mean_axis(Axis(0))
260            .expect("array should have elements for mean computation");
261        self.state.feature_mean = (&self.state.feature_mean * n_old + &new_mean * n_new) / n_total;
262
263        // Perform incremental updates
264        for i in 0..n_samples {
265            let x_i = X.row(i);
266            let y_i = y.row(i);
267
268            // Prediction
269            let pred = self.state.coef.t().dot(&x_i) + &self.state.intercept;
270
271            // Error
272            let error = &y_i - &pred;
273
274            // Update coefficients
275            for j in 0..self.state.n_features {
276                for k in 0..self.state.n_outputs {
277                    let gradient =
278                        -error[k] * x_i[j] + self.state.config.alpha * self.state.coef[[j, k]];
279                    self.state.coef[[j, k]] -= self.state.current_learning_rate * gradient;
280                }
281            }
282
283            // Update intercept
284            if self.state.config.fit_intercept {
285                for k in 0..self.state.n_outputs {
286                    self.state.intercept[k] += self.state.current_learning_rate * error[k];
287                }
288            }
289        }
290
291        // Update statistics
292        self.state.n_samples_seen += n_samples;
293
294        // Decay learning rate
295        if self.state.config.adaptive_learning_rate {
296            self.state.current_learning_rate *= self.state.config.learning_rate_decay;
297        }
298
299        Ok(self)
300    }
301
302    /// Get the current coefficients
303    pub fn coef(&self) -> &Array2<Float> {
304        &self.state.coef
305    }
306
307    /// Get the current intercept
308    pub fn intercept(&self) -> &Array1<Float> {
309        &self.state.intercept
310    }
311
312    /// Get number of samples seen
313    pub fn n_samples_seen(&self) -> usize {
314        self.state.n_samples_seen
315    }
316
317    /// Get current learning rate
318    pub fn current_learning_rate(&self) -> Float {
319        self.state.current_learning_rate
320    }
321}
322
323impl Predict<ArrayView2<'_, Float>, Array2<Float>>
324    for IncrementalMultiOutputRegression<IncrementalMultiOutputRegressionTrained>
325{
326    fn predict(&self, X: &ArrayView2<Float>) -> SklResult<Array2<Float>> {
327        if X.ncols() != self.state.n_features {
328            return Err(SklearsError::InvalidInput(format!(
329                "Expected {} features, got {}",
330                self.state.n_features,
331                X.ncols()
332            )));
333        }
334
335        let n_samples = X.nrows();
336        let mut predictions = Array2::zeros((n_samples, self.state.n_outputs));
337
338        for i in 0..n_samples {
339            let x_i = X.row(i);
340            let pred = self.state.coef.t().dot(&x_i) + &self.state.intercept;
341            predictions.row_mut(i).assign(&pred);
342        }
343
344        Ok(predictions)
345    }
346}
347
348impl Estimator for IncrementalMultiOutputRegression<Untrained> {
349    type Config = IncrementalMultiOutputRegressionConfig;
350    type Error = SklearsError;
351    type Float = Float;
352
353    fn config(&self) -> &Self::Config {
354        &self.config
355    }
356}
357
358impl Estimator for IncrementalMultiOutputRegression<IncrementalMultiOutputRegressionTrained> {
359    type Config = IncrementalMultiOutputRegressionConfig;
360    type Error = SklearsError;
361    type Float = Float;
362
363    fn config(&self) -> &Self::Config {
364        &self.state.config
365    }
366}
367
368// ============================================================================
369// Streaming Multi-Output with Mini-Batches
370// ============================================================================
371
372/// Configuration for streaming multi-output learning
373#[derive(Debug, Clone)]
374pub struct StreamingMultiOutputConfig {
375    /// Mini-batch size for streaming updates
376    pub batch_size: usize,
377    /// Maximum buffer size before forced update
378    pub max_buffer_size: usize,
379    /// Learning rate
380    pub learning_rate: Float,
381    /// Whether to detect concept drift
382    pub detect_drift: bool,
383    /// Window size for drift detection
384    pub drift_window_size: usize,
385    /// Threshold for drift detection
386    pub drift_threshold: Float,
387}
388
389impl Default for StreamingMultiOutputConfig {
390    fn default() -> Self {
391        Self {
392            batch_size: 32,
393            max_buffer_size: 1000,
394            learning_rate: 0.01,
395            detect_drift: true,
396            drift_window_size: 100,
397            drift_threshold: 0.1,
398        }
399    }
400}
401
402/// Streaming Multi-Output Learner
403///
404/// Handles streaming data with mini-batch processing and concept drift detection.
405///
406/// # Examples
407///
408/// ```rust
409/// use sklears_multioutput::streaming::{StreamingMultiOutput, StreamingMultiOutputConfig};
410/// // Use SciRS2-Core for arrays and random number generation (SciRS2 Policy)
411/// use scirs2_core::ndarray::array;
412/// use sklears_core::traits::{Fit, Predict};
413///
414/// let X = array![[1.0, 2.0], [2.0, 3.0], [3.0, 4.0]];
415/// let y = array![[1.0, 2.0], [2.0, 3.0], [3.0, 4.0]];
416///
417/// let mut model = StreamingMultiOutput::new()
418///     .batch_size(2)
419///     .learning_rate(0.1);
420///
421/// let trained = model.fit(&X.view(), &y.view()).unwrap();
422///
423/// // Add streaming data
424/// let X_stream = array![[4.0, 5.0]];
425/// let y_stream = array![[4.0, 5.0]];
426/// let updated = trained.update_stream(&X_stream.view(), &y_stream.view()).unwrap();
427///
428/// let predictions = updated.predict(&X.view()).unwrap();
429/// assert_eq!(predictions.dim(), (3, 2));
430/// ```
431#[derive(Debug, Clone)]
432pub struct StreamingMultiOutput<S = Untrained> {
433    state: S,
434    config: StreamingMultiOutputConfig,
435}
436
437/// Trained state for Streaming Multi-Output
438#[derive(Debug, Clone)]
439pub struct StreamingMultiOutputTrained {
440    /// Base incremental model
441    pub base_model: IncrementalMultiOutputRegressionTrained,
442    /// Buffer for mini-batch processing
443    pub buffer_X: VecDeque<Array1<Float>>,
444    pub buffer_y: VecDeque<Array1<Float>>,
445    /// Performance history for drift detection
446    pub error_history: VecDeque<Float>,
447    /// Whether drift was detected
448    pub drift_detected: bool,
449    /// Number of drift events detected
450    pub n_drift_events: usize,
451    /// Configuration
452    pub config: StreamingMultiOutputConfig,
453}
454
455impl StreamingMultiOutput<Untrained> {
456    /// Create a new streaming multi-output learner
457    pub fn new() -> Self {
458        Self {
459            state: Untrained,
460            config: StreamingMultiOutputConfig::default(),
461        }
462    }
463
464    /// Set the configuration
465    pub fn config(mut self, config: StreamingMultiOutputConfig) -> Self {
466        self.config = config;
467        self
468    }
469
470    /// Set the batch size
471    pub fn batch_size(mut self, batch_size: usize) -> Self {
472        self.config.batch_size = batch_size;
473        self
474    }
475
476    /// Set the learning rate
477    pub fn learning_rate(mut self, lr: Float) -> Self {
478        self.config.learning_rate = lr;
479        self
480    }
481
482    /// Enable/disable drift detection
483    pub fn detect_drift(mut self, detect: bool) -> Self {
484        self.config.detect_drift = detect;
485        self
486    }
487}
488
489impl Default for StreamingMultiOutput<Untrained> {
490    fn default() -> Self {
491        Self::new()
492    }
493}
494
495impl Fit<ArrayView2<'_, Float>, ArrayView2<'_, Float>> for StreamingMultiOutput<Untrained> {
496    type Fitted = StreamingMultiOutput<StreamingMultiOutputTrained>;
497
498    fn fit(self, X: &ArrayView2<Float>, y: &ArrayView2<Float>) -> SklResult<Self::Fitted> {
499        // Initialize base model
500        let base_config = IncrementalMultiOutputRegressionConfig {
501            learning_rate: self.config.learning_rate,
502            ..Default::default()
503        };
504
505        let base_model = IncrementalMultiOutputRegression::new()
506            .config(base_config)
507            .fit(X, y)?;
508
509        Ok(StreamingMultiOutput {
510            state: StreamingMultiOutputTrained {
511                base_model: base_model.state,
512                buffer_X: VecDeque::new(),
513                buffer_y: VecDeque::new(),
514                error_history: VecDeque::new(),
515                drift_detected: false,
516                n_drift_events: 0,
517                config: self.config,
518            },
519            config: StreamingMultiOutputConfig::default(),
520        })
521    }
522}
523
524impl StreamingMultiOutput<StreamingMultiOutputTrained> {
525    /// Update with streaming data
526    pub fn update_stream(
527        mut self,
528        X: &ArrayView2<Float>,
529        y: &ArrayView2<Float>,
530    ) -> SklResult<Self> {
531        // Add to buffer
532        for i in 0..X.nrows() {
533            self.state.buffer_X.push_back(X.row(i).to_owned());
534            self.state.buffer_y.push_back(y.row(i).to_owned());
535        }
536
537        // Process if buffer is full
538        if self.state.buffer_X.len() >= self.state.config.batch_size {
539            self = self.process_buffer()?;
540        }
541
542        Ok(self)
543    }
544
545    /// Process the current buffer
546    fn process_buffer(mut self) -> SklResult<Self> {
547        let batch_size = self.state.config.batch_size.min(self.state.buffer_X.len());
548
549        if batch_size == 0 {
550            return Ok(self);
551        }
552
553        // Extract batch from buffer
554        let mut X_batch = Array2::zeros((batch_size, self.state.base_model.n_features));
555        let mut y_batch = Array2::zeros((batch_size, self.state.base_model.n_outputs));
556
557        for i in 0..batch_size {
558            let x = self
559                .state
560                .buffer_X
561                .pop_front()
562                .expect("operation should succeed");
563            let y = self
564                .state
565                .buffer_y
566                .pop_front()
567                .expect("operation should succeed");
568            X_batch.row_mut(i).assign(&x);
569            y_batch.row_mut(i).assign(&y);
570        }
571
572        // Detect drift if enabled
573        if self.state.config.detect_drift {
574            let pred = self.predict(&X_batch.view())?;
575            let error: Float = (&y_batch - &pred)
576                .mapv(|x| x.powi(2))
577                .mean()
578                .expect("array should have elements for mean computation");
579
580            self.state.error_history.push_back(error);
581            if self.state.error_history.len() > self.state.config.drift_window_size {
582                self.state.error_history.pop_front();
583            }
584
585            // Check for drift
586            if self.state.error_history.len() >= self.state.config.drift_window_size {
587                let recent_error: Float = self
588                    .state
589                    .error_history
590                    .iter()
591                    .rev()
592                    .take(self.state.config.drift_window_size / 2)
593                    .sum::<Float>()
594                    / (self.state.config.drift_window_size / 2) as Float;
595
596                let old_error: Float = self
597                    .state
598                    .error_history
599                    .iter()
600                    .take(self.state.config.drift_window_size / 2)
601                    .sum::<Float>()
602                    / (self.state.config.drift_window_size / 2) as Float;
603
604                if recent_error > old_error * (1.0 + self.state.config.drift_threshold) {
605                    self.state.drift_detected = true;
606                    self.state.n_drift_events += 1;
607                    // Could reset model here if needed
608                }
609            }
610        }
611
612        // Update base model
613        let base_wrapper = IncrementalMultiOutputRegression {
614            state: self.state.base_model.clone(),
615            config: IncrementalMultiOutputRegressionConfig::default(),
616        };
617
618        let updated = base_wrapper.partial_fit(&X_batch.view(), &y_batch.view())?;
619        self.state.base_model = updated.state;
620
621        Ok(self)
622    }
623
624    /// Force processing of remaining buffer
625    pub fn flush_buffer(mut self) -> SklResult<Self> {
626        while !self.state.buffer_X.is_empty() {
627            self = self.process_buffer()?;
628        }
629        Ok(self)
630    }
631
632    /// Check if drift was detected
633    pub fn drift_detected(&self) -> bool {
634        self.state.drift_detected
635    }
636
637    /// Get number of drift events
638    pub fn n_drift_events(&self) -> usize {
639        self.state.n_drift_events
640    }
641
642    /// Get buffer size
643    pub fn buffer_size(&self) -> usize {
644        self.state.buffer_X.len()
645    }
646}
647
648impl Predict<ArrayView2<'_, Float>, Array2<Float>>
649    for StreamingMultiOutput<StreamingMultiOutputTrained>
650{
651    fn predict(&self, X: &ArrayView2<Float>) -> SklResult<Array2<Float>> {
652        let base_wrapper = IncrementalMultiOutputRegression {
653            state: self.state.base_model.clone(),
654            config: IncrementalMultiOutputRegressionConfig::default(),
655        };
656        base_wrapper.predict(X)
657    }
658}
659
660impl Estimator for StreamingMultiOutput<Untrained> {
661    type Config = StreamingMultiOutputConfig;
662    type Error = SklearsError;
663    type Float = Float;
664
665    fn config(&self) -> &Self::Config {
666        &self.config
667    }
668}
669
670impl Estimator for StreamingMultiOutput<StreamingMultiOutputTrained> {
671    type Config = StreamingMultiOutputConfig;
672    type Error = SklearsError;
673    type Float = Float;
674
675    fn config(&self) -> &Self::Config {
676        &self.state.config
677    }
678}
679
680// ============================================================================
681// Tests
682// ============================================================================
683
684#[cfg(test)]
685mod tests {
686    use super::*;
687    // Use SciRS2-Core for arrays and random number generation (SciRS2 Policy)
688    use scirs2_core::ndarray::array;
689
690    #[test]
691    #[allow(non_snake_case)]
692    fn test_incremental_regression_basic() {
693        let X = array![[1.0, 2.0], [2.0, 3.0], [3.0, 4.0]];
694        let y = array![[1.0, 2.0], [2.0, 3.0], [3.0, 4.0]];
695
696        let model = IncrementalMultiOutputRegression::new()
697            .learning_rate(0.1)
698            .alpha(0.0001);
699
700        let trained = model
701            .fit(&X.view(), &y.view())
702            .expect("model fitting should succeed");
703        let predictions = trained
704            .predict(&X.view())
705            .expect("prediction should succeed");
706
707        assert_eq!(predictions.dim(), (3, 2));
708        assert_eq!(trained.n_samples_seen(), 3);
709    }
710
711    #[test]
712    #[allow(non_snake_case)]
713    fn test_incremental_regression_partial_fit() {
714        let X1 = array![[1.0, 2.0], [2.0, 3.0]];
715        let y1 = array![[1.0, 2.0], [2.0, 3.0]];
716
717        let model = IncrementalMultiOutputRegression::new().learning_rate(0.1);
718        let trained = model
719            .fit(&X1.view(), &y1.view())
720            .expect("model fitting should succeed");
721
722        // Partial fit with new data
723        let X2 = array![[3.0, 4.0], [4.0, 5.0]];
724        let y2 = array![[3.0, 4.0], [4.0, 5.0]];
725        let updated = trained
726            .partial_fit(&X2.view(), &y2.view())
727            .expect("operation should succeed");
728
729        assert_eq!(updated.n_samples_seen(), 4);
730
731        let predictions = updated
732            .predict(&X2.view())
733            .expect("prediction should succeed");
734        assert_eq!(predictions.dim(), (2, 2));
735    }
736
737    #[test]
738    #[allow(non_snake_case)]
739    fn test_incremental_regression_learning_rate_decay() {
740        let X = array![[1.0, 2.0], [2.0, 3.0]];
741        let y = array![[1.0, 2.0], [2.0, 3.0]];
742
743        let model = IncrementalMultiOutputRegression::new().learning_rate(0.1);
744        let trained = model
745            .fit(&X.view(), &y.view())
746            .expect("model fitting should succeed");
747
748        let initial_lr = trained.current_learning_rate();
749
750        // Partial fit should decay learning rate
751        let X2 = array![[3.0, 4.0]];
752        let y2 = array![[3.0, 4.0]];
753        let updated = trained
754            .partial_fit(&X2.view(), &y2.view())
755            .expect("operation should succeed");
756
757        assert!(updated.current_learning_rate() < initial_lr);
758    }
759
760    #[test]
761    #[allow(non_snake_case)]
762    fn test_streaming_basic() {
763        let X = array![[1.0, 2.0], [2.0, 3.0], [3.0, 4.0]];
764        let y = array![[1.0, 2.0], [2.0, 3.0], [3.0, 4.0]];
765
766        let model = StreamingMultiOutput::new().batch_size(2).learning_rate(0.1);
767
768        let trained = model
769            .fit(&X.view(), &y.view())
770            .expect("model fitting should succeed");
771        let predictions = trained
772            .predict(&X.view())
773            .expect("prediction should succeed");
774
775        assert_eq!(predictions.dim(), (3, 2));
776    }
777
778    #[test]
779    #[allow(non_snake_case)]
780    fn test_streaming_update() {
781        let X = array![[1.0, 2.0], [2.0, 3.0]];
782        let y = array![[1.0, 2.0], [2.0, 3.0]];
783
784        let model = StreamingMultiOutput::new().batch_size(2);
785        let trained = model
786            .fit(&X.view(), &y.view())
787            .expect("model fitting should succeed");
788
789        // Stream new data
790        let X_stream = array![[3.0, 4.0], [4.0, 5.0]];
791        let y_stream = array![[3.0, 4.0], [4.0, 5.0]];
792        let updated = trained
793            .update_stream(&X_stream.view(), &y_stream.view())
794            .expect("operation should succeed");
795
796        let predictions = updated
797            .predict(&X_stream.view())
798            .expect("prediction should succeed");
799        assert_eq!(predictions.dim(), (2, 2));
800    }
801
802    #[test]
803    #[allow(non_snake_case)]
804    fn test_streaming_buffer() {
805        let X = array![[1.0, 2.0], [2.0, 3.0]];
806        let y = array![[1.0, 2.0], [2.0, 3.0]];
807
808        let model = StreamingMultiOutput::new().batch_size(5); // Large batch size
809        let trained = model
810            .fit(&X.view(), &y.view())
811            .expect("model fitting should succeed");
812
813        // Add small amount of data (should buffer)
814        let X_stream = array![[3.0, 4.0]];
815        let y_stream = array![[3.0, 4.0]];
816        let updated = trained
817            .update_stream(&X_stream.view(), &y_stream.view())
818            .expect("operation should succeed");
819
820        assert_eq!(updated.buffer_size(), 1);
821
822        // Flush buffer
823        let flushed = updated.flush_buffer().expect("operation should succeed");
824        assert_eq!(flushed.buffer_size(), 0);
825    }
826
827    #[test]
828    #[allow(non_snake_case)]
829    fn test_streaming_drift_detection() {
830        let X = array![[1.0, 2.0], [2.0, 3.0], [3.0, 4.0]];
831        let y = array![[1.0, 2.0], [2.0, 3.0], [3.0, 4.0]];
832
833        let model = StreamingMultiOutput::new()
834            .batch_size(2)
835            .detect_drift(true)
836            .learning_rate(0.1);
837
838        let trained = model
839            .fit(&X.view(), &y.view())
840            .expect("model fitting should succeed");
841
842        // The model should track drift events
843        assert_eq!(trained.n_drift_events(), 0);
844    }
845
846    #[test]
847    #[allow(non_snake_case)]
848    fn test_incremental_regression_error_handling() {
849        let X = array![[1.0, 2.0], [2.0, 3.0]];
850        let y = array![[1.0, 2.0], [2.0, 3.0], [3.0, 4.0]]; // Mismatched
851
852        let model = IncrementalMultiOutputRegression::new();
853        assert!(model.fit(&X.view(), &y.view()).is_err());
854    }
855
856    #[test]
857    #[allow(non_snake_case)]
858    fn test_incremental_regression_prediction_error() {
859        let X = array![[1.0, 2.0], [2.0, 3.0]];
860        let y = array![[1.0, 2.0], [2.0, 3.0]];
861
862        let model = IncrementalMultiOutputRegression::new();
863        let trained = model
864            .fit(&X.view(), &y.view())
865            .expect("model fitting should succeed");
866
867        // Wrong number of features
868        let X_test = array![[1.0]];
869        assert!(trained.predict(&X_test.view()).is_err());
870    }
871}
sklears_multioutput/streaming.rs

sklears_multioutput/
streaming.rs
