Worked examples

End-to-end recipes covering the integration shapes common to institutional pricing / risk stacks. Every example runs on the public, synchronous kairos::Client.

1. Historical Greeks export

Drive a bounded historical Greeks stream over a date range and write each row to a sink as it arrives.

// Cargo.toml:
//   kairos = { version = "0.1", features = ["historical-mode"] }

use kairos::{Client, GreeksRow};

let client = Client::connect((email, password))?;
let mut writer = GreeksCsv::create("greeks.csv")?;

let sub = client
    .historical(20251103, 20251107)
    .greeks(["QQQ", "AAPL"])
    .on_event(move |row: &GreeksRow| {
        writer.write(row.date, row.ms_of_day, row.delta, row.gamma);
    })?;

// The bounded stream ends after the last row; wait for it, then flush.
while sub.is_active() {
    std::thread::sleep(std::time::Duration::from_millis(50));
}

2. Replay parity regression

// Cargo.toml:
//   kairos = { version = "0.1", features = ["replay-mode"] }

use std::sync::{Arc, Mutex};
use kairos::{Client, OhlcvcRow};

let client = Client::connect((email, password))?;

let actual = Arc::new(Mutex::new(Vec::<OhlcvcRow>::new()));
let sink = Arc::clone(&actual);

let sub = client
    .replay()
    .ohlcvc(["QQQ"])
    .on_event(move |bar: &OhlcvcRow| sink.lock().unwrap().push(bar.clone()))?;

while sub.is_active() {
    std::thread::sleep(std::time::Duration::from_millis(50));
}

let actual = actual.lock().unwrap();
let expected = load_recorded_rows("captures/baseline.parquet")?;
assert_eq!(actual.len(), expected.len(), "row count differs");
for (a, e) in actual.iter().zip(expected.iter()) {
    assert_eq!(a, e, "row differs");
}

3. VPIN with a downstream alert predicate

use kairos::{Client, VpinRow};
use kairos::AlertPredicate;

let client = Client::connect((email, password))?;

// `AlertPredicate::new(closure)` builds a value-typed predicate.
// Evaluate it against each `VpinRow` as it arrives.
let predicate = AlertPredicate::new(|vpin_value: &f64| *vpin_value > 0.35);

let sub = client
    .live()
    .vpin(["SPY"])
    .bucket_volume(50_000)
    .window_size(50)
    .min_buckets_for_emit(30)
    .on_event(move |row: &VpinRow| {
        if predicate.evaluate(&row.vpin) {
            notify_pnl_desk(row.vpin);
        }
    })?;

4. Multi-analytic dashboard

Four live streams driving onto a shared event queue (the canonical institutional-dashboard shape). Each .on_event callback forwards its typed row over an mpsc::sync_channel to the dashboard worker; all four callbacks run on the client's shared delivery worker, so there is no thread per subscription.

use std::sync::mpsc;
use kairos::{Client, GreeksRow, OhlcvcRow, BlockTradeRow, VolumeAnomalyRow};

enum DashboardEvent {
    Ohlcvc(OhlcvcRow),
    Greeks(GreeksRow),
    Block(BlockTradeRow),
    Unusual(VolumeAnomalyRow),
}

let client = Client::connect((email, password))?;
let (tx, rx) = mpsc::sync_channel::<DashboardEvent>(4096);

let bars_tx = tx.clone();
let _bars = client.live().ohlcvc(["QQQ"]).on_event(move |r: &OhlcvcRow| {
    let _ = bars_tx.send(DashboardEvent::Ohlcvc(r.clone()));
})?;

let greeks_tx = tx.clone();
let _greeks = client.live().greeks(["QQQ"]).on_event(move |r: &GreeksRow| {
    let _ = greeks_tx.send(DashboardEvent::Greeks(r.clone()));
})?;

let block_tx = tx.clone();
let _block = client.live().block_trade(["QQQ"]).on_event(move |r: &BlockTradeRow| {
    let _ = block_tx.send(DashboardEvent::Block(r.clone()));
})?;

let _unusual = client.live().volume_anomaly(["QQQ"]).on_event(move |r: &VolumeAnomalyRow| {
    let _ = tx.send(DashboardEvent::Unusual(r.clone()));
})?;

while let Ok(event) = rx.recv() {
    dashboard.push(event);
}

Each subscription's handle (_bars, _greeks, …) keeps its stream alive for the life of the dashboard; bind them to named locals and call unsubscribe() on shutdown to stop delivery.