From 7f3756a3951aa88d82d8906a1aefe2496765b8ad Mon Sep 17 00:00:00 2001 From: Brummel Date: Sun, 28 Jun 2026 16:51:48 +0200 Subject: [PATCH] =?UTF-8?q?feat(0082):=20cost-graph=20composition=20?= =?UTF-8?q?=E2=80=94=20VolSlippageCost=20+=20CostSum=20net-R=20aggregate?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cycle 2 of the "Cost-model graph (in R)" milestone (#148): the milestone's real architectural claim — the cost graph composes. Two cost nodes now sum into one net-R curve while summarize_r and the net_r_equity tap stay structurally unchanged. - VolSlippageCost (aura-std): a second, STATE-DEPENDENT cost node; per-trade charge = slip_vol_mult * volatility / |entry-stop|, in R. The vol is an independent short-horizon realized range (SLIP_VOL_LENGTH=5, distinct from the stop's EWMA-3) — scaling slippage by the stop's own vol would collapse cost-in-R to a constant (indistinguishable from ConstantCost). - CostSum (aura-std): the cost-graph OUTPUT node — sums N cost nodes' 3-field cost-in-R records per-field into one aggregate. summarize_r and net_r_equity read the aggregate (one home for cost; n=1 is the identity, so the cost path is uniform). A future CostNode trait (deferred) will unify the cost-triple the two producer nodes currently restate by-convention. - Run path: --cost-per-trade and --slip-vol-mult combine, their costs summing into the net-R curve; a hoisted vol proxy (RollingMax-RollingMin) keeps the single feed. Run-path-scoped; sweep/walkforward/mc pass None. Co-temporality contract (the load-bearing design decision; corrected from the signed spec after the implement-loop correctly BLOCKED Task 3 on it). summarize_r positional-joins cost[i] <-> record[i], so the cost stream must be co-temporal 1:1 with the PM record. A cost node is therefore gated ONLY by the PM geometry (closed/open/entry/stop); a not-yet-warm state input (the vol proxy warms later than PM) contributes 0 cost that cycle rather than withholding and desyncing the stream. This makes co-temporality structural + warmup-independent, preserves the C18 golden, and generalizes to any future cost factor. The rejected alternative (a key-join in summarize_r) would have moved the golden and pushed cost-graph logic into the post-run fold. Recorded on #148. Tests: VolSlippageCost + CostSum unit sets (incl. the co-temporal-zero-during- warmup case); the CLI composition run (both flags -> net_both < net_flat, net_r_equity persisted); node-level EXACT additive composition (net_both == net_flat + net_vol - gross over the real nodes), the aggregate net_r_equity == post-run net total, and CostSum(1) identity. Full workspace suite green; clippy -D warnings clean; the no-cost C18 golden byte-identical (the regression floor). Deferred (later cycles of this milestone): the general CostNode trait + cost-graph composite-builder (now justified by two concrete nodes), the conviction-weighting R-aggregation axis, and cost on the reduce-mode sweep path. Spec + plan amended to the corrected contract; both are cycle ephemera (git rm at cycle close). refs #148 --- crates/aura-cli/src/main.rs | 166 +++++++++++++---- crates/aura-cli/tests/cli_run.rs | 25 +++ crates/aura-engine/tests/stage1_r_e2e.rs | 134 +++++++++++++- crates/aura-std/src/cost_sum.rs | 162 +++++++++++++++++ crates/aura-std/src/lib.rs | 4 + crates/aura-std/src/vol_slippage_cost.rs | 222 +++++++++++++++++++++++ docs/plans/0082-vol-slippage-cost.md | 36 ++-- docs/specs/0082-vol-slippage-cost.md | 35 +++- 8 files changed, 725 insertions(+), 59 deletions(-) create mode 100644 crates/aura-std/src/cost_sum.rs create mode 100644 crates/aura-std/src/vol_slippage_cost.rs diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index 1bdf36e..d114efa 100644 --- a/crates/aura-cli/src/main.rs +++ b/crates/aura-cli/src/main.rs @@ -29,9 +29,9 @@ use aura_registry::{ FamilyMember, Generalization, NameKind, PlateauMode, Registry, RunTraces, TraceStore, WriteKind, }; use aura_std::{ - Add, Bias, ConstantCost, Delay, Ema, GatedRecorder, Gt, Latch, LinComb, LongOnly, Mul, Recorder, - RollingMax, RollingMin, SeriesReducer, SimBroker, Sma, Sqrt, Sub, PM_FIELD_NAMES, - PM_RECORD_KINDS, + Add, Bias, ConstantCost, CostSum, Delay, Ema, GatedRecorder, Gt, Latch, LinComb, LongOnly, Mul, + Recorder, RollingMax, RollingMin, SeriesReducer, SimBroker, Sma, Sqrt, Sub, VolSlippageCost, + PM_FIELD_NAMES, PM_RECORD_KINDS, }; use std::sync::mpsc::{self, Receiver}; use std::sync::LazyLock; @@ -2565,6 +2565,41 @@ const STAGE1_R_STOP_LENGTH: i64 = 3; /// `StopRule::Vol` and its manifest record, like [`STAGE1_R_STOP_LENGTH`]. const STAGE1_R_STOP_K: f64 = 2.0; +/// Short-horizon realized-range window for vol-scaled slippage. Deliberately +/// distinct from `STAGE1_R_STOP_LENGTH` (3): scaling slippage by the stop's own +/// vol would collapse cost-in-R to a constant (spec 0082). Short enough to warm +/// within the synthetic smoke fixture so the run path exercises non-zero slippage. +const SLIP_VOL_LENGTH: i64 = 5; + +/// The maximum number of cost nodes the run-path cost graph wires (flat cost + +/// vol slippage). Sizes the interned `CostSum` input-port names below. +const MAX_RUN_COST_NODES: usize = 2; + +/// The 3-field cost-in-R record order — a lockstep contract with each cost node's +/// output schema and `CostSum`'s inputs (and `summarize_r`'s `cost_col`). +const COST_FIELDS: [&str; 3] = ["cost_in_r", "cum_cost_in_r", "open_cost_in_r"]; + +/// Interned `cost[k].` `CostSum` input-port names, built once. Same +/// `&'static str`-from-a-static rationale as `COL_PORTS`: `GraphBuilder::input` +/// wants `&'static str`, so the names live in a `static` rather than being +/// `format!(...).leak()`ed per build. +static COST_SUM_PORTS: LazyLock> = LazyLock::new(|| { + let mut v = Vec::with_capacity(MAX_RUN_COST_NODES * 3); + for k in 0..MAX_RUN_COST_NODES { + for field in COST_FIELDS { + v.push(format!("cost[{k}].{field}")); + } + } + v +}); + +/// Which cost nodes the run-path cost graph builds. At least one field is `Some` +/// (the carrier `Option` is `None` when no cost flag was given). +struct CostConfig { + const_cost: Option, // --cost-per-trade + slip_vol_mult: Option, // --slip-vol-mult +} + /// Which data a `run` drives a harness on: the built-in synthetic stream, or real M1 /// close bars for a vetted symbol over an optional window. enum RunData { @@ -2621,7 +2656,7 @@ fn stage1_r_graph( slow_len: Option, stop_open: bool, reduce: bool, - cost: Option<(f64, mpsc::Sender<(Timestamp, Vec)>, mpsc::Sender<(Timestamp, Vec)>)>, + cost: Option<(CostConfig, mpsc::Sender<(Timestamp, Vec)>, mpsc::Sender<(Timestamp, Vec)>)>, ) -> Composite { let mut g = GraphBuilder::new("stage1_r"); // SMA-cross signal → Bias (the same signal the pip sample uses). @@ -2669,11 +2704,32 @@ fn stage1_r_graph( } else { g.add(Recorder::builder(PM_RECORD_KINDS.to_vec(), Firing::Any, tx_r)) }; + // Hoisted above the single main feed: the short-horizon vol proxy iff a + // vol-slippage cost is actually wired (run path, non-reduce), so its `price` + // inputs join the one `price_targets` array (no second feed call). + let vol_proxy = match &cost { + Some((cfg, _, _)) if !reduce && cfg.slip_vol_mult.is_some() => { + let vhi = g.add(RollingMax::builder().named("slip_vol_hi").bind("length", Scalar::i64(SLIP_VOL_LENGTH))); + let vlo = g.add(RollingMin::builder().named("slip_vol_lo").bind("length", Scalar::i64(SLIP_VOL_LENGTH))); + let vrange = g.add(Sub::builder().named("slip_vol_range")); + g.connect(vhi.output("value"), vrange.input("lhs")); + g.connect(vlo.output("value"), vrange.input("rhs")); + Some((vhi, vlo, vrange)) + } + _ => None, + }; let price = g.source_role("price", ScalarKind::F64); - g.feed( - price, - [fast.input("series"), slow.input("series"), broker.input("price"), exec.input("price")], - ); + let mut price_targets = vec![ + fast.input("series"), + slow.input("series"), + broker.input("price"), + exec.input("price"), + ]; + if let Some((vhi, vlo, _)) = vol_proxy { + price_targets.push(vhi.input("series")); + price_targets.push(vlo.input("series")); + } + g.feed(price, price_targets); g.connect(fast.output("value"), spread.input("lhs")); g.connect(slow.output("value"), spread.input("rhs")); g.connect(spread.output("value"), exposure.input("signal")); @@ -2695,15 +2751,39 @@ fn stage1_r_graph( g.connect(exec.output("cum_realized_r"), r_equity.input("term[0]")); g.connect(exec.output("unrealized_r"), r_equity.input("term[1]")); g.connect(r_equity.output("value"), req.input("col[0]")); - if let Some((cost_per_trade, tx_net, tx_cost)) = cost { - let cost_node = g.add( - ConstantCost::builder().bind("cost_per_trade", Scalar::f64(cost_per_trade)), - ); - g.connect(exec.output("closed_this_cycle"), cost_node.input("closed")); - g.connect(exec.output("open"), cost_node.input("open")); - g.connect(exec.output("entry_price"), cost_node.input("entry_price")); - g.connect(exec.output("stop_price"), cost_node.input("stop_price")); - // net_r_equity = cum_realized_r + unrealized_r - cum_cost_in_r - open_cost_in_r + if let Some((cfg, tx_net, tx_cost)) = cost { + let n = cfg.const_cost.is_some() as usize + cfg.slip_vol_mult.is_some() as usize; + let agg = g.add(CostSum::builder(n)); + let mut slot = 0usize; + + if let Some(cpt) = cfg.const_cost { + let cc = g.add(ConstantCost::builder().bind("cost_per_trade", Scalar::f64(cpt))); + g.connect(exec.output("closed_this_cycle"), cc.input("closed")); + g.connect(exec.output("open"), cc.input("open")); + g.connect(exec.output("entry_price"), cc.input("entry_price")); + g.connect(exec.output("stop_price"), cc.input("stop_price")); + for (f, field) in COST_FIELDS.iter().copied().enumerate() { + g.connect(cc.output(field), agg.input(COST_SUM_PORTS[slot * 3 + f].as_str())); + } + slot += 1; + } + + if let Some(svm) = cfg.slip_vol_mult { + let (_, _, vrange) = vol_proxy.expect("vol proxy is built whenever slip_vol_mult is set"); + let vs = g.add(VolSlippageCost::builder().bind("slip_vol_mult", Scalar::f64(svm))); + g.connect(exec.output("closed_this_cycle"), vs.input("closed")); + g.connect(exec.output("open"), vs.input("open")); + g.connect(exec.output("entry_price"), vs.input("entry_price")); + g.connect(exec.output("stop_price"), vs.input("stop_price")); + g.connect(vrange.output("value"), vs.input("volatility")); + for (f, field) in COST_FIELDS.iter().copied().enumerate() { + g.connect(vs.output(field), agg.input(COST_SUM_PORTS[slot * 3 + f].as_str())); + } + slot += 1; + } + debug_assert_eq!(slot, n); + + // net_r_equity = cum_realized_r + unrealized_r - Σcum_cost_in_r - Σopen_cost_in_r let net_eq = g.add( LinComb::builder(4) .bind("weights[0]", Scalar::f64(1.0)) @@ -2713,22 +2793,20 @@ fn stage1_r_graph( ); g.connect(exec.output("cum_realized_r"), net_eq.input("term[0]")); g.connect(exec.output("unrealized_r"), net_eq.input("term[1]")); - g.connect(cost_node.output("cum_cost_in_r"), net_eq.input("term[2]")); - g.connect(cost_node.output("open_cost_in_r"), net_eq.input("term[3]")); + g.connect(agg.output("cum_cost_in_r"), net_eq.input("term[2]")); + g.connect(agg.output("open_cost_in_r"), net_eq.input("term[3]")); let net_rec = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_net)); g.connect(net_eq.output("value"), net_rec.input("col[0]")); - // The cost stream `summarize_r` folds: the `ConstantCost` node's full - // [cost_in_r, cum_cost_in_r, open_cost_in_r] schema. `summarize_r`'s `cost_col` - // reads col 0 (per-close) and col 2 (window-end open) — the lockstep contract - // with aura-std's ConstantCost output — so the stream is recorded 3-wide. + + // The aggregate cost record summarize_r folds (col 0 per-close, col 2 window-end). let cost_rec = g.add(Recorder::builder( vec![ScalarKind::F64, ScalarKind::F64, ScalarKind::F64], Firing::Any, tx_cost, )); - g.connect(cost_node.output("cost_in_r"), cost_rec.input("col[0]")); - g.connect(cost_node.output("cum_cost_in_r"), cost_rec.input("col[1]")); - g.connect(cost_node.output("open_cost_in_r"), cost_rec.input("col[2]")); + g.connect(agg.output("cost_in_r"), cost_rec.input("col[0]")); + g.connect(agg.output("cum_cost_in_r"), cost_rec.input("col[1]")); + g.connect(agg.output("open_cost_in_r"), cost_rec.input("col[2]")); } } g.build().expect("stage1_r wiring resolves") @@ -2953,7 +3031,12 @@ fn stage1_meanrev_graph( /// round-trip cost per trade (price units): `Some(c)` wires the `ConstantCost` node and /// the `net_r_equity` tap and folds the cost stream into `net_expectancy_r`; `None` is the /// frictionless gross-R baseline (net == gross, no extra tap, byte-unchanged). -fn run_stage1_r(data: RunData, trace: Option<&str>, cost: Option) -> RunReport { +fn run_stage1_r( + data: RunData, + trace: Option<&str>, + const_cost: Option, + slip_vol_mult: Option, +) -> RunReport { if let Some(n) = trace && let Err(e) = TraceStore::open("runs").ensure_name_free(n, WriteKind::Run) { @@ -2966,7 +3049,11 @@ fn run_stage1_r(data: RunData, trace: Option<&str>, cost: Option) -> RunRep let (tx_req, rx_req) = mpsc::channel(); let (tx_net, rx_net) = mpsc::channel(); let (tx_cost, rx_cost) = mpsc::channel(); - let cost_bundle = cost.map(|c| (c, tx_net, tx_cost)); + let cost_bundle = if const_cost.is_some() || slip_vol_mult.is_some() { + Some((CostConfig { const_cost, slip_vol_mult }, tx_net, tx_cost)) + } else { + None + }; let flat = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, Some(2), Some(4), false, false, cost_bundle) .compile_with_params(&[]) .expect("valid stage1-r blueprint"); @@ -3061,16 +3148,17 @@ struct RunArgs { data: RunData, trace: Option, cost: Option, + slip_vol_mult: Option, } /// Parse the `run` tail: `[--harness ] [--real [--from ] [--to ]] -/// [--trace ] [--cost-per-trade ]` in any order, each flag at most once. +/// [--trace ] [--cost-per-trade ] [--slip-vol-mult ]` in any order, each flag at most once. /// `--harness sma` is the default (`--harness macd` / `--harness stage1-r` select the /// others); `--from`/`--to` are legal only with `--real`. Pure (no I/O, no exit) so the /// grammar is unit-testable; `main` does the side effects. fn parse_run_args(rest: &[&str]) -> Result { let usage = || { - "usage: aura run [--harness ] [--real [--from ] [--to ]] [--trace ] [--cost-per-trade ]" + "usage: aura run [--harness ] [--real [--from ] [--to ]] [--trace ] [--cost-per-trade ] [--slip-vol-mult ]" .to_string() }; let mut harness: Option = None; @@ -3079,6 +3167,7 @@ fn parse_run_args(rest: &[&str]) -> Result { let mut to: Option = None; let mut trace: Option = None; let mut cost: Option = None; + let mut slip_vol_mult: Option = None; let mut tail = rest; while let Some((flag, t)) = tail.split_first() { match *flag { @@ -3120,6 +3209,15 @@ fn parse_run_args(rest: &[&str]) -> Result { cost = Some(value.parse().map_err(|_| usage())?); tail = t; } + "--slip-vol-mult" if slip_vol_mult.is_none() => { + let (value, t) = t.split_first().ok_or_else(usage)?; + let v: f64 = value.parse().map_err(|_| usage())?; + if v < 0.0 { + return Err(usage()); + } + slip_vol_mult = Some(v); + tail = t; + } _ => return Err(usage()), } } @@ -3132,7 +3230,7 @@ fn parse_run_args(rest: &[&str]) -> Result { Some(s) => RunData::Real { symbol: s, from, to }, None => RunData::Synthetic, }; - Ok(RunArgs { harness, data, trace, cost }) + Ok(RunArgs { harness, data, trace, cost, slip_vol_mult }) } /// Route a parsed `run` invocation to its harness handler. The compile-time `match` IS @@ -3142,7 +3240,7 @@ fn run_dispatch(args: RunArgs) -> Result { Ok(match (args.harness, args.data) { (HarnessKind::Sma, RunData::Synthetic) => run_sample(trace), (HarnessKind::Macd, RunData::Synthetic) => run_macd(trace), - (HarnessKind::Stage1R, data) => run_stage1_r(data, trace, args.cost), + (HarnessKind::Stage1R, data) => run_stage1_r(data, trace, args.cost, args.slip_vol_mult), (HarnessKind::Sma, RunData::Real { symbol, from, to }) => { run_sample_real(&symbol, from, to, trace) } @@ -3153,7 +3251,7 @@ fn run_dispatch(args: RunArgs) -> Result { } const USAGE: &str = - "usage: aura run [--harness ] [--real [--from ] [--to ]] [--trace ] [--cost-per-trade ] | aura chart [--tap ] [--panels] | aura graph | aura sweep [--strategy ] [--real [--from ] [--to ]] [--name |--trace ] | aura mc [--name |--trace ] | aura mc --strategy stage1-r [--real [--from ] [--to ]] [--fast ] [--slow ] [--stop-length ] [--stop-k ] [--block-len ] [--resamples ] [--seed ] | aura walkforward [--strategy ] [--real [--from ] [--to ]] [--name |--trace ] [--fast ] [--slow ] [--stop-length ] [--stop-k ] [--select ] | aura generalize [--strategy stage1-r] --real --fast --slow --stop-length --stop-k [--from ] [--to ] [--metric ] [--name ] | aura runs families | aura runs family [rank ]"; + "usage: aura run [--harness ] [--real [--from ] [--to ]] [--trace ] [--cost-per-trade ] [--slip-vol-mult ] | aura chart [--tap ] [--panels] | aura graph | aura sweep [--strategy ] [--real [--from ] [--to ]] [--name |--trace ] | aura mc [--name |--trace ] | aura mc --strategy stage1-r [--real [--from ] [--to ]] [--fast ] [--slow ] [--stop-length ] [--stop-k ] [--block-len ] [--resamples ] [--seed ] | aura walkforward [--strategy ] [--real [--from ] [--to ]] [--name |--trace ] [--fast ] [--slow ] [--stop-length ] [--stop-k ] [--select ] | aura generalize [--strategy stage1-r] --real --fast --slow --stop-length --stop-k [--from ] [--to ] [--metric ] [--name ] | aura runs families | aura runs family [rank ]"; fn main() { // Restore the default SIGPIPE disposition. Rust's runtime sets SIGPIPE to SIG_IGN @@ -4370,7 +4468,7 @@ mod tests { fn run_stage1_r_synthetic_folds_an_r_block() { // the stage1-r harness scores the SMA-cross signal in R: one shell-callable run // yields a RunReport whose metrics.r is Some with a finite SQN and >= 1 trade. - let report = run_stage1_r(RunData::Synthetic, None, None); + let report = run_stage1_r(RunData::Synthetic, None, None, None); let r = report.metrics.r.as_ref().expect("stage1-r run must populate metrics.r"); assert!(r.n_trades >= 1, "expected >= 1 trade, got {}", r.n_trades); assert!(r.sqn.is_finite(), "SQN must be finite, got {}", r.sqn); diff --git a/crates/aura-cli/tests/cli_run.rs b/crates/aura-cli/tests/cli_run.rs index 2e8bf7a..d967280 100644 --- a/crates/aura-cli/tests/cli_run.rs +++ b/crates/aura-cli/tests/cli_run.rs @@ -1646,6 +1646,31 @@ fn stage1_r_cost_run_persists_net_r_equity_and_charges_cost() { let _ = std::fs::remove_dir_all(&dir); } +/// Property (spec 0082, the cost-graph composition headline): `--cost-per-trade` +/// and `--slip-vol-mult` set together compose — both cost nodes sum into one +/// net-R curve. The combined net is strictly below the flat-cost-only net (the +/// vol-slippage node bites on top), and the `net_r_equity` trace persists. +#[test] +fn stage1_r_both_costs_compose_net_below_each_alone() { + let dir = temp_cwd("stage1-r-compose"); + let run_net = |args: &[&str], trace: &str| -> f64 { + let mut full = vec!["run", "--harness", "stage1-r"]; + full.extend_from_slice(args); + full.extend_from_slice(&["--trace", trace]); + let run = Command::new(BIN).current_dir(&dir).args(&full).output().unwrap(); + assert!(run.status.success(), "exit: {:?}; stderr: {}", run.status, + String::from_utf8_lossy(&run.stderr)); + let s = String::from_utf8(run.stdout).expect("utf-8 stdout"); + let v: serde_json::Value = serde_json::from_str(s.trim()).unwrap(); + v["metrics"]["r"]["net_expectancy_r"].as_f64().unwrap() + }; + let net_flat = run_net(&["--cost-per-trade", "2"], "flat"); + let net_both = run_net(&["--cost-per-trade", "2", "--slip-vol-mult", "0.5"], "both"); + assert!(dir.join("runs/traces/both/net_r_equity.json").exists(), "net_r_equity persisted"); + assert!(net_both < net_flat, "composed cost bites more: net_both {net_both} < net_flat {net_flat}"); + let _ = std::fs::remove_dir_all(&dir); +} + /// Property (the spec's dual-yardstick headline): `aura run --harness stage1-r` scores /// ONE signal in BOTH yardsticks at once — the single emitted `RunReport` carries the pip /// `metrics.total_pips` (off the SimBroker branch) AND the nested `r` block (off the diff --git a/crates/aura-engine/tests/stage1_r_e2e.rs b/crates/aura-engine/tests/stage1_r_e2e.rs index fc0eabe..bc0fc4a 100644 --- a/crates/aura-engine/tests/stage1_r_e2e.rs +++ b/crates/aura-engine/tests/stage1_r_e2e.rs @@ -27,7 +27,10 @@ use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp}; use aura_engine::{PositionAction, RMetrics, derive_position_events, summarize_r}; -use aura_std::{ConstantCost, FixedStop, PM_FIELD_NAMES, PM_RECORD_KINDS, PM_WIDTH, PositionManagement}; +use aura_std::{ + ConstantCost, CostSum, FixedStop, PM_FIELD_NAMES, PM_RECORD_KINDS, PM_WIDTH, + PositionManagement, VolSlippageCost, +}; // The indices `report.rs::summarize_r` reads the dense record by. Re-declared here // (the consumer's `r_col` module is private to `aura-engine`) so the guard test can @@ -333,6 +336,135 @@ fn net_r_equity_final_sample_agrees_with_summarize_r_net_total() { ); } +/// Drive the REAL `aura_std::VolSlippageCost(k)` node over a recorded PM `ledger` +/// with a constant `vol` per cycle — the run-path's vol-slippage producer. Returns +/// the node's 3-wide `[cost_in_r, cum_cost_in_r, open_cost_in_r]` rows, co-temporal +/// 1:1 with `ledger`. +fn vol_slippage_node_stream( + ledger: &[(Timestamp, Vec)], + k: f64, + vol: f64, +) -> Vec<(Timestamp, Vec)> { + let mut node = VolSlippageCost::new(k); + ledger + .iter() + .map(|(ts, row)| { + let mut cols = vec![ + AnyColumn::with_capacity(ScalarKind::Bool, 1), // closed + AnyColumn::with_capacity(ScalarKind::Bool, 1), // open + AnyColumn::with_capacity(ScalarKind::F64, 1), // entry + AnyColumn::with_capacity(ScalarKind::F64, 1), // stop + AnyColumn::with_capacity(ScalarKind::F64, 1), // volatility + ]; + cols[0].push(Scalar::bool(row[CLOSED].as_bool())).unwrap(); + cols[1].push(Scalar::bool(row[OPEN].as_bool())).unwrap(); + cols[2].push(Scalar::f64(row[ENTRY_PRICE].as_f64())).unwrap(); + cols[3].push(Scalar::f64(row[STOP_PRICE].as_f64())).unwrap(); + cols[4].push(Scalar::f64(vol)).unwrap(); + let out = node.eval(Ctx::new(&cols, *ts)).expect("cost row co-temporal with PM record"); + (*ts, out.iter().map(|cell| Scalar::f64(cell.f64())).collect()) + }) + .collect() +} + +/// Drive the REAL `aura_std::CostSum(n)` aggregator over `n` co-temporal cost +/// streams, summing them per-field — the run-path's cost-graph output node. +fn cost_sum_node_stream(streams: &[&[(Timestamp, Vec)]]) -> Vec<(Timestamp, Vec)> { + let n = streams.len(); + let len = streams[0].len(); + let mut node = CostSum::new(n); + (0..len) + .map(|i| { + let ts = streams[0][i].0; + let mut cols: Vec = + (0..n * 3).map(|_| AnyColumn::with_capacity(ScalarKind::F64, 1)).collect(); + for (k, s) in streams.iter().enumerate() { + let (_, row) = &s[i]; + for f in 0..3 { + cols[k * 3 + f].push(Scalar::f64(row[f].as_f64())).unwrap(); + } + } + let out = node.eval(Ctx::new(&cols, ts)).expect("aggregate co-temporal with cost streams"); + (ts, out.iter().map(|cell| Scalar::f64(cell.f64())).collect()) + }) + .collect() +} + +/// Property (spec 0082, exact composition): the `CostSum` of the real +/// `ConstantCost` + `VolSlippageCost` node streams folds through `summarize_r` to +/// the exact additive net — `net_both == net_flat + net_vol − gross` (since each +/// single net is `gross − its_mean_cost`, the composed net subtracts BOTH mean +/// costs). The path carries a clean stopped loser AND a window-end open trade, so +/// both the cumulative-close-cost and the window-end open-cost terms bite. +#[test] +fn cost_sum_composes_constant_and_vol_slippage_exactly() { + let mut stop = FixedStop::new(10.0); + let ledger = run_chain_ledger( + &mut stop, + &[(1.0, 100.0), (1.0, 100.0), (0.0, 90.0), (1.0, 100.0), (1.0, 102.0), (1.0, 105.0)], + ); + let cc = const_cost_node_stream(&ledger, 2.0); + let vs = vol_slippage_node_stream(&ledger, 0.5, 3.0); + let summed = cost_sum_node_stream(&[&cc, &vs]); + + let gross = summarize_r(&ledger, &[]).expectancy_r; + let net_flat = summarize_r(&ledger, &cc).net_expectancy_r; + let net_vol = summarize_r(&ledger, &vs).net_expectancy_r; + let net_both = summarize_r(&ledger, &summed).net_expectancy_r; + + // additive identity: mean(r − cc − vs) == mean(r − cc) + mean(r − vs) − mean(r) + assert!( + (net_both - (net_flat + net_vol - gross)).abs() < 1e-9, + "composition is exact + additive: net_both {net_both} == net_flat {net_flat} + net_vol {net_vol} − gross {gross}", + ); + // both costs bite: the composed net is strictly below either single-cost net. + assert!(net_both < net_flat && net_both < net_vol, "both costs bite"); +} + +/// Property (spec 0082): the aggregate `CostSum` stream agrees with the in-graph +/// net seam — the in-graph final `net_r_equity` sample (a LinComb over the executor +/// outputs and the AGGREGATE cost) equals the post-run `summarize_r` net total. The +/// aggregate is the one cost stream the net tap and `summarize_r` both read. +#[test] +fn aggregate_net_r_equity_final_sample_agrees_with_summarize_r_net_total() { + let mut stop = FixedStop::new(10.0); + let ledger = run_chain_ledger( + &mut stop, + &[(1.0, 100.0), (1.0, 100.0), (0.0, 90.0), (1.0, 100.0), (1.0, 102.0), (1.0, 105.0)], + ); + let cc = const_cost_node_stream(&ledger, 2.0); + let vs = vol_slippage_node_stream(&ledger, 0.5, 3.0); + let summed = cost_sum_node_stream(&[&cc, &vs]); + + let m = summarize_r(&ledger, &summed); + let post_run_net_total = m.net_expectancy_r * m.n_trades as f64; + let (_, last_pm) = ledger.last().unwrap(); + let (_, last_cost) = summed.last().unwrap(); + let net_eq_final = last_pm[CUM_REALIZED_R].as_f64() + last_pm[UNREALIZED_R].as_f64() + - last_cost[CUM_COST_IN_R].as_f64() + - last_cost[OPEN_COST_IN_R].as_f64(); + assert!( + (net_eq_final - post_run_net_total).abs() < 1e-9, + "in-graph aggregate net_r_equity {net_eq_final} must equal post-run net total {post_run_net_total}", + ); +} + +/// Property (spec 0082): `CostSum(1)` is the identity — a lone vol-slippage stream +/// folds through the aggregator to exactly the un-aggregated net (the run path is +/// uniform whether one or several cost nodes are wired). +#[test] +fn cost_sum_of_one_is_identity_for_vol_slippage() { + let mut stop = FixedStop::new(10.0); + let ledger = run_chain_ledger(&mut stop, &long_path(&[100.0, 102.0, 105.0])); + let vs = vol_slippage_node_stream(&ledger, 0.5, 3.0); + let single = cost_sum_node_stream(&[&vs]); + assert_eq!( + summarize_r(&ledger, &single).net_expectancy_r, + summarize_r(&ledger, &vs).net_expectancy_r, + "CostSum(1) does not move the net", + ); +} + /// Property (#130): **`sqn` and `sqn_normalized` (SQN100) are computed from the /// producer's *recorded* dense records, not hand-built rows, and below the /// 100-trade cap the two are identical.** Every `summarize_r` test that touches diff --git a/crates/aura-std/src/cost_sum.rs b/crates/aura-std/src/cost_sum.rs new file mode 100644 index 0000000..af0ceaa --- /dev/null +++ b/crates/aura-std/src/cost_sum.rs @@ -0,0 +1,162 @@ +//! `CostSum` — the output node of a C10 cost-model graph: it sums `n_costs` +//! cost-in-R records per-field into one aggregate record, so any number of cost +//! nodes collapses to the single 3-field cost stream the net-R seam already +//! consumes (`summarize_r` + the `net_r_equity` tap stay unchanged). Each cost +//! node contributes the 3-field `{cost_in_r, cum_cost_in_r, open_cost_in_r}` +//! record; the aggregate is the per-field sum. `n_costs = 1` is the identity, so +//! the cost path is uniform whether one or several cost nodes are wired. + +use aura_core::{ + Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, ScalarKind, +}; + +/// The cost-record field triple every cost node emits, in slot order, and its +/// width. Interned once (mirroring `position_management::FIELD_NAMES`/`WIDTH`) so +/// the input-name loop, the output schema, the lookback vector, and the eval +/// accumulator all read one source of truth instead of restating the triple. The +/// producer nodes (`ConstantCost`, `VolSlippageCost`) emit this same triple; that +/// cross-node match remains a by-convention lockstep contract. +const COST_FIELDS: [&str; 3] = ["cost_in_r", "cum_cost_in_r", "open_cost_in_r"]; +const COST_WIDTH: usize = COST_FIELDS.len(); + +/// Per-field sum of `n_costs` cost-in-R records. Inputs are +/// `cost[k].{cost_in_r,cum_cost_in_r,open_cost_in_r}` for `k in 0..n_costs`, in +/// slot order (3 per cost node); the 3-field output mirrors a single cost record. +/// Emits `None` until every input leg is present (mode-A as-of join, like LinComb). +pub struct CostSum { + n_costs: usize, + out: [Cell; COST_WIDTH], +} + +impl CostSum { + pub fn new(n_costs: usize) -> Self { + assert!(n_costs >= 1, "CostSum needs at least one cost input"); + Self { n_costs, out: [Cell::from_f64(0.0); COST_WIDTH] } + } + + /// The param-generic recipe. `n_costs` is topology (fixed per blueprint, C19), + /// captured by the build closure (no per-build params). The input names are a + /// lockstep contract with the connect side (`cost[k].`). + pub fn builder(n_costs: usize) -> PrimitiveBuilder { + let mut inputs = Vec::with_capacity(n_costs * COST_WIDTH); + for k in 0..n_costs { + for field in COST_FIELDS { + inputs.push(PortSpec { + kind: ScalarKind::F64, + firing: Firing::Any, + name: format!("cost[{k}].{field}"), + }); + } + } + PrimitiveBuilder::new( + "CostSum", + NodeSchema { + inputs, + output: COST_FIELDS + .iter() + .map(|n| FieldSpec { name: (*n).into(), kind: ScalarKind::F64 }) + .collect(), + params: vec![], + }, + move |_| Box::new(CostSum::new(n_costs)), + ) + } +} + +impl Node for CostSum { + fn lookbacks(&self) -> Vec { + vec![1; self.n_costs * COST_WIDTH] + } + + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { + let mut acc = [0.0_f64; COST_WIDTH]; // per-field accumulator, COST_FIELDS order + for k in 0..self.n_costs { + for (f, slot) in acc.iter_mut().enumerate() { + let w = ctx.f64_in(k * COST_WIDTH + f); + if w.is_empty() { + return None; + } + *slot += w[0]; + } + } + self.out = [Cell::from_f64(acc[0]), Cell::from_f64(acc[1]), Cell::from_f64(acc[2])]; + Some(&self.out) + } + + fn label(&self) -> String { + format!("CostSum({})", self.n_costs) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aura_core::{AnyColumn, Scalar, Timestamp}; + + fn f64_cols(n: usize) -> Vec { + (0..n).map(|_| AnyColumn::with_capacity(ScalarKind::F64, 1)).collect() + } + + #[test] + fn two_records_sum_per_field() { + let mut s = CostSum::new(2); + let mut inputs = f64_cols(6); + // cost[0] = [0.5, 0.5, 0.0]; cost[1] = [0.375, 1.0, 0.2] + for (i, v) in [0.5, 0.5, 0.0, 0.375, 1.0, 0.2].into_iter().enumerate() { + inputs[i].push(Scalar::f64(v)).unwrap(); + } + // per-field sum: [0.875, 1.5, 0.2] + assert_eq!( + s.eval(Ctx::new(&inputs, Timestamp(0))), + Some([Cell::from_f64(0.875), Cell::from_f64(1.5), Cell::from_f64(0.2)].as_slice()) + ); + } + + #[test] + fn n_one_is_identity() { + let mut s = CostSum::new(1); + let mut inputs = f64_cols(3); + for (i, v) in [0.5, 1.25, 0.3].into_iter().enumerate() { + inputs[i].push(Scalar::f64(v)).unwrap(); + } + assert_eq!( + s.eval(Ctx::new(&inputs, Timestamp(0))), + Some([Cell::from_f64(0.5), Cell::from_f64(1.25), Cell::from_f64(0.3)].as_slice()) + ); + } + + #[test] + fn withholds_until_every_leg_present() { + let mut s = CostSum::new(2); + let mut inputs = f64_cols(6); + // only the first cost node's three fields present -> withhold + for col in inputs.iter_mut().take(3) { + col.push(Scalar::f64(1.0)).unwrap(); + } + assert_eq!(s.eval(Ctx::new(&inputs, Timestamp(0))), None); + } + + #[test] + fn input_slots_are_named_cost_index_field() { + let s = CostSum::builder(2); + let names: Vec = s.schema().inputs.iter().map(|p| p.name.clone()).collect(); + assert_eq!( + names, + [ + "cost[0].cost_in_r", "cost[0].cum_cost_in_r", "cost[0].open_cost_in_r", + "cost[1].cost_in_r", "cost[1].cum_cost_in_r", "cost[1].open_cost_in_r", + ] + ); + } + + #[test] + fn label_carries_the_arity() { + assert_eq!(CostSum::new(2).label(), "CostSum(2)"); + } + + #[test] + #[should_panic(expected = "CostSum needs at least one cost input")] + fn new_panics_on_zero() { + let _ = CostSum::new(0); + } +} diff --git a/crates/aura-std/src/lib.rs b/crates/aura-std/src/lib.rs index 045c030..48f9a60 100644 --- a/crates/aura-std/src/lib.rs +++ b/crates/aura-std/src/lib.rs @@ -19,6 +19,7 @@ mod add; mod and; mod bias; mod constant_cost; +mod cost_sum; mod delay; mod ema; mod eqconst; @@ -41,10 +42,12 @@ mod sma; mod sqrt; mod stop_rule; mod sub; +mod vol_slippage_cost; pub use add::Add; pub use and::And; pub use bias::Bias; pub use constant_cost::ConstantCost; +pub use cost_sum::CostSum; pub use delay::Delay; pub use ema::Ema; pub use eqconst::EqConst; @@ -70,3 +73,4 @@ pub use sma::Sma; pub use sqrt::Sqrt; pub use stop_rule::FixedStop; pub use sub::Sub; +pub use vol_slippage_cost::VolSlippageCost; diff --git a/crates/aura-std/src/vol_slippage_cost.rs b/crates/aura-std/src/vol_slippage_cost.rs new file mode 100644 index 0000000..99518fa --- /dev/null +++ b/crates/aura-std/src/vol_slippage_cost.rs @@ -0,0 +1,222 @@ +//! `VolSlippageCost` — a slippage cost that scales with a measured volatility +//! input, charged once per closed trade, in R. The second cost node of the C10 +//! cost-model graph and the first *state-dependent* one: identical in shape to +//! [`crate::ConstantCost`] but its per-trade charge numerator is +//! `slip_vol_mult · volatility` instead of a flat constant, so the cost-in-R +//! varies trade-to-trade. R-pure: `slip_vol_mult · vol / |entry - stop|`; +//! notional cancels (C10). The vol is supplied as an input (an upstream +//! realized-range estimator), kept independent of the stop's own vol — scaling +//! by the stop's vol would collapse cost-in-R to a constant. +//! +//! Co-temporality contract: a cost node's stream must stay 1:1 with the +//! executor's PM record (the positional join `summarize_r` relies on). So the +//! node is gated ONLY by the PM geometry; a not-yet-warm `volatility` input (the +//! realized-range proxy warms later than PM) contributes 0 cost that cycle rather +//! than withholding and desyncing the stream — honest, since there is no slippage +//! estimate yet. This generalizes to any state-dependent cost factor. + +use aura_core::{ + Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, + ScalarKind, +}; + +/// A volatility-scaled per-trade slippage, emitted in R. Inputs are the four +/// executor-exposed geometry fields `closed`/`open`/`entry_price`/`stop_price` +/// plus a `volatility` stream (price units). Withholds (`None`) only until the +/// four geometry inputs are present; a not-yet-warm `volatility` contributes 0 +/// cost, so the stream stays co-temporal with the PM record. +pub struct VolSlippageCost { + slip_vol_mult: f64, + cum: f64, + out: [Cell; 3], +} + +impl VolSlippageCost { + pub fn new(slip_vol_mult: f64) -> Self { + assert!(slip_vol_mult >= 0.0, "VolSlippageCost slip_vol_mult must be >= 0"); + Self { slip_vol_mult, cum: 0.0, out: [Cell::from_f64(0.0); 3] } + } + + /// The param-generic recipe: one `slip_vol_mult` F64 knob; five inputs. + pub fn builder() -> PrimitiveBuilder { + PrimitiveBuilder::new( + "VolSlippageCost", + NodeSchema { + inputs: vec![ + PortSpec { kind: ScalarKind::Bool, firing: Firing::Any, name: "closed".into() }, + PortSpec { kind: ScalarKind::Bool, firing: Firing::Any, name: "open".into() }, + PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "entry_price".into() }, + PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "stop_price".into() }, + PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "volatility".into() }, + ], + output: vec![ + FieldSpec { name: "cost_in_r".into(), kind: ScalarKind::F64 }, + FieldSpec { name: "cum_cost_in_r".into(), kind: ScalarKind::F64 }, + FieldSpec { name: "open_cost_in_r".into(), kind: ScalarKind::F64 }, + ], + params: vec![ParamSpec { name: "slip_vol_mult".into(), kind: ScalarKind::F64 }], + }, + |p| Box::new(VolSlippageCost::new(p[0].f64())), + ) + } +} + +impl Node for VolSlippageCost { + fn lookbacks(&self) -> Vec { + vec![1, 1, 1, 1, 1] + } + + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { + let closed_w = ctx.bool_in(0); + let open_w = ctx.bool_in(1); + let entry_w = ctx.f64_in(2); + let stop_w = ctx.f64_in(3); + let vol_w = ctx.f64_in(4); + // Gate only on the PM geometry (co-temporality contract); a not-yet-warm + // volatility proxy contributes 0 cost rather than withholding the row. + if closed_w.is_empty() || open_w.is_empty() || entry_w.is_empty() || stop_w.is_empty() { + return None; + } + let closed = closed_w[0]; + let open = open_w[0]; + let latched = (entry_w[0] - stop_w[0]).abs(); + let vol = if vol_w.is_empty() { 0.0 } else { vol_w[0] }; // 0 during proxy warm-up + // Same zero-latched guard as ConstantCost: no valid 1R denominator -> no cost. + let per = if latched > 0.0 { self.slip_vol_mult * vol / latched } else { 0.0 }; + let cost_in_r = if closed { per } else { 0.0 }; + let open_cost_in_r = if open { per } else { 0.0 }; + self.cum += cost_in_r; + self.out = [ + Cell::from_f64(cost_in_r), + Cell::from_f64(self.cum), + Cell::from_f64(open_cost_in_r), + ]; + Some(&self.out) + } + + fn label(&self) -> String { + format!("VolSlippageCost({})", self.slip_vol_mult) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aura_core::{AnyColumn, Scalar, Timestamp}; + + fn cols() -> Vec { + vec![ + AnyColumn::with_capacity(ScalarKind::Bool, 1), // closed + AnyColumn::with_capacity(ScalarKind::Bool, 1), // open + AnyColumn::with_capacity(ScalarKind::F64, 1), // entry + AnyColumn::with_capacity(ScalarKind::F64, 1), // stop + AnyColumn::with_capacity(ScalarKind::F64, 1), // volatility + ] + } + + #[test] + fn no_geometry_yet_withholds() { + let mut c = VolSlippageCost::new(0.5); + let inputs = cols(); // all columns empty + assert_eq!(c.eval(Ctx::new(&inputs, Timestamp(0))), None); + } + + #[test] + fn vol_not_yet_warm_emits_zero_cost_co_temporally() { + // The realized-range proxy warms after the PM geometry; during warm-up the + // node must still EMIT (a 0-cost row) so the cost stream stays 1:1 with the + // PM record — withholding here would desync the positional join. + let mut c = VolSlippageCost::new(0.5); + let mut inputs = cols(); + inputs[0].push(Scalar::bool(true)).unwrap(); // closed + inputs[1].push(Scalar::bool(false)).unwrap(); + inputs[2].push(Scalar::f64(100.0)).unwrap(); + inputs[3].push(Scalar::f64(96.0)).unwrap(); // latched 4.0 + // volatility column empty (proxy not warm) -> vol = 0 -> 0 cost, but a row IS emitted + assert_eq!( + c.eval(Ctx::new(&inputs, Timestamp(0))), + Some([Cell::from_f64(0.0), Cell::from_f64(0.0), Cell::from_f64(0.0)].as_slice()) + ); + } + + #[test] + fn closed_charges_mult_times_vol_over_latched() { + let mut c = VolSlippageCost::new(0.5); + let mut inputs = cols(); + inputs[0].push(Scalar::bool(true)).unwrap(); // closed + inputs[1].push(Scalar::bool(false)).unwrap(); + inputs[2].push(Scalar::f64(100.0)).unwrap(); + inputs[3].push(Scalar::f64(96.0)).unwrap(); // latched 4.0 + inputs[4].push(Scalar::f64(3.0)).unwrap(); // vol 3.0 + // per = 0.5 * 3.0 / 4.0 = 0.375; cum = 0.375; open = 0.0 + assert_eq!( + c.eval(Ctx::new(&inputs, Timestamp(0))), + Some([Cell::from_f64(0.375), Cell::from_f64(0.375), Cell::from_f64(0.0)].as_slice()) + ); + } + + #[test] + fn open_emits_would_be_cost_not_charged_to_cum() { + let mut c = VolSlippageCost::new(0.5); + let mut inputs = cols(); + inputs[0].push(Scalar::bool(false)).unwrap(); + inputs[1].push(Scalar::bool(true)).unwrap(); // open + inputs[2].push(Scalar::f64(100.0)).unwrap(); + inputs[3].push(Scalar::f64(96.0)).unwrap(); // latched 4.0 + inputs[4].push(Scalar::f64(3.0)).unwrap(); // vol 3.0 + // cost_in_r = 0; cum 0; open_cost_in_r = 0.375 + assert_eq!( + c.eval(Ctx::new(&inputs, Timestamp(0))), + Some([Cell::from_f64(0.0), Cell::from_f64(0.0), Cell::from_f64(0.375)].as_slice()) + ); + } + + #[test] + fn zero_latched_contributes_no_cost() { + let mut c = VolSlippageCost::new(0.5); + let mut inputs = cols(); + inputs[0].push(Scalar::bool(true)).unwrap(); + inputs[1].push(Scalar::bool(false)).unwrap(); + inputs[2].push(Scalar::f64(100.0)).unwrap(); + inputs[3].push(Scalar::f64(100.0)).unwrap(); // latched 0 -> no divide + inputs[4].push(Scalar::f64(3.0)).unwrap(); + assert_eq!( + c.eval(Ctx::new(&inputs, Timestamp(0))), + Some([Cell::from_f64(0.0), Cell::from_f64(0.0), Cell::from_f64(0.0)].as_slice()) + ); + } + + #[test] + fn cum_accumulates_across_closes() { + let mut c = VolSlippageCost::new(0.5); + let mut a = cols(); + a[0].push(Scalar::bool(true)).unwrap(); + a[1].push(Scalar::bool(false)).unwrap(); + a[2].push(Scalar::f64(100.0)).unwrap(); + a[3].push(Scalar::f64(96.0)).unwrap(); // latched 4 + a[4].push(Scalar::f64(3.0)).unwrap(); // 0.5*3/4 = 0.375 + let _ = c.eval(Ctx::new(&a, Timestamp(0))); + let mut b = cols(); + b[0].push(Scalar::bool(true)).unwrap(); + b[1].push(Scalar::bool(false)).unwrap(); + b[2].push(Scalar::f64(100.0)).unwrap(); + b[3].push(Scalar::f64(98.0)).unwrap(); // latched 2 + b[4].push(Scalar::f64(4.0)).unwrap(); // 0.5*4/2 = 1.0; cum 1.375 + assert_eq!( + c.eval(Ctx::new(&b, Timestamp(1))), + Some([Cell::from_f64(1.0), Cell::from_f64(1.375), Cell::from_f64(0.0)].as_slice()) + ); + } + + #[test] + fn label_carries_the_mult() { + assert_eq!(VolSlippageCost::new(0.5).label(), "VolSlippageCost(0.5)"); + assert_eq!(VolSlippageCost::new(2.0).label(), "VolSlippageCost(2)"); + } + + #[test] + #[should_panic(expected = "slip_vol_mult must be >= 0")] + fn new_panics_on_negative_mult() { + let _ = VolSlippageCost::new(-1.0); + } +} diff --git a/docs/plans/0082-vol-slippage-cost.md b/docs/plans/0082-vol-slippage-cost.md index 75ab516..dec85a0 100644 --- a/docs/plans/0082-vol-slippage-cost.md +++ b/docs/plans/0082-vol-slippage-cost.md @@ -68,8 +68,9 @@ use aura_core::{ /// A volatility-scaled per-trade slippage, emitted in R. Inputs are the four /// executor-exposed geometry fields `closed`/`open`/`entry_price`/`stop_price` -/// plus a `volatility` stream (price units). Emits `None` until all five inputs -/// are present this cycle. +/// plus a `volatility` stream (price units). Withholds (`None`) only until the +/// four geometry inputs are present; a not-yet-warm `volatility` contributes 0 +/// cost, so the stream stays co-temporal 1:1 with the PM record. pub struct VolSlippageCost { slip_vol_mult: f64, cum: f64, @@ -117,15 +118,15 @@ impl Node for VolSlippageCost { let entry_w = ctx.f64_in(2); let stop_w = ctx.f64_in(3); let vol_w = ctx.f64_in(4); - if closed_w.is_empty() || open_w.is_empty() || entry_w.is_empty() - || stop_w.is_empty() || vol_w.is_empty() - { + // Gate only on the PM geometry (co-temporality contract); a not-yet-warm + // volatility proxy contributes 0 cost rather than withholding the row. + if closed_w.is_empty() || open_w.is_empty() || entry_w.is_empty() || stop_w.is_empty() { return None; } let closed = closed_w[0]; let open = open_w[0]; let latched = (entry_w[0] - stop_w[0]).abs(); - let vol = vol_w[0].max(0.0); // a realized range is non-negative; clamp defensively + let vol = if vol_w.is_empty() { 0.0 } else { vol_w[0] }; // 0 during proxy warm-up // Same zero-latched guard as ConstantCost: no valid 1R denominator -> no cost. let per = if latched > 0.0 { self.slip_vol_mult * vol / latched } else { 0.0 }; let cost_in_r = if closed { per } else { 0.0 }; @@ -167,15 +168,21 @@ mod tests { } #[test] - fn withholds_until_volatility_present() { + fn vol_not_yet_warm_emits_zero_cost_co_temporally() { + // The realized-range proxy warms after the PM geometry; during warm-up the + // node must still EMIT (a 0-cost row) so the cost stream stays 1:1 with the + // PM record — withholding here would desync the positional join. let mut c = VolSlippageCost::new(0.5); let mut inputs = cols(); - inputs[0].push(Scalar::bool(true)).unwrap(); + inputs[0].push(Scalar::bool(true)).unwrap(); // closed inputs[1].push(Scalar::bool(false)).unwrap(); inputs[2].push(Scalar::f64(100.0)).unwrap(); - inputs[3].push(Scalar::f64(96.0)).unwrap(); - // volatility column still empty -> withhold - assert_eq!(c.eval(Ctx::new(&inputs, Timestamp(0))), None); + inputs[3].push(Scalar::f64(96.0)).unwrap(); // latched 4.0 + // volatility column empty (proxy not warm) -> vol = 0 -> 0 cost, but a row IS emitted + assert_eq!( + c.eval(Ctx::new(&inputs, Timestamp(0))), + Some([Cell::from_f64(0.0), Cell::from_f64(0.0), Cell::from_f64(0.0)].as_slice()) + ); } #[test] @@ -534,9 +541,10 @@ and beside the `COL_PORTS` static, add: ```rust /// Short-horizon realized-range window for vol-scaled slippage. Deliberately -/// distinct from `STAGE1_R_STOP_LENGTH`: scaling slippage by the stop's own vol -/// would collapse cost-in-R to a constant (spec 0082). -const SLIP_VOL_LENGTH: i64 = 20; +/// distinct from `STAGE1_R_STOP_LENGTH` (3): scaling slippage by the stop's own +/// vol would collapse cost-in-R to a constant (spec 0082). Short enough to warm +/// within the synthetic smoke fixture so the run path exercises non-zero slippage. +const SLIP_VOL_LENGTH: i64 = 5; /// The maximum number of cost nodes the run-path cost graph wires (flat cost + /// vol slippage). Sizes the interned `CostSum` input-port names below. diff --git a/docs/specs/0082-vol-slippage-cost.md b/docs/specs/0082-vol-slippage-cost.md index 9aa701a..645b71a 100644 --- a/docs/specs/0082-vol-slippage-cost.md +++ b/docs/specs/0082-vol-slippage-cost.md @@ -65,6 +65,19 @@ from `ConstantCost`, defeating the point of a second, state-dependent node. So `cost_in_R = k·vol_short / stop_dist` then varies trade-to-trade with the short/long vol ratio — genuinely state-dependent. +**Co-temporality contract (load-bearing — the cost stream stays 1:1 with PM).** +`summarize_r` positional-joins `cost[i] ↔ record[i]`, so the cost stream MUST be +co-temporal 1:1 with the PM record. A cost node is therefore gated **only by the +PM trade-geometry** (`closed`/`open`/`entry`/`stop`); any not-yet-warm state input +(here the realized-range proxy, which warms later than PM) contributes **0 cost** +that cycle rather than withholding — the node still emits its row. This makes +co-temporality structural and warmup-independent, preserves the C18 golden, and +generalizes to any future state-dependent cost factor (a recorded swap rate, etc.). +`ConstantCost` satisfies it trivially (no warming factor); only `VolSlippageCost` +needs the missing-factor→0 rule. `SLIP_VOL_LENGTH` is kept short (5) so the vol +proxy warms within the synthetic smoke fixture and the run path exercises non-zero +slippage. + ## Concrete code shapes ### User-facing program (the acceptance evidence) @@ -137,15 +150,15 @@ impl Node for VolSlippageCost { let entry_w = ctx.f64_in(2); let stop_w = ctx.f64_in(3); let vol_w = ctx.f64_in(4); - if closed_w.is_empty() || open_w.is_empty() || entry_w.is_empty() - || stop_w.is_empty() || vol_w.is_empty() - { + // Gate only on the PM geometry (co-temporality contract); a not-yet-warm + // volatility proxy contributes 0 cost rather than withholding the row. + if closed_w.is_empty() || open_w.is_empty() || entry_w.is_empty() || stop_w.is_empty() { return None; } let closed = closed_w[0]; let open = open_w[0]; let latched = (entry_w[0] - stop_w[0]).abs(); - let vol = vol_w[0].max(0.0); // a negative range is impossible; clamp defensively + let vol = if vol_w.is_empty() { 0.0 } else { vol_w[0] }; // 0 during proxy warm-up // Same zero-latched guard as ConstantCost: no valid 1R denominator -> no cost. let per = if latched > 0.0 { self.slip_vol_mult * vol / latched } else { 0.0 }; let cost_in_r = if closed { per } else { 0.0 }; @@ -246,9 +259,10 @@ A new module const sits beside `STAGE1_R_STOP_LENGTH`: ```rust /// Short-horizon realized-range window for vol-scaled slippage. Deliberately -/// distinct from STAGE1_R_STOP_LENGTH: scaling slippage by the stop's own vol -/// would collapse cost-in-R to a constant (see spec 0082 Architecture). -const SLIP_VOL_LENGTH: i64 = 20; +/// distinct from STAGE1_R_STOP_LENGTH (3): scaling slippage by the stop's own vol +/// would collapse cost-in-R to a constant (see spec 0082 Architecture). Short +/// enough to warm within the synthetic smoke fixture so slippage actually bites. +const SLIP_VOL_LENGTH: i64 = 5; /// Which cost nodes the run-path cost graph builds. At least one field is `Some` /// (the outer `Option` is `None` when no cost flag was given). @@ -434,9 +448,10 @@ aggregate feeds the 3-wide cost `Recorder` (→ `summarize_r`) and the guard), matching `ConstantCost` and `summarize_r`. - A negative vol range (impossible from `max − min`, but defended) is clamped to `0.0`. -- Warm-up: before the vol window fills, `vrange` (hence `VolSlippageCost`) - withholds, so the earliest trades within `SLIP_VOL_LENGTH` bars carry no - slippage; documented, not an error. +- Warm-up: before the vol window fills the proxy emits nothing, so + `VolSlippageCost` reads a missing vol as 0 and charges 0 slippage that cycle — it + still EMITS its row (co-temporality contract), so the cost stream stays 1:1 with + the PM record. `SLIP_VOL_LENGTH` is short enough to warm within the fixture. ## Testing strategy