egui_charts/data/
recording.rs

1//! Versioned, length-framed binary recordings. Hosts supply files or any `Read`/`Write` stream.
2
3use super::{MarketEvent, OrderEvent, ReplayEvent, RestingOrder};
4use crate::model::{Level1Quote, OrderBook, OrderBookLevel, OrderSide, Tick, TradeConditions};
5use chrono::{DateTime, Utc};
6use std::io::{self, Cursor, Read, Write};
7
8const MAGIC: &[u8; 8] = b"EGCFLOW2";
9const MAX_RECORD: usize = 16 * 1024 * 1024;
10
11/// Writes a compact binary recording: a version header, then length-framed events with fixed
12/// little-endian numbers, optional-id flags and UTF-8 symbols. Version 2 retains resting order
13/// participant and placement metadata. No runtime or filesystem policy
14/// is imposed; a host can use a buffered file on native targets or bytes in a browser.
15pub struct RecordingWriter<W: Write> {
16    writer: W,
17}
18
19impl<W: Write> RecordingWriter<W> {
20    /// Writes the version header to a new recording.
21    pub fn new(mut writer: W) -> io::Result<Self> {
22        writer.write_all(MAGIC)?;
23        Ok(Self { writer })
24    }
25
26    /// Appends an event without flushing the underlying stream. Invalid data writes nothing.
27    pub fn write(&mut self, event: &ReplayEvent) -> io::Result<()> {
28        if !super::replay::valid(event) || event.symbol.len() > u16::MAX as usize {
29            return Err(invalid());
30        }
31        let mut bytes = Vec::new();
32        bytes.extend_from_slice(&(event.symbol.len() as u16).to_le_bytes());
33        bytes.extend_from_slice(event.symbol.as_bytes());
34        put_time(&mut bytes, event.time);
35        put_u64(&mut bytes, event.sequence);
36        match &event.event {
37            MarketEvent::Trade(t) => {
38                bytes.push(0);
39                put_tick(&mut bytes, t);
40            }
41            MarketEvent::Quote(q) => {
42                bytes.push(1);
43                for value in [q.bid, q.ask, q.bid_size, q.ask_size] {
44                    put_f64(&mut bytes, value);
45                }
46            }
47            MarketEvent::DepthSnapshot { sequence, book }
48            | MarketEvent::DepthDelta { sequence, book } => {
49                bytes.push(
50                    if matches!(event.event, MarketEvent::DepthSnapshot { .. }) {
51                        2
52                    } else {
53                        3
54                    },
55                );
56                put_u64(&mut bytes, *sequence);
57                for side in [&book.bids, &book.asks] {
58                    bytes.extend_from_slice(&(side.len() as u32).to_le_bytes());
59                    for level in side {
60                        put_f64(&mut bytes, level.price);
61                        put_f64(&mut bytes, level.size);
62                        bytes.push(u8::from(level.order_cnt.is_some()));
63                        if let Some(n) = level.order_cnt {
64                            bytes.extend_from_slice(&n.to_le_bytes());
65                        }
66                    }
67                }
68            }
69            MarketEvent::OrderSnapshot { sequence, orders } => {
70                bytes.push(4);
71                put_u64(&mut bytes, *sequence);
72                bytes.extend_from_slice(&(orders.len() as u32).to_le_bytes());
73                for order in orders {
74                    put_order(&mut bytes, order);
75                }
76            }
77            MarketEvent::Order { sequence, event } => {
78                bytes.push(5);
79                put_u64(&mut bytes, *sequence);
80                match event {
81                    OrderEvent::Add(order) => {
82                        bytes.push(0);
83                        put_order(&mut bytes, order);
84                    }
85                    OrderEvent::Modify {
86                        id,
87                        price,
88                        size,
89                        keep_priority,
90                    } => {
91                        bytes.push(1);
92                        put_u64(&mut bytes, *id);
93                        for value in [price, size] {
94                            bytes.push(u8::from(value.is_some()));
95                            if let Some(value) = value {
96                                put_f64(&mut bytes, *value);
97                            }
98                        }
99                        bytes.push(u8::from(*keep_priority));
100                    }
101                    OrderEvent::Cancel { id } => {
102                        bytes.push(2);
103                        put_u64(&mut bytes, *id);
104                    }
105                    OrderEvent::Execute { id, size, trade } => {
106                        bytes.push(3);
107                        put_u64(&mut bytes, *id);
108                        put_f64(&mut bytes, *size);
109                        bytes.push(u8::from(trade.is_some()));
110                        if let Some(trade) = trade {
111                            put_tick(&mut bytes, trade);
112                        }
113                    }
114                }
115            }
116        }
117        if bytes.len() > MAX_RECORD {
118            return Err(invalid());
119        }
120        self.writer.write_all(&(bytes.len() as u32).to_le_bytes())?;
121        self.writer.write_all(&bytes)
122    }
123
124    /// Records a host event stream until it ends or either reading or writing fails. The first
125    /// error is returned unchanged; previously written complete events remain in the stream.
126    /// Run blocking streams outside the UI thread and call [`Self::finish`] to flush storage.
127    pub fn capture(
128        &mut self,
129        events: impl IntoIterator<Item = io::Result<ReplayEvent>>,
130    ) -> io::Result<()> {
131        for event in events {
132            self.write(&event?)?;
133        }
134        Ok(())
135    }
136
137    /// Borrows the underlying writer without flushing buffered storage.
138    pub fn get_ref(&self) -> &W {
139        &self.writer
140    }
141
142    /// Flushes and returns the underlying writer. Propagates disk and stream errors.
143    pub fn finish(mut self) -> io::Result<W> {
144        self.writer.flush()?;
145        Ok(self.writer)
146    }
147}
148
149/// Reads version 1 or 2 binary events, including order placement and participant metadata in
150/// version 2. Limits allocation to 16 MiB per record, rejects unknown
151/// versions, malformed values and truncated records, and distinguishes clean EOF from truncation.
152pub struct RecordingReader<R: Read> {
153    reader: R,
154    order_metadata: bool,
155}
156impl<R: Read> RecordingReader<R> {
157    /// Reads and validates the recording header.
158    pub fn new(mut reader: R) -> io::Result<Self> {
159        let header = read::<8>(&mut reader)?;
160        let order_metadata = match &header {
161            b"EGCFLOW1" => false,
162            MAGIC => true,
163            _ => return Err(invalid()),
164        };
165        Ok(Self {
166            reader,
167            order_metadata,
168        })
169    }
170
171    /// Reads the next event, or `None` only at a complete record boundary.
172    pub fn read(&mut self) -> io::Result<Option<ReplayEvent>> {
173        let mut length = [0; 4];
174        loop {
175            match self.reader.read(&mut length[..1]) {
176                Ok(0) => return Ok(None),
177                Ok(_) => break,
178                Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
179                Err(e) => return Err(e),
180            }
181        }
182        self.reader.read_exact(&mut length[1..])?;
183        let length = u32::from_le_bytes(length) as usize;
184        if length > MAX_RECORD {
185            return Err(invalid());
186        }
187        let mut bytes = vec![0; length];
188        self.reader.read_exact(&mut bytes)?;
189        let mut reader = Cursor::new(bytes.as_slice());
190        let symbol_len = u16::from_le_bytes(read(&mut reader)?) as usize;
191        let mut symbol = vec![0; symbol_len];
192        reader.read_exact(&mut symbol)?;
193        let symbol = String::from_utf8(symbol).map_err(|_| invalid())?;
194        let time = get_time(&mut reader)?;
195        let sequence = get_u64(&mut reader)?;
196        let event = match get_byte(&mut reader)? {
197            0 => MarketEvent::Trade(get_tick(&mut reader)?),
198            1 => MarketEvent::Quote(Level1Quote {
199                ts: time,
200                symbol: symbol.clone(),
201                bid: get_f64(&mut reader)?,
202                ask: get_f64(&mut reader)?,
203                bid_size: get_f64(&mut reader)?,
204                ask_size: get_f64(&mut reader)?,
205            }),
206            kind @ (2 | 3) => {
207                let sequence = get_u64(&mut reader)?;
208                let book = OrderBook {
209                    ts: time,
210                    symbol: symbol.clone(),
211                    bids: get_levels(&mut reader)?,
212                    asks: get_levels(&mut reader)?,
213                };
214                if kind == 2 {
215                    MarketEvent::DepthSnapshot { sequence, book }
216                } else {
217                    MarketEvent::DepthDelta { sequence, book }
218                }
219            }
220            4 => {
221                let sequence = get_u64(&mut reader)?;
222                let count = u32::from_le_bytes(read(&mut reader)?) as usize;
223                if count > super::depth::MAX_LEVELS * 2 {
224                    return Err(invalid());
225                }
226                let orders = (0..count)
227                    .map(|_| get_order(&mut reader, self.order_metadata))
228                    .collect::<io::Result<Vec<_>>>()?;
229                MarketEvent::OrderSnapshot { sequence, orders }
230            }
231            5 => {
232                let sequence = get_u64(&mut reader)?;
233                let event = match get_byte(&mut reader)? {
234                    0 => OrderEvent::Add(get_order(&mut reader, self.order_metadata)?),
235                    1 => OrderEvent::Modify {
236                        id: get_u64(&mut reader)?,
237                        price: get_optional(&mut reader, get_f64)?,
238                        size: get_optional(&mut reader, get_f64)?,
239                        keep_priority: get_bool(&mut reader)?,
240                    },
241                    2 => OrderEvent::Cancel {
242                        id: get_u64(&mut reader)?,
243                    },
244                    3 => OrderEvent::Execute {
245                        id: get_u64(&mut reader)?,
246                        size: get_f64(&mut reader)?,
247                        trade: get_optional(&mut reader, get_tick)?,
248                    },
249                    _ => return Err(invalid()),
250                };
251                MarketEvent::Order { sequence, event }
252            }
253            _ => return Err(invalid()),
254        };
255        let event = ReplayEvent::new(symbol, time, sequence, event);
256        if reader.position() as usize != bytes.len() || !super::replay::valid(&event) {
257            return Err(invalid());
258        }
259        Ok(Some(event))
260    }
261}
262
263fn invalid() -> io::Error {
264    io::Error::new(io::ErrorKind::InvalidData, "invalid market recording")
265}
266fn read<const N: usize>(reader: &mut impl Read) -> io::Result<[u8; N]> {
267    let mut b = [0; N];
268    reader.read_exact(&mut b)?;
269    Ok(b)
270}
271fn put_u64(bytes: &mut Vec<u8>, value: u64) {
272    bytes.extend_from_slice(&value.to_le_bytes());
273}
274fn put_f64(bytes: &mut Vec<u8>, value: f64) {
275    bytes.extend_from_slice(&value.to_le_bytes());
276}
277fn get_u64(reader: &mut impl Read) -> io::Result<u64> {
278    Ok(u64::from_le_bytes(read(reader)?))
279}
280fn get_f64(reader: &mut impl Read) -> io::Result<f64> {
281    Ok(f64::from_le_bytes(read(reader)?))
282}
283fn get_byte(reader: &mut impl Read) -> io::Result<u8> {
284    Ok(read::<1>(reader)?[0])
285}
286fn get_bool(reader: &mut impl Read) -> io::Result<bool> {
287    match get_byte(reader)? {
288        0 => Ok(false),
289        1 => Ok(true),
290        _ => Err(invalid()),
291    }
292}
293fn get_optional<R: Read, T>(
294    reader: &mut R,
295    get: impl FnOnce(&mut R) -> io::Result<T>,
296) -> io::Result<Option<T>> {
297    if get_bool(reader)? {
298        get(reader).map(Some)
299    } else {
300        Ok(None)
301    }
302}
303fn put_time(bytes: &mut Vec<u8>, time: DateTime<Utc>) {
304    bytes.extend_from_slice(&time.timestamp().to_le_bytes());
305    bytes.extend_from_slice(&time.timestamp_subsec_nanos().to_le_bytes());
306}
307fn get_time(reader: &mut impl Read) -> io::Result<DateTime<Utc>> {
308    DateTime::from_timestamp(
309        i64::from_le_bytes(read(reader)?),
310        u32::from_le_bytes(read(reader)?),
311    )
312    .ok_or_else(invalid)
313}
314fn put_tick(bytes: &mut Vec<u8>, trade: &Tick) {
315    put_time(bytes, trade.ts);
316    put_f64(bytes, trade.price);
317    put_f64(bytes, trade.size);
318    bytes.push(match trade.side {
319        None => 0,
320        Some(OrderSide::Buy) => 1,
321        Some(OrderSide::Sell) => 2,
322    });
323    for id in [trade.id, trade.aggressor_order_id, trade.resting_order_id] {
324        bytes.push(u8::from(id.is_some()));
325        if let Some(id) = id {
326            put_u64(bytes, id);
327        }
328    }
329    bytes.extend_from_slice(&trade.conditions.bits().to_le_bytes());
330}
331fn get_tick(reader: &mut impl Read) -> io::Result<Tick> {
332    let time = get_time(reader)?;
333    let price = get_f64(reader)?;
334    let size = get_f64(reader)?;
335    let side = match get_byte(reader)? {
336        0 => None,
337        1 => Some(OrderSide::Buy),
338        2 => Some(OrderSide::Sell),
339        _ => return Err(invalid()),
340    };
341    let mut trade = Tick::new(time, price, size, side);
342    trade.id = get_optional(reader, get_u64)?;
343    trade.aggressor_order_id = get_optional(reader, get_u64)?;
344    trade.resting_order_id = get_optional(reader, get_u64)?;
345    trade.conditions = TradeConditions::from_bits(u16::from_le_bytes(read(reader)?));
346    Ok(trade)
347}
348fn put_order(bytes: &mut Vec<u8>, order: &RestingOrder) {
349    put_u64(bytes, order.id);
350    bytes.push(u8::from(order.side == OrderSide::Sell));
351    put_f64(bytes, order.price);
352    put_f64(bytes, order.size);
353    bytes.push(u8::from(order.participant.is_some()));
354    if let Some(participant) = order.participant {
355        put_u64(bytes, participant);
356    }
357    bytes.push(u8::from(order.placed.is_some()));
358    if let Some(placed) = order.placed {
359        put_time(bytes, placed);
360    }
361}
362fn get_order(reader: &mut impl Read, metadata: bool) -> io::Result<RestingOrder> {
363    let mut order = RestingOrder::new(
364        get_u64(reader)?,
365        if get_bool(reader)? {
366            OrderSide::Sell
367        } else {
368            OrderSide::Buy
369        },
370        get_f64(reader)?,
371        get_f64(reader)?,
372    );
373    if metadata {
374        order.participant = get_optional(reader, get_u64)?;
375        order.placed = get_optional(reader, get_time)?;
376    }
377    Ok(order)
378}
379fn get_levels(reader: &mut impl Read) -> io::Result<Vec<OrderBookLevel>> {
380    let count = u32::from_le_bytes(read(reader)?) as usize;
381    if count > super::depth::MAX_LEVELS {
382        return Err(invalid());
383    }
384    (0..count)
385        .map(|_| {
386            Ok(OrderBookLevel {
387                price: get_f64(reader)?,
388                size: get_f64(reader)?,
389                order_cnt: if get_bool(reader)? {
390                    Some(u32::from_le_bytes(read(reader)?))
391                } else {
392                    None
393                },
394            })
395        })
396        .collect()
397}
398
399#[cfg(test)]
400mod tests {
401    use super::*;
402
403    #[test]
404    fn binary_recording_round_trips_market_data_and_rejects_truncation() {
405        let time = DateTime::from_timestamp(1_700_000_000, 123).unwrap();
406        let trade = Tick::new(time, 101.25, 3.0, Some(OrderSide::Buy))
407            .with_id(7)
408            .with_aggressor_order_id(8)
409            .with_resting_order_id(9)
410            .with_conditions(TradeConditions::AUCTION);
411        let book = OrderBook {
412            ts: time,
413            symbol: "A".into(),
414            bids: vec![OrderBookLevel {
415                price: 100.0,
416                size: 2.0,
417                order_cnt: Some(1),
418            }],
419            asks: vec![],
420        };
421        let order = RestingOrder::new(9, OrderSide::Sell, 101.25, 5.0)
422            .with_participant(42)
423            .with_placed(time - chrono::TimeDelta::seconds(30));
424        let events = vec![
425            MarketEvent::Trade(trade.clone()),
426            MarketEvent::Quote(Level1Quote {
427                ts: time,
428                symbol: "A".into(),
429                bid: 100.0,
430                ask: 101.25,
431                bid_size: 2.0,
432                ask_size: 5.0,
433            }),
434            MarketEvent::DepthSnapshot {
435                sequence: 1,
436                book: book.clone(),
437            },
438            MarketEvent::DepthDelta { sequence: 2, book },
439            MarketEvent::OrderSnapshot {
440                sequence: 3,
441                orders: vec![order.clone()],
442            },
443            MarketEvent::Order {
444                sequence: 4,
445                event: OrderEvent::Add(order),
446            },
447            MarketEvent::Order {
448                sequence: 5,
449                event: OrderEvent::Modify {
450                    id: 9,
451                    price: None,
452                    size: Some(6.0),
453                    keep_priority: false,
454                },
455            },
456            MarketEvent::Order {
457                sequence: 6,
458                event: OrderEvent::Cancel { id: 9 },
459            },
460            MarketEvent::Order {
461                sequence: 7,
462                event: OrderEvent::Execute {
463                    id: 9,
464                    size: 3.0,
465                    trade: Some(trade),
466                },
467            },
468        ];
469        for payload in events {
470            let event = ReplayEvent::new("A", time, 10, payload);
471            let mut writer = RecordingWriter::new(Vec::new()).unwrap();
472            writer.write(&event).unwrap();
473            let bytes = writer.finish().unwrap();
474            let mut reader = RecordingReader::new(bytes.as_slice()).unwrap();
475            let decoded = reader.read().unwrap().unwrap();
476            assert_eq!(decoded, event);
477            let mut again = RecordingWriter::new(Vec::new()).unwrap();
478            again.write(&decoded).unwrap();
479            assert_eq!(again.finish().unwrap(), bytes);
480            assert!(reader.read().unwrap().is_none());
481            for end in 9..bytes.len() {
482                assert!(
483                    RecordingReader::new(&bytes[..end]).unwrap().read().is_err(),
484                    "truncated at {end}"
485                );
486            }
487        }
488        // Version 1 ended each resting order after its size, without metadata flags.
489        let legacy_event = ReplayEvent::new(
490            "A",
491            time,
492            0,
493            MarketEvent::OrderSnapshot {
494                sequence: 0,
495                orders: vec![RestingOrder::new(1, OrderSide::Buy, 100.0, 2.0)],
496            },
497        );
498        let legacy: &[u8] = &[
499            0x45, 0x47, 0x43, 0x46, 0x4c, 0x4f, 0x57, 0x31, 0x3d, 0x00, 0x00, 0x00, 0x01, 0x00,
500            0x41, 0x00, 0xf1, 0x53, 0x65, 0x00, 0x00, 0x00, 0x00, 0x7b, 0x00, 0x00, 0x00, 0x00,
501            0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
502            0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
503            0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x59, 0x40, 0x00, 0x00, 0x00, 0x00, 0x00,
504            0x00, 0x00, 0x40,
505        ];
506        assert_eq!(
507            RecordingReader::new(legacy).unwrap().read().unwrap(),
508            Some(legacy_event)
509        );
510
511        struct FailedStorage;
512        impl Write for FailedStorage {
513            fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
514                Ok(bytes.len())
515            }
516            fn flush(&mut self) -> io::Result<()> {
517                Err(io::Error::new(
518                    io::ErrorKind::StorageFull,
519                    "recording storage full",
520                ))
521            }
522        }
523        assert_eq!(
524            RecordingWriter::new(FailedStorage)
525                .unwrap()
526                .finish()
527                .err()
528                .unwrap()
529                .kind(),
530            io::ErrorKind::StorageFull
531        );
532        assert!(RecordingReader::new(b"EGCFLOW3".as_slice()).is_err());
533        let mut oversized = MAGIC.to_vec();
534        oversized.extend_from_slice(&u32::MAX.to_le_bytes());
535        assert!(
536            RecordingReader::new(oversized.as_slice())
537                .unwrap()
538                .read()
539                .is_err()
540        );
541    }
542}
egui_charts/data/recording.rs

egui_charts/data/
recording.rs
