AILang/ail-embed/src/adapter.rs

//! 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<Item = f64>` 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<TickParsed>,
    chunk: Option<Arc<[TickParsed]>>,
    idx: usize,
}

impl MidPriceStream {
    pub fn new(it: SymbolChunkIter<TickParsed>) -> Self {
        Self { it, chunk: None, idx: 0 }
    }
}

impl Iterator for MidPriceStream {
    type Item = f64;
    fn next(&mut self) -> Option<f64> {
        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<DataServer>,
    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<DataServer>,
    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);
    }
}