//! The sole `data-server`-knowing layer of `ail-embed`: maps //! `data_server::records::TickParsed` to the single scalar the M3 //! `Tick` record carries (mid price), and lazily streams //! `SymbolChunkIter` chunks into the data-server-free `Kernel` //! (host-side chunk unroll — the adapter owns chunk iteration). use std::sync::Arc; use data_server::records::TickParsed; use data_server::{DataServer, SymbolChunkIter}; use crate::Kernel; /// Mid price — the single scalar the M3 `Tick` record carries. #[inline] pub fn tick_to_px(t: &TickParsed) -> f64 { (t.ask + t.bid) / 2.0 } /// Lazy `Iterator` over a `data-server` tick stream, /// yielding the per-tick mid. Pulls `next_chunk` on demand (the /// adapter owns chunk iteration; chunk boundaries are invisible to /// the fold) so a whole symbol is never materialised. pub struct MidPriceStream { it: SymbolChunkIter, chunk: Option>, idx: usize, } impl MidPriceStream { pub fn new(it: SymbolChunkIter) -> Self { Self { it, chunk: None, idx: 0 } } } impl Iterator for MidPriceStream { type Item = f64; fn next(&mut self) -> Option { loop { if let Some(c) = &self.chunk { if self.idx < c.len() { let px = tick_to_px(&c[self.idx]); self.idx += 1; return Some(px); } } match self.it.next_chunk() { Some(c) => { self.chunk = Some(c); self.idx = 0; } None => return None, } } } } /// Fold up to `max_ticks` of `symbol`'s mid-price stream through the /// M3 kernel. `None` if the symbol has no tick stream. Returns /// `(Σ mid, count)` — the kernel's `acc += px; n += 1` over the /// bounded prefix, identical-order to a host reference fold. pub fn fold_symbol( server: &Arc, symbol: &str, max_ticks: usize, ) -> Option<(f64, i64)> { let it = server.stream_tick(symbol)?; Some(Kernel::new().run(MidPriceStream::new(it).take(max_ticks))) } /// Fold the mid-price stream of `symbol` restricted to the inclusive /// Unix-millisecond window `[from_ms, to_ms]` through the M3 kernel. /// `None` if the symbol has no tick stream in that window. The window /// is the bound (no tick cap): the result is `(Σ mid, count)` over /// exactly the records data-server yields for that window, in stream /// order — identical f64-add sequence to a single-thread host fold of /// the same window, hence bit-exact against it. The windowed sibling /// of `fold_symbol`; same `MidPriceStream` + `Kernel` path, only the /// data-server entry differs (`stream_tick_windowed`). pub fn fold_window( server: &Arc, symbol: &str, from_ms: i64, to_ms: i64, ) -> Option<(f64, i64)> { let it = server.stream_tick_windowed(symbol, Some(from_ms), Some(to_ms))?; Some(Kernel::new().run(MidPriceStream::new(it))) } #[cfg(test)] mod tests { use super::tick_to_px; use data_server::records::TickParsed; /// mid = (ask + bid) / 2. Exact in f64 for these inputs. #[test] fn tick_to_px_is_mid() { let t = TickParsed { time_ms: 0, ask: 2.0, bid: 4.0 }; assert_eq!(tick_to_px(&t), 3.0); } }