From 7d750a01390b0dedaa1763e2b300dbc74cf5f100 Mon Sep 17 00:00:00 2001 From: 1816x Date: Thu, 16 Jul 2026 16:43:58 +0000 Subject: [PATCH 1/3] feat(engine): realized volatility, volume and VWAP per window Extend the windowing pass to compute more microstructure metrics alongside order-flow imbalance and persist them to a unified table. - rename ofi.rs -> metrics.rs; OfiBucket -> MetricsBucket now carries trade_count, vwap and realized_volatility - realized volatility is continuous across window boundaries, so the per-window variances sum to the whole tape's realized variance and a single-tick window still records its move away from the prior trade - storage writes one wide `metrics` row per (bucket_start_ns, window_ns) - unit tests assert every field against hand-computed known data --- engine/src/main.rs | 8 +- engine/src/metrics.rs | 240 ++++++++++++++++++++++++++++++++++++++++++ engine/src/ofi.rs | 143 ------------------------- engine/src/storage.rs | 79 ++++++++------ 4 files changed, 291 insertions(+), 179 deletions(-) create mode 100644 engine/src/metrics.rs delete mode 100644 engine/src/ofi.rs diff --git a/engine/src/main.rs b/engine/src/main.rs index d68da5c..62209e0 100644 --- a/engine/src/main.rs +++ b/engine/src/main.rs @@ -1,4 +1,4 @@ -mod ofi; +mod metrics; mod storage; mod tick; @@ -28,14 +28,14 @@ fn main() -> anyhow::Result<()> { let ticks = tick::read_csv(&args.input) .with_context(|| format!("reading ticks from {}", args.input.display()))?; - let buckets = ofi::compute(&ticks, window_ns); + let buckets = metrics::compute(&ticks, window_ns); let mut conn = rusqlite::Connection::open(&args.db) .with_context(|| format!("opening database {}", args.db.display()))?; - storage::write_ofi(&mut conn, window_ns, &buckets)?; + storage::write_metrics(&mut conn, window_ns, &buckets)?; println!( - "{} ticks -> {} OFI buckets (window {}) -> {}", + "{} ticks -> {} metric buckets (window {}) -> {}", ticks.len(), buckets.len(), args.window, diff --git a/engine/src/metrics.rs b/engine/src/metrics.rs new file mode 100644 index 0000000..636f7da --- /dev/null +++ b/engine/src/metrics.rs @@ -0,0 +1,240 @@ +use std::collections::BTreeMap; + +use crate::tick::{Aggressor, Tick}; + +/// Microstructure metrics for one fixed time window. +/// +/// Volumes and trade count come from the ticks that land inside the window; +/// [`vwap`](Self::vwap) is the volume-weighted average price of those ticks; +/// `realized_volatility` is the square root of the summed squared log returns +/// *realizing* inside the window (see [`compute`]). +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct MetricsBucket { + /// Window start, aligned to a multiple of the window length. + pub bucket_start_ns: i64, + pub buy_volume: u64, + pub sell_volume: u64, + pub trade_count: u64, + /// Volume-weighted average price over the window's ticks. + pub vwap: f64, + /// `sqrt(Σ rᵢ²)` of the log returns realizing in this window. + pub realized_volatility: f64, +} + +impl MetricsBucket { + /// Order-flow imbalance: `(buy − sell) / (buy + sell)`, in `[-1, 1]`. + /// + /// Buckets only exist when at least one tick landed in them, so the + /// denominator is never zero. + #[must_use] + pub fn ofi(&self) -> f64 { + let buy = self.buy_volume as f64; + let sell = self.sell_volume as f64; + (buy - sell) / (buy + sell) + } + + /// Total traded volume in the window (`buy + sell`). + #[must_use] + pub fn total_volume(&self) -> u64 { + self.buy_volume + self.sell_volume + } +} + +/// Per-window running totals collected in a single time-ordered pass. +#[derive(Default)] +struct Acc { + buy_volume: u64, + sell_volume: u64, + trade_count: u64, + price_size_sum: f64, + sum_sq_returns: f64, +} + +/// Aggregate a tape into fixed windows of `window_ns`, aligned to the epoch. +/// +/// Input order does not matter — the tape is sorted by timestamp first — and +/// output is sorted by bucket start, skipping windows with no ticks. +/// +/// Realized volatility is **continuous across window boundaries**: the log +/// return between two consecutive ticks is attributed to the window of the +/// *later* tick. Summing each window's realized *variance* +/// (`realized_volatility²`) therefore recovers the whole tape's realized +/// variance, and a window holding a single tick still carries the volatility +/// of its move away from the previous trade. +/// +/// # Panics +/// +/// Panics if `window_ns` is not positive. +#[must_use] +pub fn compute(ticks: &[Tick], window_ns: i64) -> Vec { + assert!(window_ns > 0, "window_ns must be positive"); + + // Log returns need time order; sort a view so the caller's slice is left + // untouched and an out-of-order tape still yields sorted buckets. + let mut ordered: Vec<&Tick> = ticks.iter().collect(); + ordered.sort_by_key(|tick| tick.timestamp_ns); + + let mut buckets: BTreeMap = BTreeMap::new(); + let mut prev_price: Option = None; + for tick in ordered { + let bucket_start_ns = tick.timestamp_ns - tick.timestamp_ns.rem_euclid(window_ns); + let acc = buckets.entry(bucket_start_ns).or_default(); + match tick.aggressor { + Aggressor::Buy => acc.buy_volume += u64::from(tick.size), + Aggressor::Sell => acc.sell_volume += u64::from(tick.size), + } + acc.trade_count += 1; + acc.price_size_sum += tick.price * f64::from(tick.size); + if let Some(prev) = prev_price { + let log_return = (tick.price / prev).ln(); + acc.sum_sq_returns += log_return * log_return; + } + prev_price = Some(tick.price); + } + + buckets + .into_iter() + .map(|(bucket_start_ns, acc)| { + let total_volume = acc.buy_volume + acc.sell_volume; + MetricsBucket { + bucket_start_ns, + buy_volume: acc.buy_volume, + sell_volume: acc.sell_volume, + trade_count: acc.trade_count, + vwap: acc.price_size_sum / total_volume as f64, + realized_volatility: acc.sum_sq_returns.sqrt(), + } + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn tick(timestamp_ns: i64, size: u32, aggressor: Aggressor) -> Tick { + priced_tick(timestamp_ns, 21_400.0, size, aggressor) + } + + fn priced_tick(timestamp_ns: i64, price: f64, size: u32, aggressor: Aggressor) -> Tick { + Tick { + timestamp_ns, + price, + size, + aggressor, + } + } + + #[test] + fn buckets_align_to_window_and_skip_empty_ones() { + let ticks = [ + tick(0, 2, Aggressor::Buy), + tick(500, 1, Aggressor::Sell), + tick(1000, 3, Aggressor::Buy), + // nothing in [2000, 3000) + tick(3999, 4, Aggressor::Sell), + ]; + let buckets = compute(&ticks, 1000); + let shape: Vec<(i64, u64, u64, u64)> = buckets + .iter() + .map(|b| { + ( + b.bucket_start_ns, + b.buy_volume, + b.sell_volume, + b.trade_count, + ) + }) + .collect(); + assert_eq!(shape, vec![(0, 2, 1, 2), (1000, 3, 0, 1), (3000, 0, 4, 1)]); + // A flat-price tape has zero log returns and a vwap equal to that price. + for bucket in &buckets { + assert!(bucket.realized_volatility.abs() < f64::EPSILON); + assert!((bucket.vwap - 21_400.0).abs() < f64::EPSILON); + } + } + + #[test] + fn ofi_is_bounded_and_signed_correctly() { + let bucket = |buy_volume, sell_volume| MetricsBucket { + bucket_start_ns: 0, + buy_volume, + sell_volume, + trade_count: 1, + vwap: 0.0, + realized_volatility: 0.0, + }; + assert!((bucket(5, 5).ofi() - 0.0).abs() < f64::EPSILON); + assert!((bucket(7, 0).ofi() - 1.0).abs() < f64::EPSILON); + assert!((bucket(0, 7).ofi() + 1.0).abs() < f64::EPSILON); + assert!((bucket(2, 1).ofi() - 1.0 / 3.0).abs() < 1e-12); + } + + #[test] + fn unsorted_input_yields_sorted_buckets() { + let ticks = [ + tick(2500, 1, Aggressor::Buy), + tick(100, 1, Aggressor::Sell), + tick(1100, 1, Aggressor::Buy), + ]; + let starts: Vec = compute(&ticks, 1000) + .iter() + .map(|b| b.bucket_start_ns) + .collect(); + assert_eq!(starts, vec![0, 1000, 2000]); + } + + #[test] + fn realized_volatility_is_continuous_and_additive() { + // Two windows of two ticks each. Prices: 100 → 101 (in w0), then + // 100 → 100 (in w1). The 101 → 100 return crosses the boundary and is + // attributed to w1, the window of the later tick. + let ticks = [ + priced_tick(0, 100.0, 1, Aggressor::Buy), + priced_tick(400, 101.0, 1, Aggressor::Buy), + priced_tick(1000, 100.0, 2, Aggressor::Sell), + priced_tick(1500, 100.0, 1, Aggressor::Buy), + ]; + let buckets = compute(&ticks, 1000); + assert_eq!(buckets.len(), 2); + + let step = (101.0_f64 / 100.0).ln(); // magnitude of both non-zero returns + + let w0 = &buckets[0]; + assert_eq!( + (w0.bucket_start_ns, w0.buy_volume, w0.sell_volume), + (0, 2, 0) + ); + assert_eq!(w0.trade_count, 2); + assert!((w0.vwap - 100.5).abs() < 1e-12); // (100·1 + 101·1) / 2 + assert!((w0.realized_volatility - step.abs()).abs() < 1e-12); + + let w1 = &buckets[1]; + assert_eq!( + (w1.bucket_start_ns, w1.buy_volume, w1.sell_volume), + (1000, 1, 2) + ); + assert_eq!(w1.trade_count, 2); + assert!((w1.vwap - 100.0).abs() < 1e-12); // (100·2 + 100·1) / 3 + assert!((w1.realized_volatility - step.abs()).abs() < 1e-12); + + // Additivity: Σ per-window variance == variance of the whole tape. + let per_window_var: f64 = buckets.iter().map(|b| b.realized_volatility.powi(2)).sum(); + let whole_tape_var = step.powi(2) + (-step).powi(2) + 0.0; + assert!((per_window_var - whole_tape_var).abs() < 1e-12); + } + + #[test] + fn single_tick_window_carries_cross_boundary_volatility() { + let ticks = [ + priced_tick(0, 100.0, 1, Aggressor::Buy), + priced_tick(1000, 110.0, 1, Aggressor::Buy), + ]; + let buckets = compute(&ticks, 1000); + // First tick has no predecessor, so its window is flat... + assert!(buckets[0].realized_volatility.abs() < f64::EPSILON); + // ...while the lone tick in the next window still records its 100 → 110 move. + assert_eq!(buckets[1].trade_count, 1); + assert!((buckets[1].realized_volatility - (110.0_f64 / 100.0).ln()).abs() < 1e-12); + } +} diff --git a/engine/src/ofi.rs b/engine/src/ofi.rs deleted file mode 100644 index 0cd2906..0000000 --- a/engine/src/ofi.rs +++ /dev/null @@ -1,143 +0,0 @@ -use std::collections::BTreeMap; - -use crate::tick::{Aggressor, Tick}; - -/// Aggressor volume totals for one fixed time window. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct OfiBucket { - /// Window start, aligned to a multiple of the window length. - pub bucket_start_ns: i64, - pub buy_volume: u64, - pub sell_volume: u64, -} - -impl OfiBucket { - /// Order-flow imbalance: `(buy − sell) / (buy + sell)`, in `[-1, 1]`. - /// - /// Buckets only exist when at least one tick landed in them, so the - /// denominator is never zero. - #[must_use] - pub fn ofi(&self) -> f64 { - let buy = self.buy_volume as f64; - let sell = self.sell_volume as f64; - (buy - sell) / (buy + sell) - } -} - -/// Aggregate a tape into fixed windows of `window_ns`, aligned to the epoch. -/// -/// Input order does not matter; output is sorted by bucket start and skips -/// windows with no ticks. -/// -/// # Panics -/// -/// Panics if `window_ns` is not positive. -#[must_use] -pub fn compute(ticks: &[Tick], window_ns: i64) -> Vec { - assert!(window_ns > 0, "window_ns must be positive"); - let mut volumes: BTreeMap = BTreeMap::new(); - for tick in ticks { - let bucket_start_ns = tick.timestamp_ns - tick.timestamp_ns.rem_euclid(window_ns); - let (buy, sell) = volumes.entry(bucket_start_ns).or_default(); - match tick.aggressor { - Aggressor::Buy => *buy += u64::from(tick.size), - Aggressor::Sell => *sell += u64::from(tick.size), - } - } - volumes - .into_iter() - .map(|(bucket_start_ns, (buy_volume, sell_volume))| OfiBucket { - bucket_start_ns, - buy_volume, - sell_volume, - }) - .collect() -} - -#[cfg(test)] -mod tests { - use super::*; - - fn tick(timestamp_ns: i64, size: u32, aggressor: Aggressor) -> Tick { - Tick { - timestamp_ns, - price: 21400.0, - size, - aggressor, - } - } - - #[test] - fn buckets_align_to_window_and_skip_empty_ones() { - let ticks = [ - tick(0, 2, Aggressor::Buy), - tick(500, 1, Aggressor::Sell), - tick(1000, 3, Aggressor::Buy), - // nothing in [2000, 3000) - tick(3999, 4, Aggressor::Sell), - ]; - let buckets = compute(&ticks, 1000); - assert_eq!( - buckets, - vec![ - OfiBucket { - bucket_start_ns: 0, - buy_volume: 2, - sell_volume: 1 - }, - OfiBucket { - bucket_start_ns: 1000, - buy_volume: 3, - sell_volume: 0 - }, - OfiBucket { - bucket_start_ns: 3000, - buy_volume: 0, - sell_volume: 4 - }, - ] - ); - } - - #[test] - fn ofi_is_bounded_and_signed_correctly() { - let balanced = OfiBucket { - bucket_start_ns: 0, - buy_volume: 5, - sell_volume: 5, - }; - let all_buy = OfiBucket { - bucket_start_ns: 0, - buy_volume: 7, - sell_volume: 0, - }; - let all_sell = OfiBucket { - bucket_start_ns: 0, - buy_volume: 0, - sell_volume: 7, - }; - let mixed = OfiBucket { - bucket_start_ns: 0, - buy_volume: 2, - sell_volume: 1, - }; - assert!((balanced.ofi() - 0.0).abs() < f64::EPSILON); - assert!((all_buy.ofi() - 1.0).abs() < f64::EPSILON); - assert!((all_sell.ofi() + 1.0).abs() < f64::EPSILON); - assert!((mixed.ofi() - 1.0 / 3.0).abs() < 1e-12); - } - - #[test] - fn unsorted_input_yields_sorted_buckets() { - let ticks = [ - tick(2500, 1, Aggressor::Buy), - tick(100, 1, Aggressor::Sell), - tick(1100, 1, Aggressor::Buy), - ]; - let starts: Vec = compute(&ticks, 1000) - .iter() - .map(|b| b.bucket_start_ns) - .collect(); - assert_eq!(starts, vec![0, 1000, 2000]); - } -} diff --git a/engine/src/storage.rs b/engine/src/storage.rs index e513bc9..17f6fd4 100644 --- a/engine/src/storage.rs +++ b/engine/src/storage.rs @@ -1,32 +1,37 @@ use rusqlite::Connection; -use crate::ofi::OfiBucket; +use crate::metrics::MetricsBucket; -/// Persist OFI buckets, replacing any previous run for the same window size. +/// Persist metric buckets, replacing any previous run for the same window size. /// /// The `(bucket_start_ns, window_ns)` primary key makes reruns idempotent and /// lets different window sizes coexist in the same database. -pub fn write_ofi( +pub fn write_metrics( conn: &mut Connection, window_ns: i64, - buckets: &[OfiBucket], + buckets: &[MetricsBucket], ) -> rusqlite::Result<()> { conn.execute_batch( - "CREATE TABLE IF NOT EXISTS ofi ( - bucket_start_ns INTEGER NOT NULL, - window_ns INTEGER NOT NULL, - buy_volume INTEGER NOT NULL, - sell_volume INTEGER NOT NULL, - ofi REAL NOT NULL, + "CREATE TABLE IF NOT EXISTS metrics ( + bucket_start_ns INTEGER NOT NULL, + window_ns INTEGER NOT NULL, + buy_volume INTEGER NOT NULL, + sell_volume INTEGER NOT NULL, + total_volume INTEGER NOT NULL, + trade_count INTEGER NOT NULL, + ofi REAL NOT NULL, + vwap REAL NOT NULL, + realized_volatility REAL NOT NULL, PRIMARY KEY (bucket_start_ns, window_ns) )", )?; let tx = conn.transaction()?; { let mut stmt = tx.prepare( - "INSERT OR REPLACE INTO ofi - (bucket_start_ns, window_ns, buy_volume, sell_volume, ofi) - VALUES (?1, ?2, ?3, ?4, ?5)", + "INSERT OR REPLACE INTO metrics + (bucket_start_ns, window_ns, buy_volume, sell_volume, total_volume, + trade_count, ofi, vwap, realized_volatility) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)", )?; for bucket in buckets { stmt.execute(( @@ -34,7 +39,11 @@ pub fn write_ofi( window_ns, bucket.buy_volume, bucket.sell_volume, + bucket.total_volume(), + bucket.trade_count, bucket.ofi(), + bucket.vwap, + bucket.realized_volatility, ))?; } } @@ -45,34 +54,40 @@ pub fn write_ofi( mod tests { use super::*; + fn bucket(bucket_start_ns: i64, buy_volume: u64, sell_volume: u64) -> MetricsBucket { + MetricsBucket { + bucket_start_ns, + buy_volume, + sell_volume, + trade_count: buy_volume + sell_volume, + vwap: 21_400.0, + realized_volatility: 0.01, + } + } + #[test] fn writes_buckets_and_is_idempotent() { let mut conn = Connection::open_in_memory().unwrap(); - let buckets = [ - OfiBucket { - bucket_start_ns: 0, - buy_volume: 3, - sell_volume: 1, - }, - OfiBucket { - bucket_start_ns: 1000, - buy_volume: 0, - sell_volume: 2, - }, - ]; - write_ofi(&mut conn, 1000, &buckets).unwrap(); - write_ofi(&mut conn, 1000, &buckets).unwrap(); + let buckets = [bucket(0, 3, 1), bucket(1000, 0, 2)]; + write_metrics(&mut conn, 1000, &buckets).unwrap(); + write_metrics(&mut conn, 1000, &buckets).unwrap(); let count: i64 = conn - .query_row("SELECT COUNT(*) FROM ofi", [], |row| row.get(0)) + .query_row("SELECT COUNT(*) FROM metrics", [], |row| row.get(0)) .unwrap(); assert_eq!(count, 2); - let ofi: f64 = conn - .query_row("SELECT ofi FROM ofi WHERE bucket_start_ns = 0", [], |row| { - row.get(0) - }) + let (ofi, total_volume, vwap, trade_count): (f64, i64, f64, i64) = conn + .query_row( + "SELECT ofi, total_volume, vwap, trade_count + FROM metrics WHERE bucket_start_ns = 0", + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) .unwrap(); assert!((ofi - 0.5).abs() < f64::EPSILON); + assert_eq!(total_volume, 4); + assert!((vwap - 21_400.0).abs() < f64::EPSILON); + assert_eq!(trade_count, 4); } } From afe72f4d628b4fb79564681be9a10979b7f0166f Mon Sep 17 00:00:00 2001 From: 1816x Date: Thu, 16 Jul 2026 16:43:59 +0000 Subject: [PATCH 2/3] feat(backtest): serve realized volatility and volume endpoints Read the engine's unified `metrics` table through a shared _query_metrics helper and expose one view per metric family. - /metrics/ofi keeps its Phase 1 response contract - /metrics/volatility returns realized_volatility and trade_count - /metrics/volume returns buy/sell/total volume and VWAP - tests cover projection, limit capping/validation and the missing/empty-database 503s across every endpoint --- backtest/src/backtest/api.py | 35 ++++++++++++++---- backtest/tests/test_api.py | 70 ++++++++++++++++++++++++++---------- 2 files changed, 80 insertions(+), 25 deletions(-) diff --git a/backtest/src/backtest/api.py b/backtest/src/backtest/api.py index 17f197a..027cc03 100644 --- a/backtest/src/backtest/api.py +++ b/backtest/src/backtest/api.py @@ -21,9 +21,12 @@ def _db_path() -> Path: return Path(os.environ.get(DB_PATH_ENV, "metrics.db")) -@app.get("/metrics/ofi") -def get_ofi(limit: int = Query(default=100, ge=1, le=10_000)) -> list[dict[str, Any]]: - """Return the most recent OFI buckets, newest first.""" +def _query_metrics(columns: str, limit: int) -> list[dict[str, Any]]: + """Return `columns` from the newest metric buckets, newest first. + + Shared by every endpoint: they differ only in which columns they project + out of the single `metrics` table the engine writes. + """ path = _db_path() if not path.exists(): raise HTTPException( @@ -34,13 +37,33 @@ def get_ofi(limit: int = Query(default=100, ge=1, le=10_000)) -> list[dict[str, conn.row_factory = sqlite3.Row try: rows = conn.execute( - "SELECT bucket_start_ns, window_ns, buy_volume, sell_volume, ofi" - " FROM ofi ORDER BY bucket_start_ns DESC LIMIT ?", + f"SELECT {columns} FROM metrics ORDER BY bucket_start_ns DESC LIMIT ?", (limit,), ).fetchall() except sqlite3.OperationalError as exc: raise HTTPException( status_code=503, - detail=f"metrics database at '{path}' has no OFI data yet: {exc}", + detail=f"metrics database at '{path}' has no metrics yet: {exc}", ) from exc return [dict(row) for row in rows] + + +@app.get("/metrics/ofi") +def get_ofi(limit: int = Query(default=100, ge=1, le=10_000)) -> list[dict[str, Any]]: + """Return the most recent order-flow-imbalance buckets, newest first.""" + return _query_metrics("bucket_start_ns, window_ns, buy_volume, sell_volume, ofi", limit) + + +@app.get("/metrics/volatility") +def get_volatility(limit: int = Query(default=100, ge=1, le=10_000)) -> list[dict[str, Any]]: + """Return the most recent realized-volatility buckets, newest first.""" + return _query_metrics("bucket_start_ns, window_ns, realized_volatility, trade_count", limit) + + +@app.get("/metrics/volume") +def get_volume(limit: int = Query(default=100, ge=1, le=10_000)) -> list[dict[str, Any]]: + """Return the most recent volume buckets (with VWAP), newest first.""" + return _query_metrics( + "bucket_start_ns, window_ns, buy_volume, sell_volume, total_volume, vwap, trade_count", + limit, + ) diff --git a/backtest/tests/test_api.py b/backtest/tests/test_api.py index 7085601..094786b 100644 --- a/backtest/tests/test_api.py +++ b/backtest/tests/test_api.py @@ -5,11 +5,15 @@ from backtest import api +COLUMNS = ( + "bucket_start_ns, window_ns, buy_volume, sell_volume, total_volume," + " trade_count, ofi, vwap, realized_volatility" +) ROWS = [ - # (bucket_start_ns, window_ns, buy_volume, sell_volume, ofi) - (1_000_000_000, 1_000_000_000, 30, 10, 0.5), - (2_000_000_000, 1_000_000_000, 5, 15, -0.5), - (3_000_000_000, 1_000_000_000, 7, 7, 0.0), + # bucket_start_ns, window_ns, buy, sell, total, count, ofi, vwap, rvol + (1_000_000_000, 1_000_000_000, 30, 10, 40, 5, 0.5, 21_400.0, 0.001), + (2_000_000_000, 1_000_000_000, 5, 15, 20, 4, -0.5, 21_401.0, 0.002), + (3_000_000_000, 1_000_000_000, 7, 7, 14, 3, 0.0, 21_402.0, 0.0), ] @@ -17,18 +21,19 @@ def client(tmp_path, monkeypatch): db = tmp_path / "metrics.db" with sqlite3.connect(db) as conn: - conn.execute( - "CREATE TABLE ofi (bucket_start_ns INTEGER, window_ns INTEGER," - " buy_volume INTEGER, sell_volume INTEGER, ofi REAL)" - ) - conn.executemany("INSERT INTO ofi VALUES (?, ?, ?, ?, ?)", ROWS) + conn.execute(f"CREATE TABLE metrics ({COLUMNS})") + conn.executemany(f"INSERT INTO metrics VALUES ({', '.join(['?'] * len(ROWS[0]))})", ROWS) monkeypatch.setenv(api.DB_PATH_ENV, str(db)) return TestClient(app=api.app) -def test_returns_buckets_newest_first(client): +def test_ofi_returns_buckets_newest_first(client): body = client.get("/metrics/ofi").json() - assert [row["bucket_start_ns"] for row in body] == [3_000_000_000, 2_000_000_000, 1_000_000_000] + assert [row["bucket_start_ns"] for row in body] == [ + 3_000_000_000, + 2_000_000_000, + 1_000_000_000, + ] assert body[1] == { "bucket_start_ns": 2_000_000_000, "window_ns": 1_000_000_000, @@ -38,23 +43,50 @@ def test_returns_buckets_newest_first(client): } -def test_limit_caps_rows(client): - assert len(client.get("/metrics/ofi", params={"limit": 2}).json()) == 2 +def test_volatility_endpoint_projects_expected_columns(client): + body = client.get("/metrics/volatility").json() + assert body[0] == { + "bucket_start_ns": 3_000_000_000, + "window_ns": 1_000_000_000, + "realized_volatility": 0.0, + "trade_count": 3, + } + + +def test_volume_endpoint_projects_expected_columns(client): + body = client.get("/metrics/volume").json() + assert body[2] == { + "bucket_start_ns": 1_000_000_000, + "window_ns": 1_000_000_000, + "buy_volume": 30, + "sell_volume": 10, + "total_volume": 40, + "vwap": 21_400.0, + "trade_count": 5, + } + + +@pytest.mark.parametrize("endpoint", ["ofi", "volatility", "volume"]) +def test_limit_caps_rows(client, endpoint): + assert len(client.get(f"/metrics/{endpoint}", params={"limit": 2}).json()) == 2 -def test_limit_is_validated(client): - assert client.get("/metrics/ofi", params={"limit": 0}).status_code == 422 +@pytest.mark.parametrize("endpoint", ["ofi", "volatility", "volume"]) +def test_limit_is_validated(client, endpoint): + assert client.get(f"/metrics/{endpoint}", params={"limit": 0}).status_code == 422 -def test_missing_database_returns_503(monkeypatch, tmp_path): +@pytest.mark.parametrize("endpoint", ["ofi", "volatility", "volume"]) +def test_missing_database_returns_503(monkeypatch, tmp_path, endpoint): monkeypatch.setenv(api.DB_PATH_ENV, str(tmp_path / "nope.db")) - response = TestClient(app=api.app).get("/metrics/ofi") + response = TestClient(app=api.app).get(f"/metrics/{endpoint}") assert response.status_code == 503 assert "run the engine first" in response.json()["detail"] -def test_empty_database_returns_503(monkeypatch, tmp_path): +@pytest.mark.parametrize("endpoint", ["ofi", "volatility", "volume"]) +def test_empty_database_returns_503(monkeypatch, tmp_path, endpoint): db = tmp_path / "empty.db" sqlite3.connect(db).close() monkeypatch.setenv(api.DB_PATH_ENV, str(db)) - assert TestClient(app=api.app).get("/metrics/ofi").status_code == 503 + assert TestClient(app=api.app).get(f"/metrics/{endpoint}").status_code == 503 From ffefaa05e08e667114b954b290f68df07952d32b Mon Sep 17 00:00:00 2001 From: 1816x Date: Thu, 16 Jul 2026 16:43:59 +0000 Subject: [PATCH 3/3] docs: mark Phase 2 done and record the metrics design decisions Check off Phase 2 in the roadmap, add curl examples for the new volatility and volume endpoints, and document three calls: the unified metrics table, continuous cross-window realized volatility, and folding VWAP into the same pass. --- README.md | 25 +++++++++++++++++++++---- 1 file changed, 21 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 3b05431..8cea1d3 100644 --- a/README.md +++ b/README.md @@ -35,7 +35,7 @@ Early scaffolding. Roadmap: - [x] Phase 0 — Scaffolding: monorepo layout, linters, CI. - [x] Phase 1 — Vertical slice: sample tick CSV → order-flow imbalance → SQLite → API endpoint. -- [ ] Phase 2 — More metrics (realized volatility, volume), tests against known data. +- [x] Phase 2 — More metrics (realized volatility, volume, VWAP), tests against known data. - [ ] Phase 3 — Trade journal model + Claude API behavioral agent. - [ ] Phase 4 — Next.js dashboard. - [ ] Phase 5 — v0.1.0 release with sample-data demo. @@ -49,7 +49,7 @@ trades or amounts. # 1. (Optional) regenerate the sample tape — deterministic, seeded python3 data/generate_ticks.py -# 2. Compute order-flow imbalance per 1s window into metrics.db +# 2. Compute microstructure metrics per 1s window into metrics.db cargo run --manifest-path engine/Cargo.toml --release -- \ --input data/sample_mnq_ticks.csv --db metrics.db --window 1s @@ -57,8 +57,10 @@ cargo run --manifest-path engine/Cargo.toml --release -- \ python3 -m venv .venv && .venv/bin/pip install -e "backtest[dev]" METRICS_DB=metrics.db .venv/bin/uvicorn backtest.api:app -# 4. Query them -curl 'localhost:8000/metrics/ofi?limit=5' +# 4. Query them — one table, three views +curl 'localhost:8000/metrics/ofi?limit=5' # order-flow imbalance +curl 'localhost:8000/metrics/volatility?limit=5' # realized volatility + trade count +curl 'localhost:8000/metrics/volume?limit=5' # buy/sell/total volume + VWAP ``` Run the checks the same way CI does: @@ -95,3 +97,18 @@ calls were made by me — as the project progresses. bugs — sign errors produce equally plausible noise. The generator drifts its buy-pressure between 0.35 and 0.65 so imbalance windows show persistent, verifiable signal. +- **One unified `metrics` table instead of a table per metric** (Phase 2, Claude's + proposal, accepted): every metric shares the same `(bucket_start_ns, window_ns)` key + and is produced in a single windowing pass, so a wide row is the natural shape. The + Python API projects the columns each endpoint needs (`/metrics/ofi`, + `/metrics/volatility`, `/metrics/volume`) out of that one table; the Phase 1 + `/metrics/ofi` contract is unchanged. +- **Realized volatility is continuous across window boundaries** (Phase 2, mine): a + window's realized variance includes the log return from the previous window's last + trade, so summing per-window variances recovers the whole tape's realized variance and + a window with a single tick still carries the volatility of its move. The alternative — + resetting returns at each window edge — silently discards the boundary moves and + understates volatility. +- **VWAP folded into the same pass** (Phase 2, Claude's proposal, accepted): it needs + only the running `Σ price·size` the engine already touches per tick, so it comes almost + for free and gives the dashboard a price anchor alongside the volume figures.