opendeviationbar-streaming 13.58.2

Real-time streaming engine for open deviation 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

# opendeviationbar-streaming

Real-time streaming engine for open deviation bar construction over Binance WebSocket feeds, with bounded memory, circuit breaker pattern, and checkpoint recovery.

[![crates.io](https://img.shields.io/crates/v/opendeviationbar-streaming.svg)](https://crates.io/crates/opendeviationbar-streaming)
[![docs.rs](https://docs.rs/opendeviationbar-streaming/badge.svg)](https://docs.rs/opendeviationbar-streaming)

## Installation

### From crates.io

```toml
[dependencies]
opendeviationbar-streaming = "12"
```

### From git (latest)

```toml
[dependencies]
opendeviationbar-streaming = { git = "https://github.com/terrylica/opendeviationbar-py", path = "crates/opendeviationbar-streaming" }
```

### With Binance WebSocket integration

```toml
[dependencies]
opendeviationbar-streaming = { version = "12", features = ["binance-integration"] }
```

## Quick Start

### Stream Processing

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

// Create processor: 250 dbps = 0.25% threshold
let mut processor = StreamingProcessor::new(250)?;

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

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

### Live Bar Engine (multiplexed WebSocket)

```rust
use opendeviationbar_streaming::LiveBarEngine;

// Process multiple symbols × thresholds concurrently
let engine = LiveBarEngine::new(
    vec!["BTCUSDT".into(), "ETHUSDT".into()],
    vec![250, 500],
    true, // include microstructure features
);

engine.start();

loop {
    if let Some(bar) = engine.next_bar(timeout_ms: 5000) {
        let symbol = bar.get("_symbol");
        let threshold = bar.get("_threshold");
        println!("{symbol}@{threshold}: close={}", bar.get("close"));
    }
}
```

### Custom Configuration

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

let config = StreamingProcessorConfig {
    max_buffer_size: 50_000,
    circuit_breaker_threshold: 100,
    metrics_interval_secs: 60,
    ..Default::default()
};

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

## Features

| Feature               | Default | Description                                  |
| --------------------- | ------- | -------------------------------------------- |
| `binance-integration` | No      | Binance WebSocket stream support             |
| `stats`               | No      | Rolling statistics (rolling-stats, tdigests) |
| `indicators`          | No      | Technical indicators                         |
| `all`                 | No      | Enable all features                          |

## Architecture

```
Binance WS → LiveBarEngine (tokio, multiplexed)
                    │
                    ├─ OpenDeviationBarProcessor (per symbol × threshold)
                    │  └─ process_single_trade() → Option<CompletedBar>
                    │
                    └─ next_bar() → completed bars to consumer
```

- **Bounded memory**: Configurable buffer size prevents exhaustion
- **Circuit breaker**: Fault tolerance with automatic error threshold detection
- **Graceful shutdown**: `engine.stop()` + `engine.collect_checkpoints()` for recovery

## MSRV

Minimum supported Rust version: **1.90**

## License

MIT