fin_stream/book/
mod.rs

1//! Order book — delta streaming with full reconstruction.
2//!
3//! ## Responsibility
4//! Maintain a live order book per symbol by applying incremental deltas
5//! received from exchange WebSocket feeds. Supports full snapshot reset
6//! and crossed-book detection.
7//!
8//! ## Guarantees
9//! - Deterministic: applying the same delta sequence always yields the same book
10//! - Non-panicking: all mutations return Result
11//! - Thread-safe: OrderBook is Send + Sync via internal RwLock
12
13use crate::error::StreamError;
14use rust_decimal::Decimal;
15use std::collections::BTreeMap;
16
17/// Side of the order book.
18#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
19pub enum BookSide {
20    /// Bid (buy) side.
21    Bid,
22    /// Ask (sell) side.
23    Ask,
24}
25
26/// A single price level in the order book.
27#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
28pub struct PriceLevel {
29    /// Price of this level.
30    pub price: Decimal,
31    /// Resting quantity at this price.
32    pub quantity: Decimal,
33}
34
35impl PriceLevel {
36    /// Construct a price level from a price and resting quantity.
37    pub fn new(price: Decimal, quantity: Decimal) -> Self {
38        Self { price, quantity }
39    }
40}
41
42/// Incremental order book update.
43#[derive(Debug, Clone)]
44pub struct BookDelta {
45    /// Symbol this delta applies to.
46    pub symbol: String,
47    /// Side of the book (bid or ask).
48    pub side: BookSide,
49    /// Price level to update. `quantity == 0` means remove the level.
50    pub price: Decimal,
51    /// New resting quantity at this price. Zero removes the level.
52    pub quantity: Decimal,
53    /// Optional exchange-assigned sequence number for gap detection.
54    pub sequence: Option<u64>,
55}
56
57impl BookDelta {
58    /// Construct a delta without a sequence number.
59    ///
60    /// Use [`BookDelta::with_sequence`] to attach the exchange sequence number
61    /// when available; sequenced deltas enable gap detection.
62    pub fn new(symbol: impl Into<String>, side: BookSide, price: Decimal, quantity: Decimal) -> Self {
63        Self { symbol: symbol.into(), side, price, quantity, sequence: None }
64    }
65
66    /// Attach an exchange sequence number to this delta.
67    pub fn with_sequence(mut self, seq: u64) -> Self {
68        self.sequence = Some(seq);
69        self
70    }
71}
72
73/// Live order book for a single symbol.
74pub struct OrderBook {
75    symbol: String,
76    bids: BTreeMap<Decimal, Decimal>, // price → quantity, descending iteration via neg key
77    asks: BTreeMap<Decimal, Decimal>, // price → quantity, ascending
78    last_sequence: Option<u64>,
79}
80
81impl OrderBook {
82    /// Create an empty order book for the given symbol.
83    pub fn new(symbol: impl Into<String>) -> Self {
84        Self {
85            symbol: symbol.into(),
86            bids: BTreeMap::new(),
87            asks: BTreeMap::new(),
88            last_sequence: None,
89        }
90    }
91
92    /// Apply an incremental delta. quantity == 0 removes the level.
93    pub fn apply(&mut self, delta: BookDelta) -> Result<(), StreamError> {
94        if delta.symbol != self.symbol {
95            return Err(StreamError::BookReconstructionFailed {
96                symbol: self.symbol.clone(),
97                reason: format!("delta symbol '{}' does not match book '{}'", delta.symbol, self.symbol),
98            });
99        }
100        let map = match delta.side {
101            BookSide::Bid => &mut self.bids,
102            BookSide::Ask => &mut self.asks,
103        };
104        if delta.quantity.is_zero() {
105            map.remove(&delta.price);
106        } else {
107            map.insert(delta.price, delta.quantity);
108        }
109        if let Some(seq) = delta.sequence {
110            self.last_sequence = Some(seq);
111        }
112        self.check_crossed()
113    }
114
115    /// Reset the book from a full snapshot.
116    pub fn reset(
117        &mut self,
118        bids: Vec<PriceLevel>,
119        asks: Vec<PriceLevel>,
120    ) -> Result<(), StreamError> {
121        self.bids.clear();
122        self.asks.clear();
123        for lvl in bids {
124            if !lvl.quantity.is_zero() {
125                self.bids.insert(lvl.price, lvl.quantity);
126            }
127        }
128        for lvl in asks {
129            if !lvl.quantity.is_zero() {
130                self.asks.insert(lvl.price, lvl.quantity);
131            }
132        }
133        self.check_crossed()
134    }
135
136    /// Best bid (highest).
137    pub fn best_bid(&self) -> Option<PriceLevel> {
138        self.bids.iter().next_back().map(|(p, q)| PriceLevel::new(*p, *q))
139    }
140
141    /// Best ask (lowest).
142    pub fn best_ask(&self) -> Option<PriceLevel> {
143        self.asks.iter().next().map(|(p, q)| PriceLevel::new(*p, *q))
144    }
145
146    /// Mid price.
147    pub fn mid_price(&self) -> Option<Decimal> {
148        let bid = self.best_bid()?.price;
149        let ask = self.best_ask()?.price;
150        Some((bid + ask) / Decimal::from(2))
151    }
152
153    /// Spread.
154    pub fn spread(&self) -> Option<Decimal> {
155        let bid = self.best_bid()?.price;
156        let ask = self.best_ask()?.price;
157        Some(ask - bid)
158    }
159
160    /// Number of bid levels.
161    pub fn bid_depth(&self) -> usize { self.bids.len() }
162
163    /// Number of ask levels.
164    pub fn ask_depth(&self) -> usize { self.asks.len() }
165
166    /// The symbol this order book tracks.
167    pub fn symbol(&self) -> &str { &self.symbol }
168
169    /// The sequence number of the most recently applied delta, if any.
170    pub fn last_sequence(&self) -> Option<u64> { self.last_sequence }
171
172    /// Top N bids (descending by price).
173    pub fn top_bids(&self, n: usize) -> Vec<PriceLevel> {
174        self.bids.iter().rev().take(n).map(|(p, q)| PriceLevel::new(*p, *q)).collect()
175    }
176
177    /// Top N asks (ascending by price).
178    pub fn top_asks(&self, n: usize) -> Vec<PriceLevel> {
179        self.asks.iter().take(n).map(|(p, q)| PriceLevel::new(*p, *q)).collect()
180    }
181
182    fn check_crossed(&self) -> Result<(), StreamError> {
183        if let (Some(bid), Some(ask)) = (self.best_bid(), self.best_ask()) {
184            if bid.price >= ask.price {
185                return Err(StreamError::BookCrossed {
186                    symbol: self.symbol.clone(),
187                    bid: bid.price.to_string(),
188                    ask: ask.price.to_string(),
189                });
190            }
191        }
192        Ok(())
193    }
194}
195
196#[cfg(test)]
197mod tests {
198    use super::*;
199    use rust_decimal_macros::dec;
200
201    fn book(symbol: &str) -> OrderBook { OrderBook::new(symbol) }
202
203    fn delta(symbol: &str, side: BookSide, price: Decimal, qty: Decimal) -> BookDelta {
204        BookDelta::new(symbol, side, price, qty)
205    }
206
207    #[test]
208    fn test_order_book_apply_bid_level() {
209        let mut b = book("BTC-USD");
210        b.apply(delta("BTC-USD", BookSide::Bid, dec!(50000), dec!(1))).unwrap();
211        assert_eq!(b.best_bid().unwrap().price, dec!(50000));
212    }
213
214    #[test]
215    fn test_order_book_apply_ask_level() {
216        let mut b = book("BTC-USD");
217        b.apply(delta("BTC-USD", BookSide::Ask, dec!(50100), dec!(2))).unwrap();
218        assert_eq!(b.best_ask().unwrap().price, dec!(50100));
219    }
220
221    #[test]
222    fn test_order_book_remove_level_with_zero_qty() {
223        let mut b = book("BTC-USD");
224        b.apply(delta("BTC-USD", BookSide::Bid, dec!(50000), dec!(1))).unwrap();
225        b.apply(delta("BTC-USD", BookSide::Bid, dec!(50000), dec!(0))).unwrap();
226        assert!(b.best_bid().is_none());
227    }
228
229    #[test]
230    fn test_order_book_best_bid_is_highest() {
231        let mut b = book("BTC-USD");
232        b.apply(delta("BTC-USD", BookSide::Bid, dec!(49900), dec!(1))).unwrap();
233        b.apply(delta("BTC-USD", BookSide::Bid, dec!(50000), dec!(2))).unwrap();
234        b.apply(delta("BTC-USD", BookSide::Bid, dec!(49800), dec!(3))).unwrap();
235        assert_eq!(b.best_bid().unwrap().price, dec!(50000));
236    }
237
238    #[test]
239    fn test_order_book_best_ask_is_lowest() {
240        let mut b = book("BTC-USD");
241        b.apply(delta("BTC-USD", BookSide::Ask, dec!(50200), dec!(1))).unwrap();
242        b.apply(delta("BTC-USD", BookSide::Ask, dec!(50100), dec!(2))).unwrap();
243        b.apply(delta("BTC-USD", BookSide::Ask, dec!(50300), dec!(3))).unwrap();
244        assert_eq!(b.best_ask().unwrap().price, dec!(50100));
245    }
246
247    #[test]
248    fn test_order_book_mid_price() {
249        let mut b = book("BTC-USD");
250        b.apply(delta("BTC-USD", BookSide::Bid, dec!(50000), dec!(1))).unwrap();
251        b.apply(delta("BTC-USD", BookSide::Ask, dec!(50100), dec!(1))).unwrap();
252        assert_eq!(b.mid_price().unwrap(), dec!(50050));
253    }
254
255    #[test]
256    fn test_order_book_spread() {
257        let mut b = book("BTC-USD");
258        b.apply(delta("BTC-USD", BookSide::Bid, dec!(50000), dec!(1))).unwrap();
259        b.apply(delta("BTC-USD", BookSide::Ask, dec!(50100), dec!(1))).unwrap();
260        assert_eq!(b.spread().unwrap(), dec!(100));
261    }
262
263    #[test]
264    fn test_order_book_crossed_returns_error() {
265        let mut b = book("BTC-USD");
266        b.apply(delta("BTC-USD", BookSide::Ask, dec!(50000), dec!(1))).unwrap();
267        let result = b.apply(delta("BTC-USD", BookSide::Bid, dec!(50001), dec!(1)));
268        assert!(matches!(result, Err(StreamError::BookCrossed { .. })));
269    }
270
271    #[test]
272    fn test_order_book_wrong_symbol_delta_rejected() {
273        let mut b = book("BTC-USD");
274        let result = b.apply(delta("ETH-USD", BookSide::Bid, dec!(3000), dec!(1)));
275        assert!(matches!(result, Err(StreamError::BookReconstructionFailed { .. })));
276    }
277
278    #[test]
279    fn test_order_book_reset_clears_and_reloads() {
280        let mut b = book("BTC-USD");
281        b.apply(delta("BTC-USD", BookSide::Bid, dec!(49000), dec!(5))).unwrap();
282        b.reset(
283            vec![PriceLevel::new(dec!(50000), dec!(1))],
284            vec![PriceLevel::new(dec!(50100), dec!(1))],
285        ).unwrap();
286        assert_eq!(b.bid_depth(), 1);
287        assert_eq!(b.best_bid().unwrap().price, dec!(50000));
288    }
289
290    #[test]
291    fn test_order_book_reset_ignores_zero_qty_levels() {
292        let mut b = book("BTC-USD");
293        b.reset(
294            vec![
295                PriceLevel::new(dec!(50000), dec!(1)),
296                PriceLevel::new(dec!(49900), dec!(0)),
297            ],
298            vec![PriceLevel::new(dec!(50100), dec!(1))],
299        ).unwrap();
300        assert_eq!(b.bid_depth(), 1);
301    }
302
303    #[test]
304    fn test_order_book_depth_counts() {
305        let mut b = book("BTC-USD");
306        b.apply(delta("BTC-USD", BookSide::Bid, dec!(49900), dec!(1))).unwrap();
307        b.apply(delta("BTC-USD", BookSide::Bid, dec!(49800), dec!(1))).unwrap();
308        b.apply(delta("BTC-USD", BookSide::Ask, dec!(50100), dec!(1))).unwrap();
309        assert_eq!(b.bid_depth(), 2);
310        assert_eq!(b.ask_depth(), 1);
311    }
312
313    #[test]
314    fn test_order_book_top_bids_descending() {
315        let mut b = book("BTC-USD");
316        b.apply(delta("BTC-USD", BookSide::Bid, dec!(49800), dec!(3))).unwrap();
317        b.apply(delta("BTC-USD", BookSide::Bid, dec!(50000), dec!(1))).unwrap();
318        b.apply(delta("BTC-USD", BookSide::Bid, dec!(49900), dec!(2))).unwrap();
319        let top = b.top_bids(2);
320        assert_eq!(top[0].price, dec!(50000));
321        assert_eq!(top[1].price, dec!(49900));
322    }
323
324    #[test]
325    fn test_order_book_top_asks_ascending() {
326        let mut b = book("BTC-USD");
327        b.apply(delta("BTC-USD", BookSide::Ask, dec!(50300), dec!(3))).unwrap();
328        b.apply(delta("BTC-USD", BookSide::Ask, dec!(50100), dec!(1))).unwrap();
329        b.apply(delta("BTC-USD", BookSide::Ask, dec!(50200), dec!(2))).unwrap();
330        let top = b.top_asks(2);
331        assert_eq!(top[0].price, dec!(50100));
332        assert_eq!(top[1].price, dec!(50200));
333    }
334
335    #[test]
336    fn test_book_delta_with_sequence() {
337        let d = BookDelta::new("BTC-USD", BookSide::Bid, dec!(50000), dec!(1))
338            .with_sequence(42);
339        assert_eq!(d.sequence, Some(42));
340    }
341
342    #[test]
343    fn test_order_book_sequence_tracking() {
344        let mut b = book("BTC-USD");
345        b.apply(
346            delta("BTC-USD", BookSide::Bid, dec!(50000), dec!(1)).with_sequence(7)
347        ).unwrap();
348        assert_eq!(b.last_sequence(), Some(7));
349    }
350
351    #[test]
352    fn test_order_book_mid_price_empty_returns_none() {
353        let b = book("BTC-USD");
354        assert!(b.mid_price().is_none());
355    }
356
357    #[test]
358    fn test_price_level_new() {
359        let lvl = PriceLevel::new(dec!(100), dec!(5));
360        assert_eq!(lvl.price, dec!(100));
361        assert_eq!(lvl.quantity, dec!(5));
362    }
363}
fin_stream/book/mod.rs

fin_stream/book/
mod.rs
