llm-optimizer-processor 0.1.0

# Stream Processor

High-performance, production-ready stream processing library for LLM Auto-Optimizer.

## Features

- **Windowing**: Tumbling, Sliding, and Session windows
- **Aggregations**: Count, Sum, Average, Min, Max, Percentiles, Standard Deviation
- **Time Semantics**: Event-time processing with watermark support
- **Late Events**: Configurable handling for out-of-order events
- **Performance**: Sub-100ms latency, 100K+ events/sec throughput
- **Statistics**: Online variance, EMA, sliding window stats, anomaly detection

## Quick Start

```rust
use processor::{StreamProcessorBuilder, aggregation::CompositeAggregator};
use chrono::Duration;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Create a stream processor with 5-minute tumbling windows
    let mut processor = StreamProcessorBuilder::new()
        .build_tumbling(
            Duration::minutes(5),
            |_key| {
                let mut agg = CompositeAggregator::new();
                agg.add_count("requests")
                   .add_avg("avg_latency")
                   .add_p95("p95_latency")
                   .add_p99("p99_latency");
                agg
            },
        );

    // Get results stream
    let mut results = processor.results().unwrap();

    // Process events
    processor.process_event(
        "my-service".to_string(),
        chrono::Utc::now(),
        123.45  // metric value
    ).await?;

    // Collect results
    tokio::spawn(async move {
        while let Some(result) = results.next().await {
            println!("Window: {}", result.window);
            println!("Results: {:?}", result.results);
        }
    });

    Ok(())
}
```

## Window Types

### Tumbling Windows
Fixed-size, non-overlapping windows for periodic aggregations:

```rust
let processor = StreamProcessorBuilder::new()
    .build_tumbling(Duration::minutes(5), aggregator_factory);
```

### Sliding Windows
Overlapping windows for moving averages and smoothed metrics:

```rust
let processor = StreamProcessorBuilder::new()
    .build_sliding(
        Duration::minutes(10),  // Window size
        Duration::seconds(30),  // Slide interval
        aggregator_factory
    );
```

### Session Windows
Gap-based windows for activity clustering:

```rust
let processor = StreamProcessorBuilder::new()
    .build_session(
        Duration::seconds(30),  // Inactivity gap
        aggregator_factory
    );
```

## Aggregations

### Composite Aggregator
Run multiple aggregations efficiently:

```rust
let mut agg = CompositeAggregator::new();
agg.add_count("total_requests")
   .add_avg("avg_response_time")
   .add_min("min_response_time")
   .add_max("max_response_time")
   .add_p50("p50_response_time")
   .add_p95("p95_response_time")
   .add_p99("p99_response_time")
   .add_stddev("stddev_response_time");
```

### Simple Aggregators
Use individual aggregators for specific metrics:

```rust
use processor::aggregation::{
    AverageAggregator,
    PercentileAggregator,
    StandardDeviationAggregator,
};

let avg = AverageAggregator::new();
let p95 = PercentileAggregator::new(95);
let stddev = StandardDeviationAggregator::new();
```

## Configuration

```rust
use processor::StreamProcessorConfig;
use chrono::Duration;

let config = StreamProcessorConfig {
    allow_late_events: true,
    late_event_threshold: Duration::seconds(30),
    watermark_interval: Duration::seconds(1),
    window_retention: Duration::hours(1),
    max_windows_per_key: 1000,
    result_buffer_size: 1000,
};

let processor = StreamProcessorBuilder::new()
    .with_config(config)
    .build_tumbling(Duration::minutes(5), aggregator_factory);
```

## Advanced Statistics

The library includes advanced statistical utilities:

### Online Variance
Numerically stable variance calculation:

```rust
use processor::aggregation::statistics::OnlineVariance;

let mut variance = OnlineVariance::new();
for value in values {
    variance.update(value);
}
let std_dev = variance.std_dev();
```

### Exponential Moving Average
Smoothed metrics over time:

```rust
use processor::aggregation::statistics::ExponentialMovingAverage;

let mut ema = ExponentialMovingAverage::with_span(10.0)?;
ema.update(value);
let smoothed = ema.value();
```

### Sliding Window Statistics
Fixed-size window with percentiles:

```rust
use processor::aggregation::statistics::SlidingWindowStats;

let mut stats = SlidingWindowStats::new(100)?;
for value in values {
    stats.update(value);
}
let p95 = stats.percentile(95.0)?;
let median = stats.median();
```

### Anomaly Detection
Z-score based anomaly detection:

```rust
use processor::aggregation::statistics::ZScoreAnomalyDetector;

let mut detector = ZScoreAnomalyDetector::new(3.0); // 3-sigma threshold
let is_anomaly = detector.check(value);
```

## Event Deduplication

Prevent duplicate events from being processed multiple times:

```rust
use processor::state::{RedisStateBackend, RedisConfig, StateBackend};
use std::sync::Arc;
use std::time::Duration;

// Create Redis backend for deduplication
let redis_config = RedisConfig::builder()
    .url("redis://localhost:6379")
    .key_prefix("dedup:")
    .default_ttl(Duration::from_secs(3600))
    .build()?;

let backend = Arc::new(RedisStateBackend::new(redis_config).await?);

// Check for duplicates
let event_id = "event_123";
let key = format!("dedup:{}", event_id);

if backend.get(key.as_bytes()).await?.is_none() {
    // First time seeing this event - process it
    backend.put_with_ttl(key.as_bytes(), b"1", Duration::from_secs(3600)).await?;
    process_event(event).await?;
} else {
    // Duplicate - skip processing
    info!("Duplicate event detected: {}", event_id);
}
```

### Features

- **Multiple Backends**: Redis (distributed), Sled (local), In-memory
- **Configurable TTL**: Automatic cleanup of old deduplication records
- **High Performance**: 100K+ events/sec with Redis, 200K+ with Sled
- **Bloom Filters**: Probabilistic deduplication for ultra-high throughput
- **Multi-Key Strategies**: Event ID, transaction ID, idempotency keys, content hash

### Configuration Options

```rust
// Basic configuration with in-memory backend
let backend = MemoryStateBackend::new(Some(Duration::from_secs(3600)));

// Production configuration with Redis
let redis_config = RedisConfig::builder()
    .url("redis://redis-cluster:6379")
    .key_prefix("dedup:")
    .default_ttl(Duration::from_secs(86400)) // 24 hours
    .pool_size(10, 50)  // min 10, max 50 connections
    .build()?;

// High-performance local storage with Sled
let sled_backend = SledStateBackend::new("./dedup.db")?;
```

### Use Cases

1. **Kafka Message Deduplication**: Handle at-least-once delivery semantics
2. **API Idempotency**: Prevent duplicate requests using idempotency keys
3. **Multi-Source Aggregation**: Deduplicate events from multiple sources
4. **Retry Safety**: Ensure safe retries without duplicate processing
5. **Exactly-Once Processing**: Maintain exactly-once semantics across restarts

See [deduplication/README.md](./src/deduplication/README.md) for complete documentation.

## Time-Series Normalization

Transform irregularly sampled time-series data into uniform intervals for analysis and processing:

```rust
use processor::normalization::{TimeSeriesNormalizer, NormalizationConfig, FillStrategy};
use chrono::Duration;

// Configure normalization with 1-second intervals
let config = NormalizationConfig::builder()
    .interval(Duration::seconds(1))
    .fill_strategy(FillStrategy::Linear)
    .max_gap_duration(Duration::seconds(60))
    .out_of_order_threshold(Duration::seconds(5))
    .build()?;

let mut normalizer = TimeSeriesNormalizer::new(config);

// Process irregular events
normalizer.process_event(timestamp1, value1).await?;
normalizer.process_event(timestamp2, value2).await?;

// Get normalized time-series with uniform intervals
let normalized = normalizer.normalize().await?;
```

### Visual Example

```
Original (irregular sampling):
  t=0.0s: 10.0
  t=0.3s: 12.0
  t=1.7s: 18.0
  t=4.9s: 35.0

Normalized (1-second intervals, linear interpolation):
  t=0s: 10.0
  t=1s: 14.4   (interpolated)
  t=2s: 18.0
  t=3s: 24.3   (interpolated)
  t=4s: 30.7   (interpolated)
  t=5s: 35.0
```

### Fill Strategies

- **Linear**: Interpolate between known points (best for continuous metrics)
- **Forward Fill**: Carry last value forward (best for discrete states)
- **Backward Fill**: Carry next value backward (batch processing)
- **Zero Fill**: Fill with zero (best for event counts)
- **Mean Fill**: Fill with average (best for noisy data)
- **Polynomial**: Higher-order interpolation (scientific data)
- **Spline**: Smooth curves without oscillation (high-quality visualization)

### Features

- **Irregular Sampling**: Handle variable-rate data collection
- **Out-of-Order Events**: Automatic buffering and reordering
- **Gap Filling**: Intelligent interpolation for missing data
- **Timestamp Alignment**: Floor, ceil, or round to intervals
- **Resampling**: Upsample or downsample to target frequencies
- **Multi-Metric**: Normalize multiple metrics with different strategies
- **Outlier Detection**: Z-score based anomaly filtering
- **High Performance**: 500K+ events/sec with linear interpolation

### Configuration Options

```rust
let config = NormalizationConfig {
    // Time interval between normalized points
    interval: Duration::seconds(1),

    // Timestamp alignment strategy
    alignment: AlignmentStrategy::Floor,  // or Ceil, Round

    // Gap filling strategy
    fill_strategy: FillStrategy::Linear,

    // Maximum gap to fill (larger gaps left as missing)
    max_gap_duration: Some(Duration::seconds(60)),

    // Out-of-order handling
    out_of_order_buffer_size: 1000,
    out_of_order_threshold: Duration::seconds(5),
};
```

### Use Cases

1. **IoT Sensor Data**: Normalize irregular sensor readings for analysis
2. **Financial Data**: Align OHLC data to uniform intervals
3. **System Metrics**: Standardize monitoring data from multiple sources
4. **ML Feature Engineering**: Create uniform input for time-series models
5. **Visualization**: Generate smooth, consistent time-series plots
6. **Aggregation Preparation**: Align data before windowing operations

### Performance Characteristics

| Strategy | Throughput | Latency (avg) | Use Case |
|----------|------------|---------------|----------|
| Linear | 500K events/sec | 2.0 μs | Continuous metrics |
| Forward Fill | 800K events/sec | 1.25 μs | Discrete states |
| Zero Fill | 1M events/sec | 1.0 μs | Event counts |
| Spline | 50K events/sec | 20 μs | Visualization |

See [normalization/README.md](./src/normalization/README.md) for complete documentation.

## Examples

Run the examples to see the stream processor in action:

```bash
# Basic stream processor demo
cargo run --example stream_processor_demo

# Multi-key sliding window
cargo run --example multi_key_sliding_window

# Event deduplication
cargo run --example event_deduplication_demo
cargo run --example advanced_deduplication

# Time-series normalization
cargo run --example timeseries_normalization_demo
cargo run --example advanced_normalization

# Other examples
cargo run --example kafka_pipeline
cargo run --example aggregation_demo
cargo run --example watermark_demo
```

## Performance

Typical performance characteristics:

- **Latency**: 1-5ms processing time per event
- **Throughput**: 100,000+ events/sec (single key)
- **Throughput**: 1,000,000+ events/sec (multiple keys with parallelization)
- **Memory**: ~1KB per window + aggregation data

## Monitoring

Get processor statistics:

```rust
let stats = processor.stats().await;
println!("Events processed: {}", stats.events_processed);
println!("Windows fired: {}", stats.windows_fired);
println!("Active windows: {}", stats.active_windows);
println!("P99 latency: {:.2}ms", stats.latency_p99_ms);
```

## Architecture

See [STREAM_PROCESSOR.md](./STREAM_PROCESSOR.md) for detailed implementation documentation including:

- Architecture overview
- Implementation details
- Performance characteristics
- Tuning guidelines
- Best practices

## Testing

```bash
# Run all tests
cargo test -p processor

# Run with output
cargo test -p processor -- --nocapture

# Run benchmarks
cargo bench -p processor
```

## Integration

The stream processor integrates with:

- **Kafka**: Source and sink for event streaming
- **State Backends**: Redis, PostgreSQL, Sled for state management
- **Watermark Generators**: For event-time progress tracking
- **Pipeline Framework**: For complex stream processing workflows

## License

Apache-2.0

## Contributing

See [CONTRIBUTING.md](../../CONTRIBUTING.md) for contribution guidelines.