Time-Series Data Analyzer with Sliding Windows

Problem Statement

Build a time-series data analyzer that computes statistics over sliding windows of data. The analyzer processes sensor readings, financial data, or metrics streams and computes aggregates (moving averages, min/max, standard deviation) using efficient windowing algorithms with zero-copy slicing.

Your analyzer should support:

• Multiple window sizes (e.g., 10-second, 1-minute, 5-minute windows)
• Computing statistics: moving average, min, max, median, percentiles
• Handling streaming data (process data as it arrives)
• Using efficient algorithms: O(n) sliding window, not O(n*w)
• Providing zero-copy views into data windows
• Detecting anomalies (values outside expected range)

Example workflow:

Input: Sensor readings stream (temperature, every 1 second)
Windows: [10s, 60s, 300s]
Operations: Compute average, min, max, std_dev per window
Anomaly detection: Flag readings > 3 std_dev from mean
Output: Real-time statistics and anomaly alerts

Key Concepts Explained

1. VecDeque (Double-Ended Queue)

VecDeque is a ring buffer that allows efficient push/pop from both ends (O(1)).

Vec limitations (no efficient front operations):

#![allow(unused)]
fn main() {
let mut vec = Vec::new();
vec.push(1);           // O(1) - push back
vec.pop();             // O(1) - pop back
vec.insert(0, 1);      // O(n) - insert front (shift all elements!)
vec.remove(0);         // O(n) - remove front (shift all elements!)
}

VecDeque efficiency (O(1) both ends):

#![allow(unused)]
fn main() {
let mut deque = VecDeque::new();
deque.push_back(1);    // O(1) - add to back
deque.push_front(1);   // O(1) - add to front
deque.pop_back();      // O(1) - remove from back
deque.pop_front();     // O(1) - remove from front
}

How VecDeque works (ring buffer):

Conceptual array:  [_, _, A, B, C, _, _]
                          ↑       ↑
                        head    tail

push_back(D):      [_, _, A, B, C, D, _]  // Increment tail
pop_front():       [_, _, _, B, C, D, _]  // Increment head

When tail reaches end, it wraps around:
                   [E, F, _, B, C, D, _]
                       ↑           ↑
                     tail        head

Sliding window with VecDeque:

#![allow(unused)]
fn main() {
let mut window = VecDeque::with_capacity(3);

window.push_back(1);  // [1]
window.push_back(2);  // [1, 2]
window.push_back(3);  // [1, 2, 3] - full!

// Maintain fixed size: pop front, push back
window.pop_front();   // [2, 3]
window.push_back(4);  // [2, 3, 4]  // Sliding window!
}

Performance comparison:

When to use VecDeque:

• ✅ Sliding windows
• ✅ Queues (FIFO)
• ✅ Need efficient front operations
• ❌ Only need back operations (use Vec)
• ❌ Random access (Vec has better cache locality)

2. Incremental Statistics (Online Algorithms)

Incremental statistics update in O(1) instead of recalculating from scratch.

Naive approach (recalculate every time):

#![allow(unused)]
fn main() {
fn average(window: &[f64]) -> f64 {
    window.iter().sum::<f64>() / window.len() as f64
}

// For window size 1000:
// Each call: 1000 additions + 1 division = O(n)
// 1M calls: 1 billion operations
}

Incremental approach (running sum):

#![allow(unused)]
fn main() {
struct IncrementalWindow {
    window: VecDeque<f64>,
    running_sum: f64,  // Maintained incrementally
}

fn push(&mut self, value: f64) {
    if self.window.len() == self.capacity {
        let evicted = self.window.pop_front().unwrap();
        self.running_sum -= evicted;  // Subtract old value
    }
    self.window.push_back(value);
    self.running_sum += value;  // Add new value
}

fn average(&self) -> f64 {
    self.running_sum / self.window.len() as f64  // O(1)!
}

// For window size 1000:
// Each call: 2 additions + 1 division = O(1)
// 1M calls: 2 million operations (500× faster!)
}

Variance with sum of squares:

Variance formula: Var(X) = E[X²] - E[X]²

Maintain two running sums:
- running_sum:    Σxᵢ
- running_sum_sq: Σxᵢ²

Variance = (running_sum_sq / n) - (running_sum / n)²

Implementation:

#![allow(unused)]
fn main() {
fn push(&mut self, value: f64) {
    if let Some(evicted) = self.window.pop_front() {
        self.running_sum -= evicted;
        self.running_sum_sq -= evicted * evicted;
    }
    self.window.push_back(value);
    self.running_sum += value;
    self.running_sum_sq += value * value;
}

fn variance(&self) -> f64 {
    let mean = self.running_sum / self.len as f64;
    let mean_of_squares = self.running_sum_sq / self.len as f64;
    mean_of_squares - (mean * mean)  // O(1)!
}
}

Performance impact (window size 1000, 1M updates):

3. Monotonic Deque (Sliding Window Min/Max)

Monotonic deque maintains min/max in O(1) amortized time by keeping only potentially useful elements.

Naive min (scan entire window):

#![allow(unused)]
fn main() {
fn min(&self) -> f64 {
    *self.window.iter().min().unwrap()  // O(n) scan
}

// Window size 1000: 1000 comparisons per call
// 1M calls: 1 billion comparisons
}

Monotonic deque min (O(1) amortized):

#![allow(unused)]
fn main() {
// Invariant: min_deque is increasing (smallest at front)
struct MinMaxWindow {
    window: VecDeque<(usize, f64)>,      // (index, value)
    min_deque: VecDeque<(usize, f64)>,   // Monotonic increasing
}

fn push(&mut self, value: f64) {
    // Remove elements from back that are >= new value
    // (they can never be minimum while new value is in window)
    while let Some(&(_, back_val)) = self.min_deque.back() {
        if back_val >= value {
            self.min_deque.pop_back();
        } else {
            break;
        }
    }

    // Add new element
    self.min_deque.push_back((self.index, value));

    // Remove elements that fell out of window
    while let Some(&(idx, _)) = self.min_deque.front() {
        if idx <= self.index - self.capacity {
            self.min_deque.pop_front();
        } else {
            break;
        }
    }

    self.index += 1;
}

fn min(&self) -> f64 {
    self.min_deque.front().unwrap().1  // O(1)!
}
}

Visual example (window size 3):

Push sequence: [3, 1, 4, 2]

Push 3:
window:     [3]
min_deque:  [(0,3)]

Push 1:
window:     [3, 1]
min_deque:  [(1,1)]      // Popped (0,3) because 3 > 1

Push 4:
window:     [3, 1, 4]
min_deque:  [(1,1), (2,4)]  // Keep 4 (might be min after 1 leaves)

Push 2:
window:     [1, 4, 2]    // Evicted 3
min_deque:  [(1,1), (3,2)]  // Popped (2,4) because 4 > 2

Why it’s O(1) amortized:

• Each element pushed once: n push operations
• Each element popped at most once: ≤n pop operations
• Total: 2n operations for n elements
• Amortized: 2n/n = O(1) per operation

Performance comparison (window size 1000, 1M updates):

4. Quickselect Algorithm (O(n) k-th Element)

Quickselect finds k-th smallest element in O(n) average time without full sort.

Sorting approach (O(n log n)):

#![allow(unused)]
fn main() {
fn median(window: &mut [f64]) -> f64 {
    window.sort_by(|a, b| a.partial_cmp(b).unwrap());  // O(n log n)
    window[window.len() / 2]
}

// For n=1000: ~10,000 operations
}

Quickselect approach (O(n) average):

#![allow(unused)]
fn main() {
fn median(window: &mut [f64]) -> f64 {
    let mid = window.len() / 2;
    window.select_nth_unstable_by(mid, |a, b| a.partial_cmp(b).unwrap());
    window[mid]  // Median is now at position mid
}

// For n=1000: ~1,000 operations (10× faster!)
}

How quickselect works:

Find median of [3, 1, 4, 1, 5, 9, 2, 6]

1. Pick pivot (e.g., 4)
2. Partition: [3, 1, 1, 2] | 4 | [5, 9, 6]
              ↑              ↑
           left (4 elements) pivot at index 4

3. Median is at index 4
   Median is in left partition
   Recursively search left: [3, 1, 1, 2]

4. Pick pivot (e.g., 2)
   Partition: [1, 1] | 2 | [3]

5. Median is at index 1 (within this subarray)
   Found: 1

Complexity:

• Average: O(n) - halves search space each time
• Worst case: O(n²) - rare with good pivot selection
• Space: O(1) - in-place partitioning

Performance comparison (1000 elements, 100 iterations):

5. Single-Pass Multi-Window Processing

Single-pass processing updates multiple windows with one iteration over data.

Naive approach (multiple passes):

#![allow(unused)]
fn main() {
let windows = vec![10, 60, 300];  // 3 windows

for &size in &windows {
    let mut window = Window::new(size);
    for &value in &data {  // Pass 1 over data
        window.push(value);
    }
    process(window);
}

for &size in &windows {
    let mut window = Window::new(size);
    for &value in &data {  // Pass 2 over data
        window.push(value);
    }
    process(window);
}
// Total: 3 passes over data
}

Single-pass approach:

#![allow(unused)]
fn main() {
let mut windows: Vec<Window> = vec![
    Window::new(10),
    Window::new(60),
    Window::new(300),
];

for &value in &data {  // Single pass over data
    for window in &mut windows {
        window.push(value);
    }
}
// Total: 1 pass over data (3× faster for data access)
}

Why it matters:

• Cache locality: Data loaded once from memory/disk
• I/O efficiency: Read data stream once
• Real-time: Process live streams without buffering
• Parallelism: Can process windows independently after ingestion

Performance impact:

6. Z-Score (Statistical Anomaly Detection)

Z-score measures how many standard deviations a value is from the mean.

Formula:

z = (x - μ) / σ

where:
- x: observed value
- μ: mean
- σ: standard deviation

Interpretation:

|z| < 1:  Within 1 std dev (68% of data in normal distribution)
|z| < 2:  Within 2 std dev (95% of data)
|z| < 3:  Within 3 std dev (99.7% of data)
|z| ≥ 3:  Outlier! (0.3% probability)

Implementation:

#![allow(unused)]
fn main() {
fn is_anomaly(&self, value: f64) -> bool {
    let mean = self.average().unwrap_or(0.0);
    let std_dev = self.std_dev().unwrap_or(1.0);

    let z_score = (value - mean) / std_dev;
    z_score.abs() > 3.0  // 3 sigma rule
}
}

Example (temperature monitoring):

Historical data: mean = 22°C, std_dev = 2°C

Reading: 28°C
z = (28 - 22) / 2 = 3.0  // At threshold

Reading: 30°C
z = (30 - 22) / 2 = 4.0  // ANOMALY! (|z| > 3)

Threshold tradeoffs:

Real-world applications:

• Server response time monitoring (detect slowdowns)
• Network traffic analysis (detect attacks)
• Sensor fault detection (detect malfunctions)
• Financial fraud detection (detect unusual transactions)

7. Ring Buffer Memory Layout

Ring buffer reuses memory by wrapping indices, avoiding allocations.

Linear buffer (reallocate on growth):

Initial:     [A, B, _, _]
Push C:      [A, B, C, _]
Push D:      [A, B, C, D]  (full)
Push E:      [B, C, D, E]  (shift left - O(n)!)
             ↑ Expensive shift

Ring buffer (wrap around):

Capacity 4:  [_, _, _, _]
             ↑
           head=0, tail=0

Push A,B,C:  [A, B, C, _]
             ↑        ↑
           head      tail

Push D:      [A, B, C, D]
             ↑           ↑
           head         tail (wraps to 0)

Push E:      [E, B, C, D]  (overwrote A at index 0)
                ↑        ↑
              head       tail
           (head wraps to 1)

No shifts! Just update indices modulo capacity.

Index math:

#![allow(unused)]
fn main() {
let physical_index = (logical_index + head) % capacity;

// push_back:
buffer[tail % capacity] = value;
tail += 1;

// pop_front:
let value = buffer[head % capacity];
head += 1;
}

Memory efficiency:

VecDeque uses ring buffer: Efficient sliding windows without reallocations.

8. Amortized Complexity Analysis

Amortized analysis computes average cost over sequence of operations.

Example: Monotonic deque min

Worst-case analysis (pessimistic):

Each push might pop all elements from deque
Worst case: O(n) per push

Amortized analysis (realistic):

Consider n pushes total:
- Each element pushed once:     n operations
- Each element popped at most once: ≤n operations
- Total: ≤2n operations
- Amortized per push: 2n/n = O(1)

Accounting method:

Assign credits to operations:
- Push: costs 1, but we pay 2 (1 for push, 1 saved for future pop)
- Pop: costs 1, use saved credit (free!)

Total credits: 2n for n pushes
All pops covered by saved credits
Amortized cost: O(1) per operation

Potential method:

Potential Φ = size of deque

Push with evictions:
- Actual cost: 1 + k (push + k pops)
- Potential change: ΔΦ = 1 - k (size increases by 1, decreases by k)
- Amortized cost: 1 + k + (1 - k) = 2 = O(1)

Why it matters: Guarantees O(1) performance in practice even though individual operations can be O(n).

9. Zero-Copy Slicing

Zero-copy slicing provides views into data without copying.

Copy approach (expensive):

#![allow(unused)]
fn main() {
fn get_window_copy(&self) -> Vec<f64> {
    self.window.iter().copied().collect()  // Copies all elements
}

// For 1000-element window: 1000 copies
// Call 1M times: 1 billion copies!
}

Zero-copy approach:

#![allow(unused)]
fn main() {
fn as_slice(&self) -> &[f64] {
    let (slice1, slice2) = self.window.as_slices();
    if slice2.is_empty() {
        slice1  // Contiguous case
    } else {
        // Non-contiguous: need to make contiguous first
        self.window.make_contiguous()
    }
}

// Returns reference: 0 copies
// Call 1M times: 0 copies!
}

VecDeque slicing (handles ring buffer complexity):

#![allow(unused)]
fn main() {
// VecDeque may wrap around:
//   [C, D, _, A, B]
//         ↑  ↑
//        tail head

// as_slices returns two slices:
// slice1: &[A, B]  (from head to end)
// slice2: &[C, D]  (from start to tail)

// make_contiguous shuffles to:
//   [A, B, C, D, _]
// Returns single slice: &[A, B, C, D]
}

Performance comparison (1000-element window, 1M accesses):

10. Numerical Stability (Avoiding Precision Loss)

Numerical stability prevents floating-point errors from accumulating.

Naive variance (unstable):

#![allow(unused)]
fn main() {
// Var(X) = E[X²] - E[X]²
let mean = running_sum / n;
let mean_sq = running_sum_sq / n;
let variance = mean_sq - (mean * mean);  // Catastrophic cancellation!

// Problem: If mean ≈ values, mean_sq ≈ mean²
// Subtraction loses precision (e.g., 123456.789 - 123456.788 = 0.001)
}

Example of catastrophic cancellation:

Values: [1000000.1, 1000000.2, 1000000.3]

mean = 1000000.2
mean² = 1000000400000.04
mean of squares = 1000000400000.14

Variance = 1000000400000.14 - 1000000400000.04 = 0.10

But with float precision:
1000000400000.14 → 1.000000400000140e12
1000000400000.04 → 1.000000400000040e12
Subtraction: might round to 0.00 (precision loss!)

Welford’s algorithm (stable):

#![allow(unused)]
fn main() {
// Update mean and variance incrementally
fn push(&mut self, value: f64) {
    self.count += 1;
    let delta = value - self.mean;
    self.mean += delta / self.count as f64;
    let delta2 = value - self.mean;
    self.m2 += delta * delta2;
}

fn variance(&self) -> f64 {
    self.m2 / self.count as f64
}

// No catastrophic cancellation!
// Numerically stable for any range of values
}

Why Welford’s algorithm works:

• Computes deviations from running mean
• Avoids subtracting large similar numbers
• Updates variance incrementally
• More complex but numerically stable

When to use:

• ✅ Large values with small variance (e.g., 1000000 ± 1)
• ✅ High-precision requirements
• ✅ Long-running statistics (errors accumulate)
• ❌ Small values (simple formula is fine)
• ❌ Low precision requirements

Connection to This Project

This project demonstrates efficient time-series analysis through progressive algorithmic optimizations and clever data structures.

Milestone 1: Basic Sliding Window with VecDeque

Concepts applied:

• VecDeque for efficient FIFO operations
• Ring buffer for fixed-size windows
• Zero-copy slicing with as_slice()
• Basic statistics (mean, min, max)

Why it matters: VecDeque enables efficient sliding windows:

• O(1) push_back and pop_front (vs O(n) with Vec)
• Ring buffer reuses memory (no reallocations)
• Fixed memory footprint (O(window_size))

Real-world impact:

#![allow(unused)]
fn main() {
// Vec approach (O(n) per slide)
let mut vec = Vec::new();
vec.push(new_value);     // O(1)
vec.remove(0);           // O(n) - shifts all elements!

// VecDeque approach (O(1) per slide)
let mut deque = VecDeque::new();
deque.push_back(new_value);  // O(1)
deque.pop_front();           // O(1) - no shifting!
}

Performance comparison (window size 1000, 1M slides):

Real-world validation: All production time-series libraries use VecDeque or equivalent ring buffers.

Milestone 2: Incremental Statistics (Avoid Re-Scanning)

Concepts applied:

• Incremental algorithms (online algorithms)
• Running sum for O(1) mean
• Running sum of squares for O(1) variance
• Algorithmic optimization (O(n) → O(1))

Why it matters: Incremental statistics eliminate redundant computation:

• Naive: Recalculate from scratch every time (O(n))
• Incremental: Update with +/- operations (O(1))

Real-world impact:

#![allow(unused)]
fn main() {
// Naive mean (O(n) per call)
fn average(&self) -> f64 {
    self.window.iter().sum::<f64>() / self.len() as f64
    // Sums 1000 elements every time!
}

// Incremental mean (O(1) per call)
fn average(&self) -> f64 {
    self.running_sum / self.len() as f64
    // Just one division!
}
}

Performance comparison (window size 1000, 1M updates):

Measured impact (real hardware):

Naive approach:  1000ms for 1M averages
Incremental:     2ms for 1M averages (500× faster!)

Milestone 3: Min/Max with Monotonic Deque

Concepts applied:

• Monotonic deque data structure
• Amortized O(1) complexity
• Invariant maintenance (increasing/decreasing order)
• Index tracking for window expiration

Why it matters: Monotonic deque provides O(1) min/max without scanning:

• Naive: Scan entire window every query (O(n))
• Monotonic deque: Front element is always min/max (O(1) amortized)

Real-world impact:

#![allow(unused)]
fn main() {
// Naive min (O(n) per call)
fn min(&self) -> f64 {
    self.window.iter().min().unwrap()
    // Scans 1000 elements!
}

// Monotonic deque min (O(1) amortized)
fn min(&self) -> f64 {
    self.min_deque.front().unwrap().1
    // Just return front element!
}
}

Performance comparison (window size 1000, 1M updates with min queries):

Amortized analysis proof:

n pushes to window:
- Each element pushed to deque once: n operations
- Each element popped from deque at most once: ≤n operations
- Total: ≤2n operations
- Amortized: 2n/n = O(1) per push

Real-world validation: Used in HFT (high-frequency trading) for real-time market statistics.

Milestone 4: Median and Percentiles with select_nth_unstable

Concepts applied:

• Quickselect algorithm (O(n) vs O(n log n))
• Partial sorting (only find k-th element)
• Percentile calculation
• Tradeoff: No incremental version (must copy window)

Why it matters: Quickselect finds median faster than sorting:

• Sorting: O(n log n) ≈ 10,000 ops for n=1000
• Quickselect: O(n) average ≈ 1,000 ops (10× faster)

Real-world impact:

#![allow(unused)]
fn main() {
// Sorting approach (O(n log n))
fn median(&self) -> f64 {
    let mut copy: Vec<_> = self.window.iter().copied().collect();
    copy.sort();  // Sorts ALL elements
    copy[copy.len() / 2]
}

// Quickselect approach (O(n) average)
fn median(&self) -> f64 {
    let mut copy: Vec<_> = self.window.iter().copied().collect();
    copy.select_nth_unstable(copy.len() / 2);  // Only partially sorts
    copy[copy.len() / 2]
}
}

Performance comparison (window size 1000, 100 median calls):

Percentile efficiency:

#![allow(unused)]
fn main() {
// Multiple percentiles: sort once, extract many
fn percentiles(&self, ps: &[f64]) -> Vec<f64> {
    let mut copy = self.window.clone();
    copy.sort();  // O(n log n) once

    ps.iter().map(|&p| {
        let idx = (p / 100.0 * (copy.len() - 1) as f64) as usize;
        copy[idx]  // O(1) per percentile
    }).collect()
}

// For 10 percentiles:
// Naive (10× quickselect): 10n ≈ 10,000 ops
// Sort once: n log n + 10 ≈ 10,010 ops (similar!)
// But sort enables binary search, interpolation, etc.
}

Milestone 5: Multiple Windows Simultaneously

Concepts applied:

• Single-pass multi-window processing
• Data structure composition (Vec of windows)
• Amortized iteration cost
• Cache locality benefits

Why it matters: Real-world monitoring needs multiple time scales:

• Short-term (1 min): Detect spikes
• Medium-term (5 min): Smooth out noise
• Long-term (1 hour): Track trends

Single-pass processing shares iteration cost.

Real-world impact:

#![allow(unused)]
fn main() {
// Separate processing (3 passes over data)
for &size in &[60, 300, 900] {
    let mut window = Window::new(size);
    for &value in &data {  // Iterate data 3 times!
        window.push(value);
    }
}

// Multi-window (1 pass over data)
let mut windows = vec![
    Window::new(60),
    Window::new(300),
    Window::new(900),
];

for &value in &data {  // Iterate data once!
    for window in &mut windows {
        window.push(value);
    }
}
}

Performance comparison (1M data points, 3 windows):

Why single-pass is faster:

• Data loaded from RAM once (cache-friendly)
• Better CPU prefetching (sequential access)
• Less memory bandwidth usage

Real-world use case (server monitoring):

Monitor response times with 3 windows:
- 1 minute (detect immediate issues)
- 5 minutes (confirm sustained problems)
- 15 minutes (track long-term degradation)

Alert logic:
- 1min p99 > 1000ms → page on-call
- 5min avg > 500ms → create incident
- 15min trend > 10% increase → capacity planning

Milestone 6: Anomaly Detection with Z-Score

Concepts applied:

• Z-score statistical metric
• Threshold-based alerting
• Statistical significance (3-sigma rule)
• Real-time monitoring pattern

Why it matters: Automated anomaly detection enables proactive monitoring:

• No human watching dashboards 24/7
• Immediate alerts on unusual patterns
• Context-aware (adapts to normal baseline)

Real-world impact:

#![allow(unused)]
fn main() {
// Without anomaly detection: manual monitoring
loop {
    let value = read_sensor();
    println!("Temperature: {}°C", value);  // Human must watch!
}

// With anomaly detection: automated alerting
loop {
    let value = read_sensor();
    if let Some(anomaly) = detector.push(value, timestamp) {
        alert_on_call("Temperature spike: {}°C (z={})",
                      anomaly.value, anomaly.z_score);
    }
}
}

Z-score interpretation:

False positive/negative tradeoff:

Real-world applications:

• Infrastructure: CPU spike detection (normal: 30%, anomaly: 95%)
• Network: DDoS attack detection (normal: 1Gbps, anomaly: 10Gbps)
• IoT: Sensor fault detection (normal: 22°C ± 2°C, anomaly: 50°C)
• Finance: Fraud detection (normal: $50, anomaly: $5000)

Production example (AWS CloudWatch):

Metric: API latency
Window: 5 minutes
Threshold: z > 3

Alert triggered when:
- Normal latency: 100ms ± 20ms (mean ± std)
- Spike: 250ms (z = (250-100)/20 = 7.5)
- Action: Page on-call engineer

Project-Wide Benefits

Algorithmic optimizations stack:

End-to-end performance (1M data points, 3 windows, full statistics):

Real-world comparison:

When to use these techniques:

• ✅ Real-time monitoring: Server metrics, IoT sensors
• ✅ Financial analysis: Stock prices, trading signals
• ✅ Network monitoring: Traffic analysis, intrusion detection
• ✅ Scientific instruments: Sensor data, experiment monitoring
• ❌ Batch processing: Use database aggregation instead
• ❌ Low-frequency data: Simple averaging is fine

Production lessons:

1. Use VecDeque for sliding windows (500× faster than Vec)
2. Incremental statistics for real-time (avoid recalculation)
3. Monotonic deque for min/max (O(1) amortized is game-changer)
4. Quickselect for percentiles (10× faster than sorting)
5. Single-pass multi-window (cache-friendly, reduces I/O)
6. Z-score for anomaly detection (simple, effective, interpretable)

Build The Project

Milestone 1: Basic Sliding Window with VecDeque

Goal: Implement fixed-size sliding window that maintains recent N elements.

What to implement:

• Use VecDeque for efficient push/pop from both ends
• push() adds element, pop_front() if window full
• Compute basic statistics (average, min, max)
• Provide slice view of window contents

Architecture:

• Structs: SlidingWindow<T>
• Fields: window: VecDeque<T>, capacity: usize
• Functions:
  • new(capacity: usize) -> Self - Create window
  • push(value: T) -> Option<T> - Add value, return evicted
  • as_slice() -> &[T] - Zero-copy view
  • len() -> usize - Current size
  • is_full() -> bool - Check capacity
  • average() -> Option<f64> - Compute mean (for f64 windows)
  • min() -> Option<f64> - Find minimum
  • max() -> Option<f64> - Find maximum

Starter Code:

#![allow(unused)]
fn main() {
use std::collections::VecDeque;

/// Fixed-size sliding window
/// Role: Maintain most recent N values
#[derive(Debug, Clone)]
pub struct SlidingWindow<T> {
    window: VecDeque<T>,               // Circular buffer  
    capacity: usize,                   // Maximum window size  
}

impl<T: Clone> SlidingWindow<T> {
    /// Create new sliding window
    /// Role: Initialize with capacity
    pub fn new(capacity: usize) -> Self {
        todo!("Create VecDeque with capacity")
    }

    /// Add value to window
    /// Role: Maintain FIFO ordering
    pub fn push(&mut self, value: T) -> Option<T> {
        todo!("Pop front if full, push back value")
    }

    /// Get slice view of window
    /// Role: Zero-copy access to data
    pub fn as_slice(&self) -> &[T] {
        todo!("Use make_contiguous or as_slices")
    }

    /// Current number of elements
    /// Role: Query window fill
    pub fn len(&self) -> usize {
        self.window.len()
    }

    /// Check if window is full
    /// Role: Determine if at capacity
    pub fn is_full(&self) -> bool {
        self.window.len() == self.capacity
    }

    /// Check if window is empty
    /// Role: Guard against empty statistics
    pub fn is_empty(&self) -> bool {
        self.window.is_empty()
    }
}

/// Statistics for numeric windows
impl SlidingWindow<f64> {
    /// Compute average
    /// Role: Basic statistic over window
    pub fn average(&self) -> Option<f64> {
        todo!("Sum all values, divide by length")
    }

    /// Find minimum value
    /// Role: Window minimum
    pub fn min(&self) -> Option<f64> {
        todo!("Use iterator min_by with partial_cmp")
    }

    /// Find maximum value
    /// Role: Window maximum
    pub fn max(&self) -> Option<f64> {
        todo!("Use iterator max_by with partial_cmp")
    }
}
}

Checkpoint Tests:

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_create_window() {
        let window: SlidingWindow<f64> = SlidingWindow::new(10);
        assert_eq!(window.len(), 0);
        assert!(!window.is_full());
        assert!(window.is_empty());
    }

    #[test]
    fn test_push_values() {
        let mut window = SlidingWindow::new(3);

        assert_eq!(window.push(1.0), None);
        assert_eq!(window.len(), 1);

        assert_eq!(window.push(2.0), None);
        assert_eq!(window.len(), 2);

        assert_eq!(window.push(3.0), None);
        assert_eq!(window.len(), 3);
        assert!(window.is_full());
    }

    #[test]
    fn test_window_eviction() {
        let mut window = SlidingWindow::new(3);

        window.push(1.0);
        window.push(2.0);
        window.push(3.0);

        // Window is full, next push should evict oldest
        let evicted = window.push(4.0);
        assert_eq!(evicted, Some(1.0));
        assert_eq!(window.len(), 3);

        let evicted = window.push(5.0);
        assert_eq!(evicted, Some(2.0));
    }

    #[test]
    fn test_window_fifo_order() {
        let mut window = SlidingWindow::new(3);

        window.push(10.0);
        window.push(20.0);
        window.push(30.0);
        window.push(40.0); // Evicts 10.0

        let slice = window.as_slice();
        assert_eq!(slice, &[20.0, 30.0, 40.0]);
    }

    #[test]
    fn test_average() {
        let mut window = SlidingWindow::new(5);

        assert_eq!(window.average(), None); // Empty

        window.push(10.0);
        assert_eq!(window.average(), Some(10.0));

        window.push(20.0);
        assert_eq!(window.average(), Some(15.0));

        window.push(30.0);
        assert_eq!(window.average(), Some(20.0));
    }

    #[test]
    fn test_min_max() {
        let mut window = SlidingWindow::new(5);

        assert_eq!(window.min(), None);
        assert_eq!(window.max(), None);

        window.push(30.0);
        window.push(10.0);
        window.push(50.0);
        window.push(20.0);

        assert_eq!(window.min(), Some(10.0));
        assert_eq!(window.max(), Some(50.0));
    }

    #[test]
    fn test_min_max_after_eviction() {
        let mut window = SlidingWindow::new(3);

        window.push(10.0);
        window.push(50.0); // Max
        window.push(30.0);

        assert_eq!(window.max(), Some(50.0));

        window.push(20.0); // Evicts 10.0
        assert_eq!(window.max(), Some(50.0));

        window.push(25.0); // Evicts 50.0 (the max!)
        assert_eq!(window.max(), Some(30.0));
    }

    #[test]
    fn test_as_slice_zero_copy() {
        let mut window = SlidingWindow::new(100);

        for i in 0..50 {
            window.push(i as f64);
        }

        let slice1 = window.as_slice();
        let slice2 = window.as_slice();

        // Should be same pointer (zero-copy)
        assert_eq!(slice1.as_ptr(), slice2.as_ptr());
    }
}
}

Milestone 2: Incremental Statistics (Avoid Re-Scanning)

Goal: Maintain running sum to compute average in O(1) instead of O(n).

Why the previous milestone is not enough: Milestone 1 computes average by summing entire window on every call (O(n)). For a stream of 1M values with window size 1000, this is 1 billion operations.

What’s the improvement: Incremental updates reduce average computation from O(n) to O(1). Instead of summing 1000 values per update, we add one and subtract one. For 1M updates:

• Before: 1M × 1000 = 1 billion operations
• After: 1M × 2 = 2 million operations (500x faster)

Optimization focus: Speed through algorithmic improvement (O(n) → O(1)).

Architecture:

• Structs: IncrementalWindow
• Fields: window: VecDeque<f64>, capacity: usize, running_sum: f64, running_sum_sq: f64
• Functions:
  • new(capacity: usize) -> Self - Create window
  • push(value: f64) - Update with new value
  • average() -> Option<f64> - O(1) mean
  • variance() -> Option<f64> - O(1) variance
  • std_dev() -> Option<f64> - O(1) standard deviation

Starter Code:

#![allow(unused)]
fn main() {
/// Sliding window with incremental statistics
/// Role: O(1) statistics computation
#[derive(Debug, Clone)]
pub struct IncrementalWindow {
    window: VecDeque<f64>,                // Data storage                          
    capacity: usize,                      // Maximum size                                
    running_sum: f64,                     // Sum of all values                          
    running_sum_sq: f64,                  // Sum of squared values (for variance)    
}

impl IncrementalWindow {
    /// Create new incremental window
    /// Role: Initialize with zero statistics
    pub fn new(capacity: usize) -> Self {
        todo!("Initialize all fields to zero/empty")
    }

    /// Add value with incremental update
    /// Role: Maintain running statistics
    pub fn push(&mut self, value: f64) {
        todo!("Evict old, update running sums, push new")
    }

    /// Get mean in O(1)
    /// Role: Fast average computation
    pub fn average(&self) -> Option<f64> {
        todo!("Return running_sum / len")
    }

    /// Get variance in O(1)
    /// Role: Fast variance using sum of squares
    pub fn variance(&self) -> Option<f64> {
        todo!("Use Var(X) = E[X²] - E[X]²")
    }

    /// Get standard deviation in O(1)
    /// Role: Square root of variance
    pub fn std_dev(&self) -> Option<f64> {
        todo!("Return sqrt(variance)")
    }

    /// Get current length
    /// Role: Query window size
    pub fn len(&self) -> usize {
        self.window.len()
    }

    /// Check if empty
    /// Role: Guard for statistics
    pub fn is_empty(&self) -> bool {
        self.window.is_empty()
    }
}
}

Checkpoint Tests:

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_incremental_average() {
        let mut window = IncrementalWindow::new(5);

        window.push(10.0);
        assert_eq!(window.average(), Some(10.0));

        window.push(20.0);
        assert_eq!(window.average(), Some(15.0));

        window.push(30.0);
        assert_eq!(window.average(), Some(20.0));
    }

    #[test]
    fn test_incremental_average_after_eviction() {
        let mut window = IncrementalWindow::new(3);

        window.push(10.0);
        window.push(20.0);
        window.push(30.0);
        assert_eq!(window.average(), Some(20.0));

        window.push(40.0); // Evicts 10.0
        // Window now: [20, 30, 40], avg = 30
        assert_eq!(window.average(), Some(30.0));
    }

    #[test]
    fn test_variance_calculation() {
        let mut window = IncrementalWindow::new(5);

        window.push(2.0);
        window.push(4.0);
        window.push(4.0);
        window.push(4.0);
        window.push(5.0);

        // Mean = 3.8, Variance = 0.96
        let variance = window.variance().unwrap();
        assert!((variance - 0.96).abs() < 0.01);
    }

    #[test]
    fn test_std_dev_calculation() {
        let mut window = IncrementalWindow::new(5);

        window.push(2.0);
        window.push(4.0);
        window.push(4.0);
        window.push(4.0);
        window.push(5.0);

        // Std dev = sqrt(0.96) ≈ 0.98
        let std_dev = window.std_dev().unwrap();
        assert!((std_dev - 0.98).abs() < 0.01);
    }

    #[test]
    fn test_incremental_vs_naive() {
        // Verify incremental matches naive computation
        let mut window = IncrementalWindow::new(100);

        let values: Vec<f64> = (0..100).map(|i| i as f64 * 1.5).collect();

        for &v in &values {
            window.push(v);
        }

        let incremental_avg = window.average().unwrap();

        // Naive average
        let naive_avg = values.iter().sum::<f64>() / values.len() as f64;

        assert!((incremental_avg - naive_avg).abs() < 0.0001);
    }

    #[test]
    fn test_performance_incremental() {
        use std::time::Instant;

        let mut window = IncrementalWindow::new(1000);

        // Fill window
        for i in 0..1000 {
            window.push(i as f64);
        }

        // Measure average computation time
        let iterations = 100000;
        let start = Instant::now();

        for _ in 0..iterations {
            let _ = window.average();
        }

        let elapsed = start.elapsed();

        println!("Time for {} incremental averages: {:?}", iterations, elapsed);

        // Should be very fast (microseconds for 100K operations)
        assert!(elapsed.as_millis() < 100);
    }

    #[test]
    fn test_variance_empty_window() {
        let window = IncrementalWindow::new(10);
        assert_eq!(window.variance(), None);
    }

    #[test]
    fn test_variance_single_value() {
        let mut window = IncrementalWindow::new(10);
        window.push(5.0);

        // Variance of single value is undefined or zero
        // (Implementation choice - we return None for < 2 values)
        assert_eq!(window.variance(), None);
    }
}
}

Milestone 3: Min/Max with Monotonic Deque

Goal: Maintain min/max in O(1) amortized time using monotonic deque.

Why the previous milestone is not enough: Finding min/max requires scanning window (O(n)). For 1M updates with window 1000, this is 1 billion operations.

What’s the improvement: Monotonic deque maintains min/max in O(1) amortized time. Algorithm keeps only elements that could be min/max in future:

• For min: if new element is smaller than back of deque, pop back (it can never be min)
• Front of deque is always current min

Complexity: Each element pushed once, popped at most once → O(1) amortized.

Optimization focus: Speed through clever data structure (O(n) → O(1) amortized).

Architecture:

• Structs: MinMaxWindow
• Fields: window: VecDeque<(usize, f64)>, min_deque: VecDeque<(usize, f64)>, max_deque: VecDeque<(usize, f64)>, capacity: usize, index: usize
• Functions:
  • new(capacity: usize) -> Self - Create window
  • push(value: f64) - Add value with deque maintenance
  • min() -> Option<f64> - O(1) minimum
  • max() -> Option<f64> - O(1) maximum

Starter Code:

#![allow(unused)]
fn main() {
/// Sliding window with O(1) min/max
/// Role: Efficient min/max tracking
#[derive(Debug)]
pub struct MinMaxWindow {
    window: VecDeque<(usize, f64)>,             // Values with indices    
    min_deque: VecDeque<(usize, f64)>,          // Monotonic increasing
    max_deque: VecDeque<(usize, f64)>,          // Monotonic decreasing
    capacity: usize,                            // Maximum size                          
    index: usize,                               // Global index counter                     
}

impl MinMaxWindow {
    /// Create new min/max window
    /// Role: Initialize deques
    pub fn new(capacity: usize) -> Self {
        todo!("Initialize all deques")
    }

    /// Add value with monotonic deque update
    /// Role: Maintain min/max invariants
    pub fn push(&mut self, value: f64) {
        todo!("Evict old, update min/max deques, push new")
    }

    /// Get minimum in O(1)
    /// Role: Return front of min_deque
    pub fn min(&self) -> Option<f64> {
        todo!("Return min_deque front value")
    }

    /// Get maximum in O(1)
    /// Role: Return front of max_deque
    pub fn max(&self) -> Option<f64> {
        todo!("Return max_deque front value")
    }

    /// Get current length
    /// Role: Query window size
    pub fn len(&self) -> usize {
        self.window.len()
    }
}
}

Checkpoint Tests:

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_min_max_basic() {
        let mut window = MinMaxWindow::new(5);

        window.push(30.0);
        assert_eq!(window.min(), Some(30.0));
        assert_eq!(window.max(), Some(30.0));

        window.push(10.0);
        assert_eq!(window.min(), Some(10.0));
        assert_eq!(window.max(), Some(30.0));

        window.push(50.0);
        assert_eq!(window.min(), Some(10.0));
        assert_eq!(window.max(), Some(50.0));
    }

    #[test]
    fn test_min_max_after_eviction() {
        let mut window = MinMaxWindow::new(3);

        window.push(10.0); // min
        window.push(30.0);
        window.push(20.0);

        assert_eq!(window.min(), Some(10.0));

        window.push(40.0); // Evicts 10.0
        // Window: [30, 20, 40]
        assert_eq!(window.min(), Some(20.0));
        assert_eq!(window.max(), Some(40.0));
    }

    #[test]
    fn test_max_eviction() {
        let mut window = MinMaxWindow::new(3);

        window.push(50.0); // max
        window.push(30.0);
        window.push(20.0);

        assert_eq!(window.max(), Some(50.0));

        window.push(25.0); // Evicts 50.0
        // Window: [30, 20, 25]
        assert_eq!(window.max(), Some(30.0));
    }

    #[test]
    fn test_monotonic_sequence() {
        let mut window = MinMaxWindow::new(5);

        // Increasing sequence
        for i in 1..=5 {
            window.push(i as f64);
        }

        assert_eq!(window.min(), Some(1.0));
        assert_eq!(window.max(), Some(5.0));

        // Continue increasing
        window.push(6.0); // Evicts 1.0
        assert_eq!(window.min(), Some(2.0));
        assert_eq!(window.max(), Some(6.0));
    }

    #[test]
    fn test_all_same_values() {
        let mut window = MinMaxWindow::new(5);

        for _ in 0..10 {
            window.push(42.0);
        }

        assert_eq!(window.min(), Some(42.0));
        assert_eq!(window.max(), Some(42.0));
    }

    #[test]
    fn test_alternating_values() {
        let mut window = MinMaxWindow::new(4);

        window.push(10.0);
        window.push(50.0);
        window.push(10.0);
        window.push(50.0);

        assert_eq!(window.min(), Some(10.0));
        assert_eq!(window.max(), Some(50.0));
    }

    #[test]
    fn test_performance_vs_naive() {
        use std::time::Instant;

        let values: Vec<f64> = (0..10000).map(|i| (i % 100) as f64).collect();

        // Monotonic deque approach
        let mut window = MinMaxWindow::new(100);
        let start = Instant::now();

        for &v in &values {
            window.push(v);
            let _ = window.min();
            let _ = window.max();
        }

        let deque_time = start.elapsed();

        // Naive approach (for comparison)
        let mut naive_window = SlidingWindow::new(100);
        let start = Instant::now();

        for &v in &values {
            naive_window.push(v);
            let _ = naive_window.min();
            let _ = naive_window.max();
        }

        let naive_time = start.elapsed();

        println!("Monotonic deque: {:?}", deque_time);
        println!("Naive approach: {:?}", naive_time);

        // Monotonic deque should be significantly faster
        assert!(deque_time < naive_time);
    }

    #[test]
    fn test_empty_window() {
        let window = MinMaxWindow::new(10);
        assert_eq!(window.min(), None);
        assert_eq!(window.max(), None);
    }
}
}

Milestone 4: Median and Percentiles with select_nth_unstable

Goal: Compute median efficiently using quickselect algorithm.

Why the previous milestone is not enough: We have mean, min, max but not median or percentiles. Naive approach sorts entire window (O(n log n)).

What’s the improvement: Quickselect finds k-th element in O(n) average time, faster than sorting:

• Sorting: O(n log n) ≈ 10,000 ops for n=1000
• Quickselect: O(n) ≈ 1,000 ops (10x faster)

For streaming percentiles, this is significant. Note: median requires copying window (can’t be incremental like mean).

Optimization focus: Speed through better algorithm (O(n log n) → O(n)).

Architecture:

• Add methods to IncrementalWindow:
  • median() -> Option<f64> - 50th percentile
  • percentile(p: f64) -> Option<f64> - Any percentile

Starter Code:

#![allow(unused)]
fn main() {
impl IncrementalWindow {
    /// Compute median using quickselect
    /// Role: O(n) median calculation
    pub fn median(&self) -> Option<f64> {
        todo!("Copy to temp buffer, use select_nth_unstable")
    }

    /// Compute arbitrary percentile
    /// Role: Find p-th percentile (0-100)
    pub fn percentile(&self, p: f64) -> Option<f64> {
        todo!("Validate p, calculate index, use select_nth_unstable")
    }

    /// Get multiple percentiles efficiently
    /// Role: Compute p50, p95, p99 in one pass
    pub fn percentiles(&self, ps: &[f64]) -> Vec<Option<f64>> {
        todo!("Sort once, extract multiple percentiles")
    }
}
}

Checkpoint Tests:

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_median_odd_count() {
        let mut window = IncrementalWindow::new(10);

        window.push(1.0);
        window.push(3.0);
        window.push(2.0);

        // Sorted: [1, 2, 3], median = 2
        assert_eq!(window.median(), Some(2.0));
    }

    #[test]
    fn test_median_even_count() {
        let mut window = IncrementalWindow::new(10);

        window.push(1.0);
        window.push(2.0);
        window.push(3.0);
        window.push(4.0);

        // Sorted: [1, 2, 3, 4], median = (2 + 3) / 2 = 2.5
        assert_eq!(window.median(), Some(2.5));
    }

    #[test]
    fn test_percentile_basic() {
        let mut window = IncrementalWindow::new(10);

        for i in 1..=10 {
            window.push(i as f64);
        }

        // p0 = 1, p50 = 5.5, p100 = 10
        assert_eq!(window.percentile(0.0), Some(1.0));
        assert_eq!(window.percentile(100.0), Some(10.0));

        let p50 = window.percentile(50.0).unwrap();
        assert!((p50 - 5.5).abs() < 0.1);
    }

    #[test]
    fn test_percentile_p95() {
        let mut window = IncrementalWindow::new(100);

        for i in 1..=100 {
            window.push(i as f64);
        }

        let p95 = window.percentile(95.0).unwrap();
        // p95 of 1..100 should be around 95
        assert!((p95 - 95.0).abs() < 2.0);
    }

    #[test]
    fn test_percentile_invalid_range() {
        let mut window = IncrementalWindow::new(10);
        window.push(5.0);

        assert_eq!(window.percentile(-1.0), None);
        assert_eq!(window.percentile(101.0), None);
    }

    #[test]
    fn test_percentile_empty() {
        let window = IncrementalWindow::new(10);
        assert_eq!(window.percentile(50.0), None);
    }

    #[test]
    fn test_multiple_percentiles() {
        let mut window = IncrementalWindow::new(100);

        for i in 1..=100 {
            window.push(i as f64);
        }

        let percentiles = window.percentiles(&[25.0, 50.0, 75.0, 95.0]);

        assert_eq!(percentiles.len(), 4);
        assert!(percentiles[0].is_some()); // p25
        assert!(percentiles[1].is_some()); // p50
        assert!(percentiles[2].is_some()); // p75
        assert!(percentiles[3].is_some()); // p95
    }

    #[test]
    fn test_median_vs_sort() {
        // Verify quickselect matches sort-based median
        let mut window = IncrementalWindow::new(1000);

        for i in 0..1000 {
            window.push((i * 7 % 1000) as f64); // Pseudo-random
        }

        let median = window.median().unwrap();

        // Manual calculation
        let mut values: Vec<f64> = window.window.iter().copied().collect();
        values.sort_by(|a, b| a.partial_cmp(b).unwrap());
        let expected = (values[499] + values[500]) / 2.0;

        assert!((median - expected).abs() < 0.01);
    }

    #[test]
    fn test_performance_quickselect_vs_sort() {
        use std::time::Instant;

        let mut window = IncrementalWindow::new(10000);

        for i in 0..10000 {
            window.push(i as f64);
        }

        // Quickselect median
        let start = Instant::now();
        for _ in 0..100 {
            let _ = window.median();
        }
        let quickselect_time = start.elapsed();

        // Sort-based median
        let start = Instant::now();
        for _ in 0..100 {
            let mut temp: Vec<f64> = window.window.iter().copied().collect();
            temp.sort_by(|a, b| a.partial_cmp(b).unwrap());
            let _ = temp[temp.len() / 2];
        }
        let sort_time = start.elapsed();

        println!("Quickselect: {:?}", quickselect_time);
        println!("Sort: {:?}", sort_time);

        // Quickselect should be faster
        assert!(quickselect_time < sort_time);
    }
}
}

Milestone 5: Multiple Windows Simultaneously

Goal: Track multiple window sizes (1min, 5min, 1hour) with single pass.

Why the previous milestone is not enough: Often we need statistics at multiple time scales (short-term and long-term trends). Processing data separately for each window multiplies computational cost.

What’s the improvement: Single-pass multi-window processing shares data ingestion cost. For 3 windows:

• Separate processing: 3 passes over data
• Combined processing: 1 pass over data (3x faster)

Optimization focus: Speed through single-pass processing.

Architecture:

• Structs: MultiWindowAnalyzer, WindowStats
• Fields: windows: Vec<IncrementalWindow>, window_sizes: Vec<usize>
• Functions:
  • new(window_sizes: Vec<usize>) -> Self - Create analyzer
  • push(value: f64) - Update all windows
  • get_stats(window_index: usize) -> Option<WindowStats> - Query specific window
  • all_stats() -> Vec<WindowStats> - Get all statistics

Starter Code:

#![allow(unused)]
fn main() {
/// Multi-window analyzer
/// Role: Track multiple time scales
#[derive(Debug)]
pub struct MultiWindowAnalyzer {
    windows: Vec<IncrementalWindow>,            // All windows       
    window_sizes: Vec<usize>,                   // Sizes for each window    
}

impl MultiWindowAnalyzer {
    /// Create multi-window analyzer
    /// Role: Initialize all windows
    pub fn new(window_sizes: Vec<usize>) -> Self {
        todo!("Create IncrementalWindow for each size")
    }

    /// Update all windows
    /// Role: Single-pass update
    pub fn push(&mut self, value: f64) {
        todo!("Call push on each window")
    }

    /// Get statistics for specific window
    /// Role: Query individual window
    pub fn get_stats(&self, window_index: usize) -> Option<WindowStats> {
        todo!("Extract stats from window at index")
    }

    /// Get statistics for all windows
    /// Role: Complete snapshot
    pub fn all_stats(&self) -> Vec<WindowStats> {
        todo!("Collect stats from all windows")
    }

    /// Get number of windows
    /// Role: Query configuration
    pub fn window_count(&self) -> usize {
        self.windows.len()
    }
}

/// Statistics for a window
#[derive(Debug, Clone)]
pub struct WindowStats {
    pub average: Option<f64>,             // Mean                    
    pub std_dev: Option<f64>,             // Standard deviation      
    pub median: Option<f64>,              // Median                   
    pub min: Option<f64>,                 // Minimum                     
    pub max: Option<f64>,                 // maximum                     
    pub window_size: usize,               // Window configuration      
}
}

Checkpoint Tests:

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_multi_window_creation() {
        let analyzer = MultiWindowAnalyzer::new(vec![10, 60, 300]);

        assert_eq!(analyzer.window_count(), 3);
    }

    #[test]
    fn test_multi_window_push() {
        let mut analyzer = MultiWindowAnalyzer::new(vec![3, 5]);

        for i in 1..=10 {
            analyzer.push(i as f64);
        }

        let stats0 = analyzer.get_stats(0).unwrap();
        let stats1 = analyzer.get_stats(1).unwrap();

        // Window 0 (size 3): last 3 values [8, 9, 10]
        assert_eq!(stats0.average, Some(9.0));

        // Window 1 (size 5): last 5 values [6, 7, 8, 9, 10]
        assert_eq!(stats1.average, Some(8.0));
    }

    #[test]
    fn test_all_stats() {
        let mut analyzer = MultiWindowAnalyzer::new(vec![5, 10, 20]);

        for i in 1..=30 {
            analyzer.push(i as f64);
        }

        let all_stats = analyzer.all_stats();

        assert_eq!(all_stats.len(), 3);
        assert!(all_stats[0].average.is_some());
        assert!(all_stats[1].average.is_some());
        assert!(all_stats[2].average.is_some());
    }

    #[test]
    fn test_different_window_behaviors() {
        let mut analyzer = MultiWindowAnalyzer::new(vec![2, 5]);

        analyzer.push(10.0);
        analyzer.push(20.0);
        analyzer.push(30.0);
        analyzer.push(40.0);
        analyzer.push(50.0);

        let stats_small = analyzer.get_stats(0).unwrap(); // Window size 2
        let stats_large = analyzer.get_stats(1).unwrap(); // Window size 5

        // Small window: [40, 50]
        assert_eq!(stats_small.average, Some(45.0));

        // Large window: [10, 20, 30, 40, 50]
        assert_eq!(stats_large.average, Some(30.0));
    }

    #[test]
    fn test_single_pass_efficiency() {
        use std::time::Instant;

        let window_sizes = vec![10, 50, 100, 500, 1000];
        let data: Vec<f64> = (0..10000).map(|i| i as f64).collect();

        // Multi-window (single pass)
        let mut analyzer = MultiWindowAnalyzer::new(window_sizes.clone());
        let start = Instant::now();

        for &value in &data {
            analyzer.push(value);
        }

        let multi_time = start.elapsed();

        // Separate windows (multiple passes)
        let start = Instant::now();

        for &size in &window_sizes {
            let mut window = IncrementalWindow::new(size);
            for &value in &data {
                window.push(value);
            }
        }

        let separate_time = start.elapsed();

        println!("Multi-window (single pass): {:?}", multi_time);
        println!("Separate windows: {:?}", separate_time);

        // Multi-window should be faster or comparable
        // (In practice, might be slightly slower due to multiple window management,
        // but saves on data iteration)
    }

    #[test]
    fn test_empty_stats() {
        let analyzer = MultiWindowAnalyzer::new(vec![10]);

        let stats = analyzer.get_stats(0).unwrap();

        assert_eq!(stats.average, None);
        assert_eq!(stats.std_dev, None);
        assert_eq!(stats.median, None);
    }

    #[test]
    fn test_invalid_window_index() {
        let analyzer = MultiWindowAnalyzer::new(vec![10, 20]);

        assert!(analyzer.get_stats(2).is_none());
        assert!(analyzer.get_stats(10).is_none());
    }
}
}

Milestone 6: Anomaly Detection with Z-Score

Goal: Detect anomalies using statistical thresholds.

Why the previous milestone is not enough: Statistics alone don’t identify problems. Anomaly detection enables proactive monitoring and alerting.

What’s the improvement: Automated anomaly detection catches issues in real-time. Instead of humans watching dashboards, systems alert on unusual patterns. Z-score method is simple yet effective: values more than 3 standard deviations from mean are flagged as anomalies.

Optimization focus: Practical application of streaming statistics.

Architecture:

• Structs: AnomalyDetector, Anomaly
• Fields: analyzer: MultiWindowAnalyzer, threshold: f64, anomalies: Vec<Anomaly>
• Functions:
  • new(window_sizes, threshold) -> Self - Create detector
  • push(value, timestamp) -> Option<Anomaly> - Check for anomaly
  • anomaly_rate(total_points) -> f64 - Calculate percentage

Starter Code:

#![allow(unused)]
fn main() {
/// Anomaly detector using z-score
/// Role: Statistical outlier detection
#[derive(Debug)]
pub struct AnomalyDetector {
    analyzer: MultiWindowAnalyzer,           // Window statistics   
    threshold: f64,                          // Z-score threshold (typically 3.0)  
    anomalies: Vec<Anomaly>,                 // Detected anomalies        
}

/// Detected anomaly
#[derive(Debug, Clone)]
pub struct Anomaly {
    pub value: f64,                             // Anomalous value               
    pub z_score: f64,                           // How many std devs from mean 
    pub timestamp: usize,                       // When detected           
    pub window_stats: WindowStats,              // Context        
}

impl AnomalyDetector {
    /// Create anomaly detector
    /// Role: Initialize with configuration
    pub fn new(window_sizes: Vec<usize>, threshold: f64) -> Self {
        todo!("Create analyzer and empty anomaly list")
    }

    /// Add value and check for anomaly
    /// Role: Real-time detection
    pub fn push(&mut self, value: f64, timestamp: usize) -> Option<Anomaly> {
        todo!("Update analyzer, calculate z-score, check threshold")
    }

    /// Calculate anomaly rate
    /// Role: Summary statistic
    pub fn anomaly_rate(&self, total_points: usize) -> f64 {
        todo!("Return anomalies.len() / total_points")
    }

    /// Get all detected anomalies
    /// Role: Retrieve history
    pub fn get_anomalies(&self) -> &[Anomaly] {
        &self.anomalies
    }

    /// Clear anomaly history
    /// Role: Reset detector
    pub fn clear_anomalies(&mut self) {
        self.anomalies.clear();
    }
}
}

Checkpoint Tests:

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_no_anomalies_in_normal_data() {
        let mut detector = AnomalyDetector::new(vec![100], 3.0);

        // Generate normal data (mean=50, std_dev small)
        for i in 0..200 {
            let value = 50.0 + ((i % 10) as f64 - 5.0);
            detector.push(value, i);
        }

        assert_eq!(detector.get_anomalies().len(), 0);
    }

    #[test]
    fn test_detect_obvious_outlier() {
        let mut detector = AnomalyDetector::new(vec![100], 3.0);

        // Normal values around 50
        for i in 0..100 {
            detector.push(50.0, i);
        }

        // Outlier
        let anomaly = detector.push(200.0, 100);

        assert!(anomaly.is_some());

        let anomaly = anomaly.unwrap();
        assert!(anomaly.z_score.abs() > 3.0);
        assert_eq!(anomaly.value, 200.0);
    }

    #[test]
    fn test_z_score_calculation() {
        let mut detector = AnomalyDetector::new(vec![10], 2.0);

        // Mean = 10, Std dev = 0
        for i in 0..10 {
            detector.push(10.0, i);
        }

        // Add value outside 2 std devs
        let anomaly = detector.push(15.0, 10);

        if let Some(anomaly) = anomaly {
            // Z-score = (15 - 10) / std_dev
            assert!(anomaly.z_score > 2.0);
        }
    }

    #[test]
    fn test_anomaly_rate() {
        let mut detector = AnomalyDetector::new(vec![50], 3.0);

        let total_points = 1000;

        for i in 0..total_points {
            let value = if i % 100 == 0 {
                // Every 100th point is anomaly
                1000.0
            } else {
                50.0
            };

            detector.push(value, i);
        }

        let rate = detector.anomaly_rate(total_points);

        // Should detect ~10 anomalies out of 1000
        assert!(rate > 0.005 && rate < 0.015); // Between 0.5% and 1.5%
    }

    #[test]
    fn test_anomaly_context() {
        let mut detector = AnomalyDetector::new(vec![20], 3.0);

        for i in 0..30 {
            detector.push(100.0, i);
        }

        // Anomaly
        let anomaly = detector.push(200.0, 30).unwrap();

        assert_eq!(anomaly.timestamp, 30);
        assert!(anomaly.window_stats.average.is_some());
        assert!(anomaly.window_stats.std_dev.is_some());
    }

    #[test]
    fn test_different_thresholds() {
        let data: Vec<f64> = (0..100).map(|i| 50.0 + (i % 20) as f64).collect();

        // Strict threshold (more sensitive)
        let mut strict = AnomalyDetector::new(vec![50], 2.0);
        for (i, &v) in data.iter().enumerate() {
            strict.push(v, i);
        }

        // Lenient threshold (less sensitive)
        let mut lenient = AnomalyDetector::new(vec![50], 4.0);
        for (i, &v) in data.iter().enumerate() {
            lenient.push(v, i);
        }

        // Strict should detect more anomalies
        assert!(strict.get_anomalies().len() >= lenient.get_anomalies().len());
    }

    #[test]
    fn test_clear_anomalies() {
        let mut detector = AnomalyDetector::new(vec![10], 2.0);

        for i in 0..10 {
            detector.push(10.0, i);
        }

        detector.push(50.0, 10); // Anomaly

        assert_eq!(detector.get_anomalies().len(), 1);

        detector.clear_anomalies();

        assert_eq!(detector.get_anomalies().len(), 0);
    }

    #[test]
    fn test_real_world_monitoring() {
        // Simulate server response time monitoring
        let mut detector = AnomalyDetector::new(
            vec![60, 300, 900], // 1min, 5min, 15min windows
            3.0
        );

        // Normal response times: 100-200ms
        for i in 0..1000 {
            let normal_time = 150.0 + ((i % 50) as f64 - 25.0);
            detector.push(normal_time, i);
        }

        // Spike: 2000ms response time
        let anomaly = detector.push(2000.0, 1000);

        assert!(anomaly.is_some());
        println!("Detected anomaly: {:?}", anomaly.unwrap());
    }
}
}

Complete Working Example

#![allow(unused)]
fn main() {
use csv::{Reader, ReaderBuilder, StringRecord};
use rayon::prelude::*;
use rusqlite::{Connection, Transaction};
use std::{
    cmp::Ordering,
    collections::HashSet,
    error::Error,
    fs::File,
    io::{self, BufRead, BufReader},
    time::Instant,
};

// =============================================================================
// Milestone 1: Basic CSV Parser with Structured Records
// =============================================================================

/// CSV record representing a user
#[derive(Debug, Clone)]
pub struct UserRecord {
    pub id: u64,
    pub name: String,
    pub email: String,
    pub age: u32,
    pub country: String,
}

/// CSV parsing errors
#[derive(Debug, thiserror::Error)]
pub enum ParseError {
    #[error("Invalid CSV format: {0}")]
    InvalidFormat(String),

    #[error("Invalid type for field '{field}': '{value}'")]
    InvalidType { field: String, value: String },

    #[error("Missing required field: {0}")]
    MissingField(String),
}

impl UserRecord {
    /// Parse CSV row into UserRecord
    /// Role: Convert StringRecord to typed struct
    pub fn from_csv_row(row: &StringRecord) -> Result<Self, ParseError> {
        if row.len() < 5 {
            return Err(ParseError::InvalidFormat(format!("{:?}", row)));
        }

        let id = row
            .get(0)
            .ok_or_else(|| ParseError::MissingField("id".into()))?
            .parse()
            .map_err(|value: std::num::ParseIntError| ParseError::InvalidType {
                field: "id".into(),
                value: value.to_string(),
            })?;

        let age = row
            .get(3)
            .ok_or_else(|| ParseError::MissingField("age".into()))?
            .parse()
            .map_err(|value: std::num::ParseIntError| ParseError::InvalidType {
                field: "age".into(),
                value: value.to_string(),
            })?;

        Ok(UserRecord {
            id,
            name: row
                .get(1)
                .ok_or_else(|| ParseError::MissingField("name".into()))?
                .to_string(),
            email: row
                .get(2)
                .ok_or_else(|| ParseError::MissingField("email".into()))?
                .to_string(),
            age,
            country: row
                .get(4)
                .ok_or_else(|| ParseError::MissingField("country".into()))?
                .to_string(),
        })
    }
}

/// Parse entire CSV file
/// Role: Read file and convert all valid rows
pub fn parse_csv(path: &str) -> Result<Vec<UserRecord>, Box<dyn Error>> {
    let mut reader = Reader::from_path(path)?;
    let mut records = Vec::new();

    for result in reader.records() {
        let record = result?;
        if let Ok(user) = UserRecord::from_csv_row(&record) {
            records.push(user);
        }
    }

    Ok(records)
}

// =============================================================================
// Milestone 2: Pre-Allocate Capacity to Eliminate Reallocations
// =============================================================================

/// Count lines in file
/// Role: Estimate capacity needed for Vec
pub fn count_lines(path: &str) -> Result<usize, io::Error> {
    let file = File::open(path)?;
    let reader = BufReader::new(file);
    Ok(reader.lines().count())
}

/// Parse CSV with pre-allocated capacity
/// Role: Eliminate reallocations during parsing
pub fn parse_csv_optimized(path: &str) -> Result<Vec<UserRecord>, Box<dyn Error>> {
    let total_lines = count_lines(path)?;
    let mut reader = Reader::from_path(path)?;
    let mut records = Vec::with_capacity(total_lines.saturating_sub(1));

    for result in reader.records() {
        let record = result?;
        if let Ok(user) = UserRecord::from_csv_row(&record) {
            records.push(user);
        }
    }

    Ok(records)
}

/// Track allocation statistics
/// Role: Measure allocation efficiency
#[derive(Debug, Default)]
pub struct AllocationStats {
    pub allocations: usize,
    pub reallocations: usize,
    pub bytes_copied: usize,
}

/// Wrapper to track Vec allocations
/// Role: Observe allocation behavior
pub struct TrackedVec<T> {
    vec: Vec<T>,
    stats: AllocationStats,
}

impl<T> TrackedVec<T> {
    /// Create with capacity tracking
    /// Role: Initialize with known capacity
    pub fn with_capacity(capacity: usize) -> Self {
        Self {
            vec: Vec::with_capacity(capacity),
            stats: AllocationStats {
                allocations: 1,
                ..Default::default()
            },
        }
    }

    /// Push with reallocation tracking
    /// Role: Monitor when reallocations occur
    pub fn push(&mut self, value: T) {
        if self.vec.len() == self.vec.capacity() {
            self.stats.reallocations += 1;
            self.stats.bytes_copied += self.vec.len() * std::mem::size_of::<T>();
        }
        self.vec.push(value);
    }

    /// Get statistics
    /// Role: Query allocation metrics
    pub fn stats(&self) -> &AllocationStats {
        &self.stats
    }
}

// =============================================================================
// Milestone 3: Streaming Processing with Chunking
// =============================================================================

/// Process CSV in chunks with callback
/// Role: Enable processing files larger than RAM
pub fn process_csv_chunked<F>(
    path: &str,
    chunk_size: usize,
    mut process_chunk: F,
) -> Result<(), Box<dyn Error>>
where
    F: FnMut(&[UserRecord]),
{
    let mut reader = ReaderBuilder::new().has_headers(true).from_path(path)?;
    let mut chunk = Vec::with_capacity(chunk_size);

    for result in reader.records() {
        let record = result?;
        if let Ok(user) = UserRecord::from_csv_row(&record) {
            chunk.push(user);
        }

        if chunk.len() == chunk_size {
            process_chunk(&chunk);
            chunk.clear();
        }
    }

    if !chunk.is_empty() {
        process_chunk(&chunk);
    }

    Ok(())
}

/// Statistics for chunked processing
#[derive(Debug, Default)]
pub struct ChunkStats {
    pub total_chunks: usize,
    pub total_records: usize,
    pub peak_memory_bytes: usize,
}

/// Process CSV with statistics tracking
/// Role: Monitor chunking efficiency
pub fn process_csv_chunked_with_stats<F>(
    path: &str,
    chunk_size: usize,
    mut process_chunk: F,
) -> Result<ChunkStats, Box<dyn Error>>
where
    F: FnMut(&[UserRecord]),
{
    let mut stats = ChunkStats::default();
    process_csv_chunked(path, chunk_size, |chunk| {
        stats.total_chunks += 1;
        stats.total_records += chunk.len();
        stats.peak_memory_bytes = stats
            .peak_memory_bytes
            .max(chunk.len() * std::mem::size_of::<UserRecord>());
        process_chunk(chunk);
    })?;
    Ok(stats)
}

// =============================================================================
// Milestone 4: Batch Database Inserts with Transactions
// =============================================================================

/// Insert batch of records in single query
/// Multi-row INSERT
/// Role: Minimize database round-trips
pub fn insert_batch(tx: &Transaction, records: &[UserRecord]) -> Result<(), rusqlite::Error> {
    if records.is_empty() {
        return Ok(());
    }

    let mut sql = String::from("INSERT INTO users (id, name, email, age, country) VALUES ");
    let mut params_vec: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(records.len() * 5);

    for (idx, record) in records.iter().enumerate() {
        if idx > 0 {
            sql.push_str(", ");
        }
        sql.push_str("(?, ?, ?, ?, ?)");
        params_vec.push(&record.id);
        params_vec.push(&record.name);
        params_vec.push(&record.email);
        params_vec.push(&record.age);
        params_vec.push(&record.country);
    }

    let mut stmt = tx.prepare(&sql)?;
    stmt.execute(params_vec.as_slice())?;
    Ok(())
}

/// Create database schema
/// Role: Initialize tables
pub fn create_schema(conn: &Connection) -> Result<(), rusqlite::Error> {
    conn.execute(
        "CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            email TEXT NOT NULL,
            age INTEGER NOT NULL,
            country TEXT NOT NULL
        )",
        [],
    )?;
    Ok(())
}

/// Import CSV to database with batching
/// Role: Production-ready CSV import
pub fn import_csv_to_db(
    path: &str,
    db_path: &str,
    batch_size: usize,
) -> Result<(), Box<dyn Error>> {
    let mut conn = Connection::open(db_path)?;
    create_schema(&conn)?;

    process_csv_chunked(path, batch_size, |chunk| {
        let tx = conn.transaction().unwrap();
        insert_batch(&tx, chunk).unwrap();
        tx.commit().unwrap();
    })?;

    Ok(())
}

/// Database import statistics
#[derive(Debug, Default)]
pub struct ImportStats {
    pub records_imported: usize,
    pub records_failed: usize,
    pub batches_processed: usize,
    pub duration_ms: u64,
}

/// Import with detailed statistics
/// Role: Monitor import performance
pub fn import_csv_to_db_with_stats(
    path: &str,
    db_path: &str,
    batch_size: usize,
) -> Result<ImportStats, Box<dyn Error>> {
    let mut conn = Connection::open(db_path)?;
    create_schema(&conn)?;

    let start = Instant::now();
    let mut stats = ImportStats::default();

    process_csv_chunked(path, batch_size, |chunk| {
        let tx = conn.transaction().unwrap();
        match insert_batch(&tx, chunk) {
            Ok(_) => {
                tx.commit().unwrap();
                stats.records_imported += chunk.len();
                stats.batches_processed += 1;
            }
            Err(_) => stats.records_failed += chunk.len(),
        }
    })?;

    stats.duration_ms = start.elapsed().as_millis() as u64;
    Ok(stats)
}

// =============================================================================
// Milestone 5: In-Place Deduplication with sort + dedup
// =============================================================================

impl PartialEq for UserRecord {
    fn eq(&self, other: &Self) -> bool {
        self.id == other.id
    }
}

impl Eq for UserRecord {}

impl PartialOrd for UserRecord {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for UserRecord {
    fn cmp(&self, other: &Self) -> Ordering {
        self.id.cmp(&other.id)
    }
}

/// In-place deduplication using sort
/// Role: Memory-efficient deduplication
pub fn deduplicate_chunk(chunk: &mut Vec<UserRecord>) {
    chunk.sort_unstable_by(|a, b| a.id.cmp(&b.id));
    chunk.dedup_by(|a, b| a.id == b.id);
}

/// HashSet-based deduplication
/// Role: Comparison baseline
pub fn deduplicate_hashset(chunk: &mut Vec<UserRecord>) {
    let mut seen = HashSet::new();
    chunk.retain(|record| seen.insert(record.id));
}

/// Benchmark deduplication strategies
/// Role: Measure optimization impact
pub fn benchmark_dedup(records: &mut Vec<UserRecord>) {
    let mut chunk_sort = records.clone();
    let start = Instant::now();
    deduplicate_chunk(&mut chunk_sort);
    let sort_time = start.elapsed();

    let mut chunk_hash = records.clone();
    let start = Instant::now();
    deduplicate_hashset(&mut chunk_hash);
    let hash_time = start.elapsed();

    println!("Sort+dedup: {:?}", sort_time);
    println!("HashSet dedup: {:?}", hash_time);
}

// =============================================================================
// Milestone 6: Parallel Processing with Rayon
// =============================================================================

/// Process CSV chunks in parallel
/// Role: Maximize CPU utilization
pub fn process_csv_parallel(
    path: &str,
    chunk_size: usize,
) -> Result<Vec<UserRecord>, Box<dyn Error>> {
    let mut chunks = Vec::new();
    process_csv_chunked(path, chunk_size, |chunk| {
        chunks.push(chunk.to_vec());
    })?;

    Ok(chunks.into_par_iter().flatten().collect())
}

/// Transform record
/// Role: Example CPU-bound operation
pub fn transform_record(record: &mut UserRecord) {
    record.email = record.email.to_lowercase();
    record.country = record.country.to_uppercase();
}

/// Parallel CSV processor with transformations
/// Role: Full parallel pipeline
pub fn process_and_transform_parallel(
    path: &str,
    chunk_size: usize,
) -> Result<Vec<UserRecord>, Box<dyn Error>> {
    let mut chunks = Vec::new();
    process_csv_chunked(path, chunk_size, |chunk| {
        chunks.push(chunk.to_vec());
    })?;

    let mut records: Vec<UserRecord> = chunks
        .into_par_iter()
        .flat_map(|mut chunk| {
            chunk.par_iter_mut().for_each(transform_record);
            deduplicate_chunk(&mut chunk);
            chunk
        })
        .collect();

    deduplicate_chunk(&mut records);
    Ok(records)
}

/// Benchmark sequential vs parallel
/// Role: Measure parallelism benefit
pub fn benchmark_parallel(path: &str, chunk_size: usize) {
    let start = Instant::now();
    let _ = parse_csv_optimized(path).unwrap();
    let seq_time = start.elapsed();

    let start = Instant::now();
    let _ = process_csv_parallel(path, chunk_size).unwrap();
    let par_time = start.elapsed();

    println!("Sequential: {:?}", seq_time);
    println!("Parallel: {:?}", par_time);
}

// =============================================================================
// Tests
// =============================================================================

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;
    use std::sync::{Arc, Mutex};
    use tempfile::NamedTempFile;

    fn create_test_csv(content: &str) -> NamedTempFile {
        let mut file = NamedTempFile::new().unwrap();
        file.write_all(content.as_bytes()).unwrap();
        file
    }

    // ----- Milestone 1 tests -----
    #[test]
    fn test_parse_valid_row() {
        let csv_content = "id,name,email,age,country\n1,Alice,alice@test.com,30,US";
        let file = create_test_csv(csv_content);

        let records = parse_csv(file.path().to_str().unwrap()).unwrap();

        assert_eq!(records.len(), 1);
        assert_eq!(records[0].id, 1);
        assert_eq!(records[0].name, "Alice");
        assert_eq!(records[0].email, "alice@test.com");
        assert_eq!(records[0].age, 30);
        assert_eq!(records[0].country, "US");
    }

    #[test]
    fn test_parse_multiple_rows() {
        let csv_content = "\
id,name,email,age,country
1,Alice,alice@test.com,30,US
2,Bob,bob@test.com,25,UK
3,Charlie,charlie@test.com,35,CA";

        let file = create_test_csv(csv_content);
        let records = parse_csv(file.path().to_str().unwrap()).unwrap();

        assert_eq!(records.len(), 3);
        assert_eq!(records[1].name, "Bob");
        assert_eq!(records[2].age, 35);
    }

    #[test]
    fn test_parse_invalid_age() {
        let row = StringRecord::from(vec!["1", "Alice", "alice@test.com", "invalid", "US"]);
        let result = UserRecord::from_csv_row(&row);

        assert!(matches!(
            result.unwrap_err(),
            ParseError::InvalidType { field, .. } if field == "age"
        ));
    }

    #[test]
    fn test_parse_missing_field() {
        let row = StringRecord::from(vec!["1", "Alice", "alice@test.com"]);
        let result = UserRecord::from_csv_row(&row);
        assert!(matches!(result.unwrap_err(), ParseError::InvalidFormat(_)));
    }

    #[test]
    fn test_parse_skips_invalid_rows() {
        let csv_content = "\
id,name,email,age,country
1,Alice,alice@test.com,30,US
2,Bob,bob@test.com,invalid_age,UK
3,Charlie,charlie@test.com,35,CA";

        let file = create_test_csv(csv_content);
        let records = parse_csv(file.path().to_str().unwrap()).unwrap();

        assert_eq!(records.len(), 2);
        assert_eq!(records[0].id, 1);
        assert_eq!(records[1].id, 3);
    }

    #[test]
    fn test_parse_empty_file() {
        let csv_content = "id,name,email,age,country\n";
        let file = create_test_csv(csv_content);

        let records = parse_csv(file.path().to_str().unwrap()).unwrap();
        assert_eq!(records.len(), 0);
    }

    // ----- Milestone 2 tests -----
    #[test]
    fn test_count_lines() {
        let csv_content = "header\nrow1\nrow2\nrow3";
        let file = create_test_csv(csv_content);

        let count = count_lines(file.path().to_str().unwrap()).unwrap();
        assert_eq!(count, 4);
    }

    #[test]
    fn test_optimized_parsing_allocates_once() {
        let csv_content = "\
id,name,email,age,country
1,Alice,alice@test.com,30,US
2,Bob,bob@test.com,25,UK
3,Charlie,charlie@test.com,35,CA";

        let file = create_test_csv(csv_content);
        let records = parse_csv_optimized(file.path().to_str().unwrap()).unwrap();

        assert_eq!(records.len(), 3);
        assert!(records.capacity() >= records.len());
    }

    #[test]
    fn test_tracked_vec_no_reallocations() {
        let mut vec = TrackedVec::with_capacity(10);
        for i in 0..10 {
            vec.push(i);
        }
        let stats = vec.stats();
        assert_eq!(stats.allocations, 1);
        assert_eq!(stats.reallocations, 0);
    }

    #[test]
    fn test_tracked_vec_with_reallocations() {
        let mut vec = TrackedVec::with_capacity(4);
        for i in 0..8 {
            vec.push(i);
        }
        let stats = vec.stats();
        assert!(stats.reallocations > 0);
    }

    #[test]
    fn test_capacity_efficiency() {
        let csv_content = "\
id,name,email,age,country
1,Alice,alice@test.com,30,US
2,Bob,bob@test.com,25,UK
3,Charlie,charlie@test.com,35,CA";

        let file = create_test_csv(csv_content);
        let records = parse_csv_optimized(file.path().to_str().unwrap()).unwrap();
        assert!(records.capacity() < records.len() * 2);
    }

    // ----- Milestone 3 tests -----
    #[test]
    fn test_chunked_processing() {
        let csv_content = "\
id,name,email,age,country
1,Alice,alice@test.com,30,US
2,Bob,bob@test.com,25,UK
3,Charlie,charlie@test.com,35,CA
4,Diana,diana@test.com,28,FR
5,Eve,eve@test.com,32,DE";

        let file = create_test_csv(csv_content);

        let chunks = Arc::new(Mutex::new(0));
        let total = Arc::new(Mutex::new(0));

        let chunks_clone = chunks.clone();
        let total_clone = total.clone();

        process_csv_chunked(file.path().to_str().unwrap(), 2, |chunk| {
            *chunks_clone.lock().unwrap() += 1;
            *total_clone.lock().unwrap() += chunk.len();
        })
        .unwrap();

        assert_eq!(*chunks.lock().unwrap(), 3);
        assert_eq!(*total.lock().unwrap(), 5);
    }

    #[test]
    fn test_chunk_buffer_reuse() {
        let mut content = String::from("id,name,email,age,country\n");
        for i in 0..100 {
            content.push_str(&format!("{},User{},user{}@test.com,{},US\n", i, i, i, 20));
        }

        let file = create_test_csv(&content);
        let sizes = Arc::new(Mutex::new(Vec::new()));
        let sizes_clone = sizes.clone();

        process_csv_chunked(file.path().to_str().unwrap(), 10, |chunk| {
            sizes_clone.lock().unwrap().push(chunk.len());
        })
        .unwrap();

        let locked = sizes.lock().unwrap();
        for &size in locked.iter().take(locked.len() - 1) {
            assert_eq!(size, 10);
        }
        assert!(*locked.last().unwrap() <= 10);
    }

    #[test]
    fn test_process_empty_file_chunked() {
        let file = create_test_csv("id,name,email,age,country\n");
        let called = Arc::new(Mutex::new(false));
        let clone = called.clone();
        process_csv_chunked(file.path().to_str().unwrap(), 10, |_| {
            *clone.lock().unwrap() = true;
        })
        .unwrap();
        assert!(!*called.lock().unwrap());
    }

    #[test]
    fn test_chunked_with_stats() {
        let mut content = String::from("id,name,email,age,country\n");
        for i in 0..50 {
            content.push_str(&format!("{},User{},user{}@test.com,{},US\n", i, i, i, 20));
        }

        let file = create_test_csv(&content);
        let stats = process_csv_chunked_with_stats(file.path().to_str().unwrap(), 10, |_| {})
            .unwrap();

        assert_eq!(stats.total_chunks, 5);
        assert_eq!(stats.total_records, 50);
        assert!(stats.peak_memory_bytes > 0);
    }

    // ----- Milestone 4 tests -----
    #[test]
    fn test_create_schema() {
        let db = NamedTempFile::new().unwrap();
        let conn = Connection::open(db.path()).unwrap();
        create_schema(&conn).unwrap();

        let mut stmt = conn
            .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='users'")
            .unwrap();
        assert!(stmt.exists([]).unwrap());
    }

    #[test]
    fn test_insert_single_batch() {
        let db = NamedTempFile::new().unwrap();
        let mut conn = Connection::open(db.path()).unwrap();
        create_schema(&conn).unwrap();

        let records = vec![
            UserRecord {
                id: 1,
                name: "Alice".into(),
                email: "alice@test.com".into(),
                age: 30,
                country: "US".into(),
            },
            UserRecord {
                id: 2,
                name: "Bob".into(),
                email: "bob@test.com".into(),
                age: 25,
                country: "UK".into(),
            },
        ];

        let tx = conn.transaction().unwrap();
        insert_batch(&tx, &records).unwrap();
        tx.commit().unwrap();

        let count: i64 = conn
            .query_row("SELECT COUNT(*) FROM users", [], |row| row.get(0))
            .unwrap();
        assert_eq!(count, 2);
    }

    #[test]
    fn test_import_csv_to_db() {
        let csv_content = "\
id,name,email,age,country
1,Alice,alice@test.com,30,US
2,Bob,bob@test.com,25,UK";

        let csv_file = create_test_csv(csv_content);
        let db = NamedTempFile::new().unwrap();

        import_csv_to_db(
            csv_file.path().to_str().unwrap(),
            db.path().to_str().unwrap(),
            10,
        )
        .unwrap();

        let conn = Connection::open(db.path()).unwrap();
        let count: i64 = conn
            .query_row("SELECT COUNT(*) FROM users", [], |row| row.get(0))
            .unwrap();
        assert_eq!(count, 2);
    }

    #[test]
    fn test_batch_transaction_atomicity() {
        let db = NamedTempFile::new().unwrap();
        let mut conn = Connection::open(db.path()).unwrap();
        create_schema(&conn).unwrap();
        conn.execute("CREATE UNIQUE INDEX idx_email ON users(email)", [])
            .unwrap();

        let records = vec![
            UserRecord {
                id: 1,
                name: "Alice".into(),
                email: "alice@test.com".into(),
                age: 30,
                country: "US".into(),
            },
            UserRecord {
                id: 2,
                name: "Bob".into(),
                email: "alice@test.com".into(),
                age: 25,
                country: "UK".into(),
            },
        ];

        let tx = conn.transaction().unwrap();
        assert!(insert_batch(&tx, &records).is_err());
        drop(tx);

        let count: i64 = conn
            .query_row("SELECT COUNT(*) FROM users", [], |row| row.get(0))
            .unwrap();
        assert_eq!(count, 0);
    }

    #[test]
    fn test_import_with_stats() {
        let mut content = String::from("id,name,email,age,country\n");
        for i in 0..50 {
            content.push_str(&format!("{},User{},user{}@test.com,{},US\n", i, i, i, 20));
        }
        let csv_file = create_test_csv(&content);
        let db = NamedTempFile::new().unwrap();

        let stats = import_csv_to_db_with_stats(
            csv_file.path().to_str().unwrap(),
            db.path().to_str().unwrap(),
            10,
        )
        .unwrap();

        assert_eq!(stats.records_imported, 50);
        assert_eq!(stats.batches_processed, 5);
        assert!(stats.duration_ms > 0);
    }

    // ----- Milestone 5 tests -----
    #[test]
    fn test_dedup_removes_duplicates() {
        let mut records = vec![
            UserRecord { id: 1, name: "A".into(), email: "a@test.com".into(), age: 30, country: "US".into() },
            UserRecord { id: 2, name: "B".into(), email: "b@test.com".into(), age: 25, country: "UK".into() },
            UserRecord { id: 1, name: "A2".into(), email: "a2@test.com".into(), age: 31, country: "CA".into() },
        ];
        deduplicate_chunk(&mut records);
        assert_eq!(records.len(), 2);
    }

    #[test]
    fn test_dedup_ordering() {
        let mut records = vec![
            UserRecord { id: 3, name: "C".into(), email: "c@test.com".into(), age: 30, country: "US".into() },
            UserRecord { id: 1, name: "A".into(), email: "a@test.com".into(), age: 25, country: "UK".into() },
            UserRecord { id: 2, name: "B".into(), email: "b@test.com".into(), age: 25, country: "UK".into() },
        ];
        deduplicate_chunk(&mut records);
        assert_eq!(records.iter().map(|r| r.id).collect::<Vec<_>>(), vec![1, 2, 3]);
    }

    #[test]
    fn test_hashset_dedup_correctness() {
        let mut records = vec![
            UserRecord { id: 1, name: "A".into(), email: "a@test.com".into(), age: 30, country: "US".into() },
            UserRecord { id: 1, name: "A2".into(), email: "a2@test.com".into(), age: 31, country: "CA".into() },
            UserRecord { id: 2, name: "B".into(), email: "b@test.com".into(), age: 25, country: "UK".into() },
        ];
        deduplicate_hashset(&mut records);
        assert_eq!(records.len(), 2);
    }

    #[test]
    fn test_dedup_methods_equivalent() {
        let original = vec![
            UserRecord { id: 3, name: "C".into(), email: "c@test.com".into(), age: 30, country: "US".into() },
            UserRecord { id: 1, name: "A".into(), email: "a@test.com".into(), age: 25, country: "UK".into() },
            UserRecord { id: 3, name: "C2".into(), email: "c2@test.com".into(), age: 32, country: "CA".into() },
        ];

        let mut sort = original.clone();
        let mut hash = original.clone();
        deduplicate_chunk(&mut sort);
        deduplicate_hashset(&mut hash);
        assert_eq!(sort.len(), hash.len());
    }

    // ----- Milestone 6 tests -----
    #[test]
    fn test_parallel_processing_correctness() {
        let mut content = String::from("id,name,email,age,country\n");
        for i in 0..100 {
            content.push_str(&format!("{},User{},user{}@test.com,{},US\n", i, i, i, 20));
        }
        let file = create_test_csv(&content);

        let seq = parse_csv_optimized(file.path().to_str().unwrap()).unwrap();
        let par = process_csv_parallel(file.path().to_str().unwrap(), 10).unwrap();
        assert_eq!(seq.len(), par.len());
    }

    #[test]
    fn test_parallel_with_transformations() {
        let csv_content = "\
id,name,email,age,country
1,Alice,ALICE@TEST.COM,30,us
2,Bob,BOB@TEST.COM,25,uk";
        let file = create_test_csv(csv_content);

        let records = process_and_transform_parallel(file.path().to_str().unwrap(), 1).unwrap();
        for record in records {
            assert_eq!(record.email, record.email.to_lowercase());
            assert_eq!(record.country, record.country.to_uppercase());
        }
    }

    #[test]
    fn test_parallel_deduplication() {
        let mut content = String::from("id,name,email,age,country\n");
        for i in 0..50 {
            content.push_str(&format!("{},User{},user{}@test.com,{},US\n", i, i, i, 20));
            content.push_str(&format!("{},UserDup{},dup{}@test.com,{},US\n", i, i, i, 21));
        }
        let file = create_test_csv(&content);

        let records = process_and_transform_parallel(file.path().to_str().unwrap(), 10).unwrap();
        assert_eq!(records.len(), 50);
    }

    #[test]
    fn test_parallel_chunk_independence() {
        let mut content = String::from("id,name,email,age,country\n");
        for i in 0..20 {
            content.push_str(&format!("{},User{},user{}@test.com,{},US\n", i, i, i, 20));
        }
        let file = create_test_csv(&content);

        let records = process_csv_parallel(file.path().to_str().unwrap(), 5).unwrap();
        assert_eq!(records.len(), 20);
    }
}

}