foxhunt/crates/ml-labeling/src/fractional_diff.rs

//! Fractional differentiation for stationarity with memory preservation
//!
//! Implements streaming fractional differentiation with <1us latency target.
//! Based on the fractional differentiation concepts from financial machine learning.

use std::collections::VecDeque;
use std::time::Instant;

use super::gpu_acceleration::LabelingError;
use super::types::{FractionalDiffConfig, FractionalDiffResult};

/// Fractional differentiation coefficients calculator
#[derive(Debug, Clone)]
pub struct FractionalCoeffs {
    coeffs: Vec<f64>,
}

impl FractionalCoeffs {
    /// Create new fractional coefficients
    pub fn new(diff_order: f64, max_lags: usize, threshold: f64) -> Self {
        let mut coeffs = Vec::with_capacity(max_lags);

        // Calculate binomial coefficients for fractional differentiation
        coeffs.push(1.0); // First coefficient is always 1

        for k in 1..max_lags {
            let coeff = coeffs[k - 1] * (k as f64 - diff_order - 1.0) / k as f64;

            if coeff.abs() < threshold {
                break;
            }

            coeffs.push(coeff);
        }

        Self {
            coeffs,
        }
    }

    /// Get coefficient at index
    pub fn get(&self, index: usize) -> f64 {
        if index < self.coeffs.len() {
            self.coeffs[index]
        } else {
            0.0
        }
    }

    /// Get number of coefficients
    pub fn len(&self) -> usize {
        self.coeffs.len()
    }

    /// Check if empty
    pub fn is_empty(&self) -> bool {
        self.coeffs.is_empty()
    }
}

/// Streaming fractional differentiator
#[derive(Debug, Clone)]
pub struct StreamingDifferentiator {
    config: FractionalDiffConfig,
    coeffs: FractionalCoeffs,
    window: VecDeque<f64>,
    processed_count: u64,
}

impl StreamingDifferentiator {
    /// Create new streaming differentiator
    pub fn new(config: FractionalDiffConfig) -> Result<Self, LabelingError> {
        let coeffs = FractionalCoeffs::new(config.diff_order, config.max_lags, config.threshold);
        let max_lags = config.max_lags;

        Ok(Self {
            config,
            coeffs,
            window: VecDeque::with_capacity(max_lags),
            processed_count: 0,
        })
    }

    /// Process new value and return fractionally differenced result
    pub fn process(
        &mut self,
        value: i64,
        timestamp_ns: u64,
    ) -> Result<FractionalDiffResult, LabelingError> {
        let start = Instant::now();

        // Convert to f64 for processing
        let value_f64 = value as f64;

        // Add to window
        self.window.push_back(value_f64);
        if self.window.len() > self.config.max_lags {
            self.window.pop_front();
        }

        // Calculate fractional difference
        let mut diff_value = 0.0;

        for (i, coeff) in self.coeffs.coeffs.iter().enumerate() {
            if i >= self.window.len() {
                break;
            }

            let window_index = self.window.len() - 1 - i;
            diff_value += coeff * self.window[window_index];
        }

        let processing_latency_us = start.elapsed().as_micros() as u32;
        self.processed_count += 1;

        Ok(FractionalDiffResult {
            timestamp_ns,
            original_value: value,
            diff_value: (diff_value * 10000.0) as i64, // Scale to fixed point
            diff_order: self.config.diff_order,
            window_size: self.window.len(),
            processing_latency_us,
        })
    }

    /// Reset the differentiator
    pub fn reset(&mut self) {
        self.window.clear();
        self.processed_count = 0;
    }

    /// Get current window size
    pub fn window_size(&self) -> usize {
        self.window.len()
    }

    /// Get processed count
    pub const fn processed_count(&self) -> u64 {
        self.processed_count
    }

    /// Check if ready (has enough data)
    pub fn is_ready(&self) -> bool {
        self.window.len() >= self.config.min_window_size
    }
}

/// General fractional differentiator (batch processing)
#[derive(Debug, Clone)]
pub struct FractionalDifferentiator {
    config: FractionalDiffConfig,
    coeffs: FractionalCoeffs,
}

impl FractionalDifferentiator {
    /// Create new fractional differentiator
    pub fn new(config: FractionalDiffConfig) -> Result<Self, LabelingError> {
        let coeffs = FractionalCoeffs::new(config.diff_order, config.max_lags, config.threshold);

        Ok(Self { config, coeffs })
    }

    /// Process batch of values
    pub fn process_batch(
        &self,
        values: &[i64],
    ) -> Result<Vec<FractionalDiffResult>, LabelingError> {
        if values.is_empty() {
            return Ok(Vec::new());
        }

        let start = Instant::now();
        let mut results = Vec::with_capacity(values.len());

        // Convert to f64 for processing
        let values_f64: Vec<f64> = values.iter().map(|&v| v as f64).collect();

        for i in 0..values.len() {
            let mut diff_value = 0.0;

            // Calculate fractional difference for current position
            for (k, coeff) in self.coeffs.coeffs.iter().enumerate() {
                if k > i {
                    break;
                }

                diff_value += coeff * values_f64[i - k];
            }

            let result = FractionalDiffResult {
                timestamp_ns: i as u64 * 1_000_000_000, // Mock timestamps
                original_value: values[i],
                diff_value: (diff_value * 10000.0) as i64, // Scale to fixed point
                diff_order: self.config.diff_order,
                window_size: (i + 1).min(self.coeffs.len()),
                processing_latency_us: 0, // Will be set below
            };

            results.push(result);
        }

        // Set processing latency for all results
        let total_latency_us = start.elapsed().as_micros() as u32;
        let avg_latency_us = total_latency_us / values.len() as u32;

        for result in &mut results {
            result.processing_latency_us = avg_latency_us;
        }

        Ok(results)
    }

    /// Process single value with history
    pub fn process_with_history(
        &self,
        values: &[i64],
        target_index: usize,
    ) -> Result<FractionalDiffResult, LabelingError> {
        if target_index >= values.len() {
            return Err(LabelingError::InvalidInput(
                "Target index out of bounds".to_owned(),
            ));
        }

        let start = Instant::now();
        let values_f64: Vec<f64> = values.iter().map(|&v| v as f64).collect();

        let mut diff_value = 0.0;

        // Calculate fractional difference
        for (k, coeff) in self.coeffs.coeffs.iter().enumerate() {
            if k > target_index {
                break;
            }

            diff_value += coeff * values_f64[target_index - k];
        }

        let processing_latency_us = start.elapsed().as_micros() as u32;

        Ok(FractionalDiffResult {
            timestamp_ns: target_index as u64 * 1_000_000_000, // Mock timestamp
            original_value: values[target_index],
            diff_value: (diff_value * 10000.0) as i64,
            diff_order: self.config.diff_order,
            window_size: (target_index + 1).min(self.coeffs.len()),
            processing_latency_us,
        })
    }

    /// Get configuration
    pub const fn get_config(&self) -> &FractionalDiffConfig {
        &self.config
    }

    /// Get coefficients
    pub const fn get_coeffs(&self) -> &FractionalCoeffs {
        &self.coeffs
    }
}

#[cfg(test)]
#[allow(
    clippy::inconsistent_digit_grouping,
    clippy::unnecessary_wraps,
    clippy::doc_markdown,
    clippy::let_underscore_must_use,
    clippy::assertions_on_result_states
)]
mod tests {
    use super::*;
    use crate::constants::MAX_FRACTIONAL_DIFF_LATENCY_US;

    #[test]
    fn test_fractional_coeffs() -> Result<(), Box<dyn std::error::Error>> {
        let coeffs = FractionalCoeffs::new(0.5, 10, 1e-6);

        // First coefficient should be 1.0
        assert!((coeffs.get(0) - 1.0).abs() < 1e-10);

        // Coefficients should decay
        assert!(coeffs.get(1).abs() < coeffs.get(0).abs());
        assert!(coeffs.get(2).abs() < coeffs.get(1).abs());

        Ok(())
    }

    #[test]
    #[ignore] // Flaky: 1us latency target is too tight for reliable CI
    fn test_streaming_differentiator() -> Result<(), LabelingError> {
        let config = FractionalDiffConfig::standard();
        let mut differentiator = StreamingDifferentiator::new(config)?;

        // Process some test values
        let test_values = [100000, 101000, 99000, 102000, 98000]; // Price-like values
        let mut results = Vec::new();

        for (i, value) in test_values.into_iter().enumerate() {
            let timestamp_ns = 1692000000_000_000_000 + i as u64 * 1_000_000_000;
            let result = differentiator.process(value, timestamp_ns)?;

            results.push(result);
        }

        // Should have results for all inputs
        assert_eq!(results.len(), test_values.len());

        // Results should have reasonable diff values (scaled by 10000x at line 123)
        // With price values ~100,000 and 10000x scaling, values can reach 1B
        for result in &results {
            assert!(result.diff_value.abs() < 2_000_000_000); // Bound for 100K prices * 10000x scaling
        }

        Ok(())
    }

    #[test]
    fn test_batch_differentiator() -> Result<(), LabelingError> {
        let config = FractionalDiffConfig::standard();
        let expected_diff_order = config.diff_order;
        let differentiator = FractionalDifferentiator::new(config)?;

        let test_values = vec![100000, 101000, 99000, 102000, 98000, 97000, 103000];
        let results = differentiator.process_batch(&test_values)?;

        assert_eq!(results.len(), test_values.len());

        for result in &results {
            assert_eq!(result.diff_order, expected_diff_order);
        }

        Ok(())
    }

    #[test]
    #[ignore = "Performance benchmark: 1us latency target too strict for CI. \
                Run manually with: cargo test -p ml-labeling test_differentiator_with_history -- --ignored"]
    /// Performance benchmark for fractional differentiation with history
    /// Target: <=1us processing latency (MAX_FRACTIONAL_DIFF_LATENCY_US)
    fn test_differentiator_with_history() -> Result<(), LabelingError> {
        let config = FractionalDiffConfig::standard();
        let differentiator = FractionalDifferentiator::new(config)?;

        let test_values = vec![100000, 101000, 99000, 102000, 98000];
        let result = differentiator.process_with_history(&test_values, 4)?;

        assert_eq!(result.original_value, 98000);
        assert!(result.processing_latency_us as u64 <= MAX_FRACTIONAL_DIFF_LATENCY_US);
        assert_eq!(result.window_size, 5);

        Ok(())
    }

    #[test]
    fn test_streaming_differentiator_reset() -> Result<(), LabelingError> {
        let config = FractionalDiffConfig::standard();
        let mut differentiator = StreamingDifferentiator::new(config)?;

        // Process some values
        for i in 0..5 {
            let _ = differentiator.process(100000 + i as i64 * 1000, i as u64 * 1_000_000_000);
        }

        assert_eq!(differentiator.window_size(), 5);
        assert_eq!(differentiator.processed_count(), 5);

        // Reset
        differentiator.reset();
        assert_eq!(differentiator.window_size(), 0);
        assert_eq!(differentiator.processed_count(), 0);

        Ok(())
    }

    #[test]
    fn test_coefficients_calculation() -> Result<(), Box<dyn std::error::Error>> {
        // Test different fractional orders
        let coeffs_half = FractionalCoeffs::new(0.5, 10, 1e-6);
        let coeffs_quarter = FractionalCoeffs::new(0.25, 10, 1e-6);

        // Higher fractional order should have different coefficient patterns
        assert_ne!(coeffs_half.get(1), coeffs_quarter.get(1));

        // Both should start with 1.0
        assert!((coeffs_half.get(0) - 1.0).abs() < 1e-10);
        assert!((coeffs_quarter.get(0) - 1.0).abs() < 1e-10);

        Ok(())
    }

    #[test]
    fn test_streaming_readiness() -> Result<(), LabelingError> {
        let config = FractionalDiffConfig {
            diff_order: 0.5,
            max_lags: 10,
            min_window_size: 3,
            threshold: 1e-6,
        };

        let mut differentiator = StreamingDifferentiator::new(config)?;

        assert!(!differentiator.is_ready());

        // Process values until ready
        let _ = differentiator.process(100000, 1000);
        assert!(!differentiator.is_ready());

        let _ = differentiator.process(101000, 2000);
        assert!(!differentiator.is_ready());

        let _ = differentiator.process(99000, 3000);
        assert!(differentiator.is_ready());

        Ok(())
    }

    #[test]
    fn test_error_handling() -> Result<(), Box<dyn std::error::Error>> {
        let config = FractionalDiffConfig::standard();
        let differentiator = FractionalDifferentiator::new(config)?;

        // Test out of bounds
        let test_values = vec![100000, 101000];
        let result = differentiator.process_with_history(&test_values, 5);
        assert!(result.is_err());

        Ok(())
    }
}