rangebar-streaming 5.0.1

Real-time streaming engine for range bar processing
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

# rangebar-streaming

Real-time streaming processor for range bars with bounded memory and circuit breaker pattern.

## Overview

`rangebar-streaming` provides a streaming implementation of the range bar algorithm optimized for real-time tick processing with bounded memory usage. Suitable for live trading systems, WebSocket data ingestion, and continuous market analysis.

## Features

- **Bounded Memory**: Configurable buffer size prevents memory exhaustion
- **Circuit Breaker Pattern**: Fault tolerance with automatic error threshold detection
- **Real-Time Metrics**: Processing statistics for monitoring
- **Async Processing**: Tokio-based async stream processing
- **Graceful Degradation**: Handles errors without cascade failures

## Usage

### Basic Streaming

```rust
use rangebar_streaming::StreamingProcessor;
use futures::stream::StreamExt;

// Create processor with 25 BPS (0.25%) threshold
let mut processor = StreamingProcessor::new(250)?; // v3.0.0: 250 = 25 BPS

// Process async stream
let metrics = processor.process_stream(agg_trade_stream).await?;

println!("Processed {} trades", metrics.trades_processed);
println!("Generated {} bars", metrics.bars_generated);
```

### Custom Configuration

```rust
use rangebar_streaming::{StreamingProcessor, StreamingProcessorConfig};

let config = StreamingProcessorConfig {
    max_buffer_size: 50_000,           // Max 50k trades in buffer
    circuit_breaker_threshold: 100,     // Open circuit after 100 errors
    metrics_interval_secs: 60,          // Report metrics every 60s
    ..Default::default()
};

let mut processor = StreamingProcessor::with_config(250, config)?;
```

### Replay Buffer

For testing and development, use the replay buffer for time-aware playback:

```rust
use rangebar_streaming::replay::ReplayBuffer;

let mut buffer = ReplayBuffer::new(10_000); // 10k trade buffer

// Push trades
for trade in trades {
    buffer.push(trade);
}

// Get trades from specific timestamp
let recent_trades = buffer.get_trades_from(start_timestamp);
```

## Metrics

```rust
pub struct StreamingMetrics {
    pub trades_processed: u64,
    pub bars_generated: u64,
    pub processing_duration: Duration,
    pub throughput_trades_per_sec: f64,
    pub errors_encountered: u64,
    pub circuit_breaker_active: bool,
}
```

## Circuit Breaker

The circuit breaker pattern prevents cascade failures:

1. **Closed State**: Normal operation, all trades processed
2. **Open State**: Error threshold exceeded, processing halted
3. **Half-Open State**: Testing if errors cleared

```rust
// Circuit breaker activates when error count exceeds threshold
if metrics.circuit_breaker_active {
    eprintln!("Circuit breaker active - too many errors");
}
```

## Performance Characteristics

- **Memory Usage**: O(buffer_size) - bounded and configurable
- **Throughput**: Moderate (single-threaded async processing)
- **Latency**: Low (< 1ms per trade in normal conditions)
- **Parallelism**: Single-threaded (use rangebar-batch for multi-threaded)

## Comparison: Streaming vs Batch

| Feature         | Streaming       | Batch                  |
| --------------- | --------------- | ---------------------- |
| Memory Usage    | Bounded         | Unbounded              |
| Throughput      | Moderate        | High                   |
| Parallelism     | Single-threaded | Multi-threaded (Rayon) |
| Real-time       | Yes             | No                     |
| Use Case        | Live trading    | Historical analysis    |
| Circuit Breaker | Yes             | No                     |

## Dependencies

- **rangebar-core** - Core algorithm and types
- **rangebar-providers** - Data fetching
- **tokio** - Async runtime
- **futures** - Stream processing

## Version

Current version: **5.0.0** (modular crate architecture)

## Documentation

- Architecture: `../../docs/ARCHITECTURE.md`
- Examples: `../../examples/interactive/`

## License

See LICENSE file in the repository root.