use super::config::*;
use super::drift_models::{
DecisionTreeDriftDetector, EnsembleDriftDetector, NeuralNetworkDriftDetector,
};
use super::drift_tests::{
AdwinTest, CusumTest, DdmTest, EddmTest, HistogramComparator, HistogramDivergence, KsTest,
LinearModelDetector, MannWhitneyUTest, PageHinkleyTest, WassersteinComparator,
};
use super::optimizer::{Adaptation, AdaptationPriority, AdaptationType, StreamingDataPoint};
use crate::utils::{scalar_or, try_scalar_str};
use scirs2_core::numeric::Float;
use std::collections::{HashMap, VecDeque};
use std::time::{Duration, Instant};
pub struct EnhancedDriftDetector<A: Float + Send + Sync> {
config: DriftConfig,
detection_method: DriftDetectionMethod,
statistical_tests: HashMap<StatisticalMethod, Box<dyn StatisticalTest<A>>>,
distribution_methods: HashMap<DistributionMethod, Box<dyn DistributionComparator<A>>>,
model_detectors: HashMap<ModelType, Box<dyn ModelBasedDetector<A>>>,
detection_history: VecDeque<DriftEvent<A>>,
false_positive_tracker: FalsePositiveTracker<A>,
reference_window: VecDeque<StreamingDataPoint<A>>,
drift_state: DriftState,
last_detection: Option<Instant>,
sensitivity_factor: A,
}
#[derive(Debug, Clone)]
pub struct DriftEvent<A: Float + Send + Sync> {
pub timestamp: Instant,
pub severity: DriftSeverity,
pub confidence: A,
pub detection_method: String,
pub p_value: Option<A>,
pub magnitude: A,
pub affected_features: Vec<usize>,
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum DriftSeverity {
Minor,
Moderate,
Major,
Critical,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DriftState {
Stable,
Warning,
Drift,
Recovery,
}
pub struct FalsePositiveTracker<A: Float + Send + Sync> {
false_positives: VecDeque<Instant>,
true_positives: VecDeque<Instant>,
current_fp_rate: A,
target_fp_rate: A,
}
pub trait StatisticalTest<A: Float + Send + Sync>: Send + Sync {
fn test_for_drift(
&mut self,
reference: &[A],
current: &[A],
) -> Result<DriftTestResult<A>, String>;
fn update_parameters(&mut self, performance_feedback: A) -> Result<(), String>;
fn reset(&mut self);
}
#[derive(Debug, Clone)]
pub struct DriftTestResult<A: Float + Send + Sync> {
pub drift_detected: bool,
pub p_value: A,
pub test_statistic: A,
pub confidence: A,
pub metadata: HashMap<String, A>,
}
pub trait DistributionComparator<A: Float + Send + Sync>: Send + Sync {
fn compare_distributions(
&self,
reference: &[A],
current: &[A],
) -> Result<DistributionComparison<A>, String>;
fn get_threshold(&self) -> A;
fn update_threshold(&mut self, new_threshold: A);
}
#[derive(Debug, Clone)]
pub struct DistributionComparison<A: Float + Send + Sync> {
pub distance: A,
pub threshold: A,
pub drift_detected: bool,
pub confidence: A,
}
pub trait ModelBasedDetector<A: Float + Send + Sync>: Send + Sync {
fn update_model(&mut self, data: &[StreamingDataPoint<A>]) -> Result<(), String>;
fn detect_drift(
&mut self,
data: &[StreamingDataPoint<A>],
) -> Result<ModelDriftResult<A>, String>;
fn reset_model(&mut self) -> Result<(), String>;
}
#[derive(Debug, Clone)]
pub struct ModelDriftResult<A: Float + Send + Sync> {
pub drift_detected: bool,
pub performance_degradation: A,
pub confidence: A,
pub feature_importance_changes: Vec<A>,
}
impl<A: Float + Default + Clone + Send + Sync + std::iter::Sum + 'static> EnhancedDriftDetector<A> {
pub fn new(config: &StreamingConfig) -> Result<Self, String> {
let drift_config = config.drift_config.clone();
let mut statistical_tests: HashMap<StatisticalMethod, Box<dyn StatisticalTest<A>>> =
HashMap::new();
let mut distribution_methods: HashMap<
DistributionMethod,
Box<dyn DistributionComparator<A>>,
> = HashMap::new();
let mut model_detectors: HashMap<ModelType, Box<dyn ModelBasedDetector<A>>> =
HashMap::new();
let sensitivity = drift_config.sensitivity;
let alpha = drift_config.significance_level;
statistical_tests.insert(
StatisticalMethod::ADWIN,
Box::new(AdwinTest::new(sensitivity, alpha)?),
);
statistical_tests.insert(
StatisticalMethod::DDM,
Box::new(DdmTest::new(sensitivity, alpha)?),
);
statistical_tests.insert(
StatisticalMethod::EDDM,
Box::new(EddmTest::new(sensitivity, alpha)?),
);
statistical_tests.insert(
StatisticalMethod::PageHinkley,
Box::new(PageHinkleyTest::new(sensitivity, alpha)?),
);
statistical_tests.insert(
StatisticalMethod::CUSUM,
Box::new(CusumTest::new(sensitivity, alpha)?),
);
statistical_tests.insert(
StatisticalMethod::KolmogorovSmirnov,
Box::new(KsTest::new(sensitivity, alpha)?),
);
statistical_tests.insert(
StatisticalMethod::MannWhitneyU,
Box::new(MannWhitneyUTest::new(sensitivity, alpha)?),
);
distribution_methods.insert(
DistributionMethod::KLDivergence,
Box::new(HistogramComparator::new(
HistogramDivergence::KullbackLeibler,
sensitivity,
)?),
);
distribution_methods.insert(
DistributionMethod::JSDivergence,
Box::new(HistogramComparator::new(
HistogramDivergence::JensenShannon,
sensitivity,
)?),
);
distribution_methods.insert(
DistributionMethod::HellingerDistance,
Box::new(HistogramComparator::new(
HistogramDivergence::Hellinger,
sensitivity,
)?),
);
distribution_methods.insert(
DistributionMethod::WassersteinDistance,
Box::new(WassersteinComparator::new(sensitivity)?),
);
distribution_methods.insert(
DistributionMethod::EarthMoverDistance,
Box::new(WassersteinComparator::new(sensitivity)?),
);
model_detectors.insert(
ModelType::Linear,
Box::new(LinearModelDetector::new(sensitivity)?),
);
model_detectors.insert(
ModelType::NeuralNetwork,
Box::new(NeuralNetworkDriftDetector::new(sensitivity)?),
);
model_detectors.insert(
ModelType::DecisionTree,
Box::new(DecisionTreeDriftDetector::new(sensitivity)?),
);
model_detectors.insert(
ModelType::Ensemble,
Box::new(EnsembleDriftDetector::new(sensitivity)?),
);
let false_positive_tracker = FalsePositiveTracker::new();
Ok(Self {
config: drift_config.clone(),
detection_method: drift_config.detection_method,
statistical_tests,
distribution_methods,
model_detectors,
detection_history: VecDeque::with_capacity(1000),
false_positive_tracker,
reference_window: VecDeque::with_capacity(drift_config.window_size),
drift_state: DriftState::Stable,
last_detection: None,
sensitivity_factor: A::one(),
})
}
pub fn detect_drift(&mut self, batch: &[StreamingDataPoint<A>]) -> Result<bool, String> {
if !self.config.enable_detection || batch.len() < self.config.min_samples {
return Ok(false);
}
self.update_reference_window(batch)?;
if self.reference_window.len() < self.config.window_size / 2 {
return Ok(false);
}
let current_features = self.extract_features(batch)?;
let reference_features = self.extract_reference_features()?;
let detection_method = self.detection_method.clone();
let drift_result = match detection_method {
DriftDetectionMethod::Statistical(method) => {
self.detect_statistical_drift(&method, &reference_features, ¤t_features)?
}
DriftDetectionMethod::Distribution(method) => {
self.detect_distribution_drift(&method, &reference_features, ¤t_features)?
}
DriftDetectionMethod::ModelBased(model_type) => {
self.detect_model_drift(&model_type, batch)?
}
DriftDetectionMethod::Ensemble {
methods,
voting_strategy,
} => self.detect_ensemble_drift(
&methods,
&voting_strategy,
&reference_features,
¤t_features,
batch,
)?,
};
if drift_result.drift_detected {
self.handle_drift_detection(drift_result)?;
Ok(true)
} else {
self.update_drift_state(false);
Ok(false)
}
}
fn update_reference_window(&mut self, batch: &[StreamingDataPoint<A>]) -> Result<(), String> {
for data_point in batch {
if self.reference_window.len() >= self.config.window_size {
self.reference_window.pop_front();
}
self.reference_window.push_back(data_point.clone());
}
Ok(())
}
fn extract_features(&self, batch: &[StreamingDataPoint<A>]) -> Result<Vec<A>, String> {
let mut features = Vec::new();
for data_point in batch {
features.extend(data_point.features.iter().cloned());
}
Ok(features)
}
fn extract_reference_features(&self) -> Result<Vec<A>, String> {
let reference_data: Vec<_> = self
.reference_window
.iter()
.take(self.reference_window.len() / 2)
.collect();
let mut features = Vec::new();
for data_point in reference_data {
features.extend(data_point.features.iter().cloned());
}
Ok(features)
}
fn detect_statistical_drift(
&mut self,
method: &StatisticalMethod,
reference: &[A],
current: &[A],
) -> Result<DriftTestResult<A>, String> {
if let Some(test) = self.statistical_tests.get_mut(method) {
let mut result = test.test_for_drift(reference, current)?;
let alpha = A::from(self.config.significance_level).ok_or_else(|| {
format!(
"significance level {} cannot be represented in the element type",
self.config.significance_level
)
})?;
let effective_alpha = alpha * self.sensitivity_factor;
result.drift_detected = result.drift_detected || result.p_value < effective_alpha;
result.confidence = (result.confidence * self.sensitivity_factor).min(A::one());
Ok(result)
} else {
Err(format!(
"no statistical drift test is registered for {method:?}; \
substituting a different statistic would misreport which \
test produced the verdict"
))
}
}
fn detect_distribution_drift(
&mut self,
method: &DistributionMethod,
reference: &[A],
current: &[A],
) -> Result<DriftTestResult<A>, String> {
if let Some(comparator) = self.distribution_methods.get(method) {
let comparison = comparator.compare_distributions(reference, current)?;
let p_value = (A::one() - comparison.confidence)
.max(A::zero())
.min(A::one());
let alpha = A::from(self.config.significance_level).ok_or_else(|| {
format!(
"significance level {} cannot be represented in the element type",
self.config.significance_level
)
})?;
let mut metadata = HashMap::new();
metadata.insert("distance".to_string(), comparison.distance);
metadata.insert("threshold".to_string(), comparison.threshold);
let result = DriftTestResult {
drift_detected: comparison.drift_detected
|| p_value < alpha * self.sensitivity_factor,
p_value,
test_statistic: comparison.distance,
confidence: (comparison.confidence * self.sensitivity_factor).min(A::one()),
metadata,
};
Ok(result)
} else {
Err(format!(
"no distribution comparator is registered for {method:?}; \
substituting a different divergence would misreport which \
measure produced the verdict"
))
}
}
fn detect_model_drift(
&mut self,
model_type: &ModelType,
batch: &[StreamingDataPoint<A>],
) -> Result<DriftTestResult<A>, String> {
if let Some(detector) = self.model_detectors.get_mut(model_type) {
let model_result = detector.detect_drift(batch)?;
let p_value = (A::one() - model_result.confidence)
.max(A::zero())
.min(A::one());
let alpha = A::from(self.config.significance_level).ok_or_else(|| {
format!(
"significance level {} cannot be represented in the element type",
self.config.significance_level
)
})?;
let mut metadata = HashMap::new();
metadata.insert(
"performance_degradation".to_string(),
model_result.performance_degradation,
);
for (index, change) in model_result.feature_importance_changes.iter().enumerate() {
metadata.insert(format!("weight_delta_{index}"), *change);
}
let result = DriftTestResult {
drift_detected: model_result.drift_detected
|| p_value < alpha * self.sensitivity_factor,
p_value,
test_statistic: model_result.performance_degradation,
confidence: (model_result.confidence * self.sensitivity_factor).min(A::one()),
metadata,
};
Ok(result)
} else {
Err(format!(
"no model-based drift detector is registered for {model_type:?}; \
a feature-mean proxy is not the model-performance signal this \
method is defined over"
))
}
}
fn detect_ensemble_drift(
&mut self,
methods: &[DriftDetectionMethod],
voting_strategy: &VotingStrategy,
reference: &[A],
current: &[A],
batch: &[StreamingDataPoint<A>],
) -> Result<DriftTestResult<A>, String> {
let mut results = Vec::new();
for method in methods {
let result = match method {
DriftDetectionMethod::Statistical(stat_method) => {
self.detect_statistical_drift(stat_method, reference, current)?
}
DriftDetectionMethod::Distribution(dist_method) => {
self.detect_distribution_drift(dist_method, reference, current)?
}
DriftDetectionMethod::ModelBased(model_type) => {
self.detect_model_drift(model_type, batch)?
}
DriftDetectionMethod::Ensemble { .. } => {
continue;
}
};
results.push(result);
}
let ensemble_result = self.apply_voting_strategy(voting_strategy, &results)?;
Ok(ensemble_result)
}
fn apply_voting_strategy(
&self,
strategy: &VotingStrategy,
results: &[DriftTestResult<A>],
) -> Result<DriftTestResult<A>, String> {
if results.is_empty() {
return Err("No results to vote on".to_string());
}
let drift_detected = match strategy {
VotingStrategy::Majority => {
let positive_votes = results.iter().filter(|r| r.drift_detected).count();
positive_votes > results.len() / 2
}
VotingStrategy::Weighted { weights } => {
if weights.len() != results.len() {
return Err("Number of weights doesn't match number of results".to_string());
}
let weighted_score: f64 = results
.iter()
.zip(weights.iter())
.map(|(result, &weight)| weight * if result.drift_detected { 1.0 } else { 0.0 })
.sum();
let total_weight: f64 = weights.iter().sum();
weighted_score / total_weight > 0.5
}
VotingStrategy::Unanimous => results.iter().all(|r| r.drift_detected),
VotingStrategy::Threshold { min_votes } => {
let positive_votes = results.iter().filter(|r| r.drift_detected).count();
positive_votes >= *min_votes
}
};
let count = A::from(results.len()).ok_or_else(|| {
format!(
"result count {} is not representable in the element type",
results.len()
)
})?;
let avg_confidence = results.iter().map(|r| r.confidence).sum::<A>() / count;
let avg_p_value = results.iter().map(|r| r.p_value).sum::<A>() / count;
let avg_test_statistic = results.iter().map(|r| r.test_statistic).sum::<A>() / count;
Ok(DriftTestResult {
drift_detected,
p_value: avg_p_value,
test_statistic: avg_test_statistic,
confidence: avg_confidence,
metadata: HashMap::new(),
})
}
fn handle_drift_detection(&mut self, result: DriftTestResult<A>) -> Result<(), String> {
let severity = self.classify_drift_severity(&result);
let drift_event = DriftEvent {
timestamp: Instant::now(),
severity: severity.clone(),
confidence: result.confidence,
detection_method: format!("{:?}", self.detection_method),
p_value: Some(result.p_value),
magnitude: result.test_statistic,
affected_features: Vec::new(), };
if self.detection_history.len() >= 1000 {
self.detection_history.pop_front();
}
self.detection_history.push_back(drift_event);
self.update_drift_state(true);
self.last_detection = Some(Instant::now());
if self.config.enable_false_positive_tracking {
self.false_positive_tracker.record_detection(true)?;
}
Ok(())
}
#[cfg(test)]
pub(crate) fn unregister_model_detector_for_test(
&mut self,
model_type: &ModelType,
) -> Option<Box<dyn ModelBasedDetector<A>>> {
self.model_detectors.remove(model_type)
}
#[cfg(test)]
pub(crate) fn classify_drift_severity_for_test(
&self,
result: &DriftTestResult<A>,
) -> DriftSeverity {
self.classify_drift_severity(result)
}
fn classify_drift_severity(&self, result: &DriftTestResult<A>) -> DriftSeverity {
let confidence = result.confidence.to_f64().unwrap_or(0.0);
let p_value = result.p_value.to_f64().unwrap_or(1.0);
let by_significance = if p_value < 0.001 && confidence > 0.95 {
DriftSeverity::Critical
} else if p_value < 0.01 && confidence > 0.9 {
DriftSeverity::Major
} else if p_value < 0.05 && confidence > 0.8 {
DriftSeverity::Moderate
} else {
DriftSeverity::Minor
};
let statistic = result.test_statistic.to_f64().unwrap_or(0.0).abs();
let by_magnitude = if statistic >= self.config.drift_threshold {
DriftSeverity::Major
} else if statistic >= self.config.warning_threshold {
DriftSeverity::Moderate
} else {
DriftSeverity::Minor
};
by_significance.max(by_magnitude)
}
fn update_drift_state(&mut self, drift_detected: bool) {
self.drift_state = match (&self.drift_state, drift_detected) {
(DriftState::Stable, true) => DriftState::Warning,
(DriftState::Warning, true) => DriftState::Drift,
(DriftState::Drift, false) => DriftState::Recovery,
(DriftState::Recovery, false) => DriftState::Stable,
(state, _) => state.clone(),
};
}
pub fn compute_sensitivity_adaptation(&mut self) -> Result<Option<Adaptation<A>>, String> {
if self.config.enable_false_positive_tracking {
let current_fp_rate = self.false_positive_tracker.current_fp_rate;
let target_fp_rate = self.false_positive_tracker.target_fp_rate;
let tolerance = scalar_or(0.02, A::zero());
let step = scalar_or(0.1, A::zero());
if (current_fp_rate - target_fp_rate).abs() > tolerance {
let adjustment = if current_fp_rate > target_fp_rate {
-step
} else {
step
};
let adaptation = Adaptation {
adaptation_type: AdaptationType::DriftSensitivity,
magnitude: adjustment,
target_component: "drift_detector".to_string(),
parameters: HashMap::new(),
priority: AdaptationPriority::Normal,
timestamp: Instant::now(),
};
return Ok(Some(adaptation));
}
}
Ok(None)
}
pub fn apply_sensitivity_adaptation(
&mut self,
adaptation: &Adaptation<A>,
) -> Result<(), String> {
if adaptation.adaptation_type == AdaptationType::DriftSensitivity {
self.sensitivity_factor = (self.sensitivity_factor + adaptation.magnitude)
.max(try_scalar_str::<A, _>(0.1)?)
.min(try_scalar_str::<A, _>(2.0)?);
}
Ok(())
}
pub fn is_drift_detected(&self) -> bool {
matches!(self.drift_state, DriftState::Drift | DriftState::Warning)
}
pub fn get_drift_state(&self) -> &DriftState {
&self.drift_state
}
pub fn get_recent_drift_events(&self, count: usize) -> Vec<&DriftEvent<A>> {
self.detection_history.iter().rev().take(count).collect()
}
pub fn reset(&mut self) -> Result<(), String> {
self.detection_history.clear();
self.reference_window.clear();
self.drift_state = DriftState::Stable;
self.last_detection = None;
self.sensitivity_factor = A::one();
for test in self.statistical_tests.values_mut() {
test.reset();
}
for detector in self.model_detectors.values_mut() {
detector.reset_model()?;
}
Ok(())
}
pub fn get_diagnostics(&self) -> DriftDiagnostics {
DriftDiagnostics {
current_state: self.drift_state.clone(),
detection_count: self.detection_history.len(),
false_positive_rate: self
.false_positive_tracker
.current_fp_rate
.to_f64()
.unwrap_or(0.0),
sensitivity_factor: self.sensitivity_factor.to_f64().unwrap_or(1.0),
last_detection_time: self.last_detection,
reference_window_size: self.reference_window.len(),
}
}
}
impl<A: Float + Send + Sync + Send + Sync> FalsePositiveTracker<A> {
fn new() -> Self {
Self {
false_positives: VecDeque::new(),
true_positives: VecDeque::new(),
current_fp_rate: A::zero(),
target_fp_rate: scalar_or(0.05, A::zero()),
}
}
fn record_detection(&mut self, is_true_positive: bool) -> Result<(), String> {
let now = Instant::now();
if is_true_positive {
self.true_positives.push_back(now);
} else {
self.false_positives.push_back(now);
}
let retention = Duration::from_secs(3600);
self.false_positives
.retain(|&time| now.duration_since(time) <= retention);
self.true_positives
.retain(|&time| now.duration_since(time) <= retention);
let total_detections = self.false_positives.len() + self.true_positives.len();
if total_detections > 0 {
self.current_fp_rate = try_scalar_str::<A, _>(self.false_positives.len())?
/ try_scalar_str::<A, _>(total_detections)?;
}
Ok(())
}
}
#[derive(Debug, Clone)]
pub struct DriftDiagnostics {
pub current_state: DriftState,
pub detection_count: usize,
pub false_positive_rate: f64,
pub sensitivity_factor: f64,
pub last_detection_time: Option<Instant>,
pub reference_window_size: usize,
}
#[cfg(test)]
mod drift_detector_regression_tests {
use super::*;
use scirs2_core::ndarray::Array1;
fn detector_with(method: DriftDetectionMethod) -> EnhancedDriftDetector<f64> {
let mut config = StreamingConfig::default();
config.drift_config.detection_method = method;
config.drift_config.min_samples = 10;
config.drift_config.window_size = 200;
EnhancedDriftDetector::new(&config).expect("drift detector")
}
fn wobble(index: usize) -> f64 {
((index as f64) * 0.7548776662).fract() - 0.5
}
fn batch(level: f64, count: usize, offset: usize) -> Vec<StreamingDataPoint<f64>> {
(0..count)
.map(|i| StreamingDataPoint {
features: Array1::from_vec(vec![level + wobble(i + offset)]),
target: Some(Array1::from_vec(vec![level])),
timestamp: Instant::now(),
source_id: None,
quality_score: 1.0,
metadata: HashMap::new(),
})
.collect()
}
#[test]
fn every_statistical_method_is_registered() {
for method in [
StatisticalMethod::ADWIN,
StatisticalMethod::DDM,
StatisticalMethod::EDDM,
StatisticalMethod::PageHinkley,
StatisticalMethod::CUSUM,
StatisticalMethod::KolmogorovSmirnov,
StatisticalMethod::MannWhitneyU,
] {
let mut detector = detector_with(DriftDetectionMethod::Statistical(method.clone()));
detector.detect_drift(&batch(10.0, 120, 0)).expect("warmup");
let result = detector.detect_drift(&batch(40.0, 120, 500));
assert!(
result.is_ok(),
"{method:?} failed on a genuine mean shift: {result:?}"
);
}
}
#[test]
fn every_distribution_method_is_registered() {
for method in [
DistributionMethod::KLDivergence,
DistributionMethod::JSDivergence,
DistributionMethod::HellingerDistance,
DistributionMethod::WassersteinDistance,
DistributionMethod::EarthMoverDistance,
] {
let mut detector = detector_with(DriftDetectionMethod::Distribution(method.clone()));
detector.detect_drift(&batch(10.0, 120, 0)).expect("warmup");
let result = detector.detect_drift(&batch(40.0, 120, 500));
assert!(
result.is_ok(),
"{method:?} failed on a genuine distribution shift: {result:?}"
);
}
}
#[test]
fn every_model_type_is_registered() {
for model_type in [
ModelType::Linear,
ModelType::NeuralNetwork,
ModelType::DecisionTree,
ModelType::Ensemble,
] {
let mut detector = detector_with(DriftDetectionMethod::ModelBased(model_type.clone()));
detector
.detect_drift(&batch(10.0, 120, 0))
.unwrap_or_else(|error| panic!("{model_type:?} warmup failed: {error}"));
let result = detector.detect_drift(&batch(40.0, 120, 500));
assert!(
result.is_ok(),
"{model_type:?} failed on a genuine mean shift: {result:?}"
);
}
}
#[test]
fn model_type_without_a_registered_detector_is_an_honest_error() {
let mut detector = detector_with(DriftDetectionMethod::ModelBased(ModelType::Linear));
detector
.unregister_model_detector_for_test(&ModelType::Linear)
.expect("Linear starts out registered");
detector.detect_drift(&batch(10.0, 120, 0)).ok();
let result = detector.detect_drift(&batch(40.0, 120, 500));
assert!(
result.is_err(),
"an unregistered model type must report an error instead of a \
feature-mean proxy dressed up as a model-drift verdict"
);
}
#[test]
fn implemented_model_types_fire_on_a_target_shift() {
for model_type in [
ModelType::NeuralNetwork,
ModelType::DecisionTree,
ModelType::Ensemble,
] {
let mut detector = detector_with(DriftDetectionMethod::ModelBased(model_type.clone()));
for round in 0..6 {
detector
.detect_drift(&batch(10.0, 60, round * 60))
.unwrap_or_else(|error| panic!("{model_type:?} warmup failed: {error}"));
}
let mut fired = false;
for round in 0..6 {
if detector
.detect_drift(&batch(400.0, 60, 5_000 + round * 60))
.unwrap_or_else(|error| panic!("{model_type:?} shift failed: {error}"))
{
fired = true;
break;
}
}
assert!(
fired,
"{model_type:?} did not report drift after a 390-unit shift in the \
target relationship"
);
}
}
#[test]
fn linear_model_drift_detection_works_end_to_end() {
let mut detector = detector_with(DriftDetectionMethod::ModelBased(ModelType::Linear));
for round in 0..6 {
detector
.detect_drift(&batch(10.0, 60, round * 60))
.expect("stable rounds must not error");
}
let diagnostics = detector.get_diagnostics();
assert!(
diagnostics.reference_window_size > 0,
"the reference window must retain real observations"
);
}
#[test]
fn stationary_stream_does_not_raise_drift() {
let mut detector = detector_with(DriftDetectionMethod::Statistical(
StatisticalMethod::KolmogorovSmirnov,
));
let mut fired = 0usize;
for round in 0..12 {
if detector
.detect_drift(&batch(10.0, 60, round * 60))
.expect("detect_drift")
{
fired += 1;
}
}
assert_eq!(
fired, 0,
"a stationary stream raised {fired} drift events out of 12 rounds"
);
assert_eq!(detector.get_drift_state(), &DriftState::Stable);
}
#[test]
fn recorded_drift_events_carry_real_p_values() {
let mut detector = detector_with(DriftDetectionMethod::Statistical(
StatisticalMethod::KolmogorovSmirnov,
));
detector.detect_drift(&batch(10.0, 120, 0)).expect("warmup");
let fired = detector
.detect_drift(&batch(100.0, 120, 500))
.expect("detect_drift");
assert!(fired, "a 90-unit mean shift must be detected");
let events = detector.get_recent_drift_events(1);
let event = events.first().expect("an event must be recorded");
let p_value = event.p_value.expect("a p-value must be recorded");
for fabricated in [0.01_f64, 0.015, 0.02, 0.5, 0.6, 0.7] {
assert!(
(p_value - fabricated).abs() > 1e-12,
"D1 regression: p-value {p_value} matches the hard-coded literal \
{fabricated}"
);
}
assert!(
(0.0..=1.0).contains(&p_value),
"a p-value must lie in [0, 1], got {p_value}"
);
assert!(
event.magnitude > 0.0,
"the recorded magnitude must be the real test statistic"
);
}
#[test]
fn ensemble_of_distinct_detectors_agrees_on_an_unmistakable_shift() {
let mut detector = detector_with(DriftDetectionMethod::Ensemble {
methods: vec![
DriftDetectionMethod::Statistical(StatisticalMethod::KolmogorovSmirnov),
DriftDetectionMethod::Statistical(StatisticalMethod::MannWhitneyU),
DriftDetectionMethod::Distribution(DistributionMethod::JSDivergence),
],
voting_strategy: VotingStrategy::Majority,
});
detector.detect_drift(&batch(10.0, 120, 0)).expect("warmup");
let fired = detector
.detect_drift(&batch(500.0, 120, 900))
.expect("detect_drift");
assert!(
fired,
"a 490-unit mean shift must be detected by a majority of three real \
detectors"
);
}
}