From 31ef46314ae94df3008ec8c4e65dca63c419b3d9 Mon Sep 17 00:00:00 2001 From: Brummel Date: Sun, 28 Jun 2026 15:59:56 +0200 Subject: [PATCH] plan: 0082 vol-slippage cost + cost-graph composition MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task-by-task plan for spec 0082: the VolSlippageCost node (Task 1) and the CostSum aggregator (Task 2) in aura-std; the run-path wiring as one compile-gate task (Task 3) — stage1_r_graph cost-param widening + vol-proxy hoist + the CostSum-aggregated cost block + the --slip-vol-mult thread-through across all call sites, with a CLI composition test and the no-cost golden floor; and the node-level exact-sum composition tests (Task 4). RED-first per task. The &'static str CostSum port names are interned (COST_SUM_PORTS, mirroring COL_PORTS); summarize_r is unchanged (it folds the CostSum aggregate). refs #148 --- docs/plans/0082-vol-slippage-cost.md | 965 +++++++++++++++++++++++++++ 1 file changed, 965 insertions(+) create mode 100644 docs/plans/0082-vol-slippage-cost.md diff --git a/docs/plans/0082-vol-slippage-cost.md b/docs/plans/0082-vol-slippage-cost.md new file mode 100644 index 0000000..75ab516 --- /dev/null +++ b/docs/plans/0082-vol-slippage-cost.md @@ -0,0 +1,965 @@ +# Vol-slippage cost node + cost-graph composition — Implementation Plan + +> **Parent spec:** `docs/specs/0082-vol-slippage-cost.md` +> +> **For agentic workers:** REQUIRED SUB-SKILL: use the `implement` skill to run +> this plan. Steps use `- [ ]` checkboxes for tracking. + +**Goal:** Ship a second, state-dependent cost node (`VolSlippageCost`) and a +cost-graph aggregator (`CostSum`), wired into the stage1-r run path so two cost +nodes compose into one net-R curve while `summarize_r` and the `net_r_equity` tap +stay structurally unchanged. + +**Architecture:** Both new nodes live in `aura-std` and emit the cycle-1 3-field +cost-in-R record `{cost_in_r, cum_cost_in_r, open_cost_in_r}`. `CostSum` is the +cost-graph output: it sums N cost nodes' records per-field, so the single 3-wide +cost stream the seam already consumes is now the aggregate (`n=1` is the +identity). The run path inserts the cost nodes → `CostSum(n)` → the existing +net/cost recorders; a hoisted short-horizon vol proxy (`RollingMax−RollingMin` of +price) feeds `VolSlippageCost.volatility` via the single `price` feed. + +**Tech Stack:** `aura-core` (`Node`/`PrimitiveBuilder`/`Cell`), `aura-std` +(node library), `aura-cli` (the stage1-r harness graph + CLI), `aura-engine` +(graph builder, E2E tests), `aura-analysis` (`summarize_r`, unchanged). + +--- + +**Files this plan creates or modifies:** + +- Create: `crates/aura-std/src/vol_slippage_cost.rs` — the `VolSlippageCost` node + unit tests +- Create: `crates/aura-std/src/cost_sum.rs` — the `CostSum` aggregator + unit tests +- Modify: `crates/aura-std/src/lib.rs:18-72` — `mod` + `pub use` for both nodes +- Modify: `crates/aura-cli/src/main.rs:31-34` — aura-std imports +- Modify: `crates/aura-cli/src/main.rs:2562-2566` — `SLIP_VOL_LENGTH`/`CostConfig`/`COST_SUM_PORTS` +- Modify: `crates/aura-cli/src/main.rs:2615-2735` — `stage1_r_graph` signature + vol hoist + cost block +- Modify: `crates/aura-cli/src/main.rs:2956-2970` — `run_stage1_r` signature + cost bundle +- Modify: `crates/aura-cli/src/main.rs:3059-3156` — `RunArgs`/`parse_run_args`/`run_dispatch`/`USAGE` +- Modify: `crates/aura-cli/src/main.rs:4373` — the test call site of `run_stage1_r` +- Test: `crates/aura-cli/tests/cli_run.rs` — CLI composition test (both flags); golden stays green +- Test: `crates/aura-engine/tests/stage1_r_e2e.rs` — node-level exact-sum composition tests + +--- + +### Task 1: `VolSlippageCost` node (aura-std) + +**Files:** +- Create: `crates/aura-std/src/vol_slippage_cost.rs` +- Modify: `crates/aura-std/src/lib.rs:18-72` + +- [ ] **Step 1: Write the node module with its tests** + +Create `crates/aura-std/src/vol_slippage_cost.rs` with exactly: + +```rust +//! `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 (spec 0082). + +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). Emits `None` until all five inputs +/// are present this cycle. +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); + if closed_w.is_empty() || open_w.is_empty() || entry_w.is_empty() + || stop_w.is_empty() || vol_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 + // 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 withholds_until_volatility_present() { + 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(96.0)).unwrap(); + // volatility column still empty -> withhold + assert_eq!(c.eval(Ctx::new(&inputs, Timestamp(0))), None); + } + + #[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); + } +} +``` + +- [ ] **Step 2: Register the module in `lib.rs`** + +In `crates/aura-std/src/lib.rs`, add to the `mod` block (after `mod sub;`, +line 43, keeping it the last entry alphabetically): + +```rust +mod vol_slippage_cost; +``` + +And to the `pub use` block (after `pub use sub::Sub;`, line 72): + +```rust +pub use vol_slippage_cost::VolSlippageCost; +``` + +- [ ] **Step 3: Run the node's tests** + +Run: `cargo test -p aura-std vol_slippage` +Expected: PASS — 8 tests in `vol_slippage_cost::tests` pass. + +--- + +### Task 2: `CostSum` aggregator node (aura-std) + +**Files:** +- Create: `crates/aura-std/src/cost_sum.rs` +- Modify: `crates/aura-std/src/lib.rs:18-72` + +- [ ] **Step 1: Write the aggregator module with its tests** + +Create `crates/aura-std/src/cost_sum.rs` with exactly: + +```rust +//! `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, +}; + +/// 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; 3], +} + +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); 3] } + } + + /// 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 * 3); + for k in 0..n_costs { + for field in ["cost_in_r", "cum_cost_in_r", "open_cost_in_r"] { + inputs.push(PortSpec { + kind: ScalarKind::F64, + firing: Firing::Any, + name: format!("cost[{k}].{field}"), + }); + } + } + PrimitiveBuilder::new( + "CostSum", + NodeSchema { + inputs, + 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![], + }, + move |_| Box::new(CostSum::new(n_costs)), + ) + } +} + +impl Node for CostSum { + fn lookbacks(&self) -> Vec { + vec![1; self.n_costs * 3] + } + + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { + let mut acc = [0.0_f64; 3]; // [cost_in_r, cum_cost_in_r, open_cost_in_r] + for k in 0..self.n_costs { + for f in 0..3 { + let w = ctx.f64_in(k * 3 + f); + if w.is_empty() { + return None; // withhold until every cost leg is present + } + acc[f] += 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 i in 0..3 { + inputs[i].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); + } +} +``` + +- [ ] **Step 2: Register the module in `lib.rs`** + +In `crates/aura-std/src/lib.rs`, add to the `mod` block between `mod constant_cost;` +(line 21) and `mod delay;` (line 22): + +```rust +mod cost_sum; +``` + +And to the `pub use` block after `pub use constant_cost::ConstantCost;` (line 47): + +```rust +pub use cost_sum::CostSum; +``` + +- [ ] **Step 3: Run the aggregator's tests** + +Run: `cargo test -p aura-std cost_sum` +Expected: PASS — 6 tests in `cost_sum::tests` pass. + +--- + +### Task 3: Run-path wiring + CLI flag (aura-cli) + +This task changes two signatures (`stage1_r_graph`'s `cost:` param type and +`run_stage1_r`'s arity), so every call site is threaded inside this one task and +the task ends on a clean `cargo build -p aura-cli` (the compile gate). + +**Files:** +- Modify: `crates/aura-cli/src/main.rs` (imports, consts, `stage1_r_graph`, `run_stage1_r`, `RunArgs`, `parse_run_args`, `run_dispatch`, `USAGE`, the test call site at 4373) +- Test: `crates/aura-cli/tests/cli_run.rs` + +- [ ] **Step 1: Write the failing CLI composition test** + +Append to `crates/aura-cli/tests/cli_run.rs` (beside +`stage1_r_cost_run_persists_net_r_equity_and_charges_cost`): + +```rust +/// 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); +} +``` + +- [ ] **Step 2: Run the new test, verify it fails** + +Run: `cargo test -p aura-cli --test cli_run stage1_r_both_costs_compose` +Expected: FAIL — the binary rejects `--slip-vol-mult` (usage error, non-zero +exit), so the `run.status.success()` assert fails. + +- [ ] **Step 3: Add the aura-std imports** + +In `crates/aura-cli/src/main.rs:31-34`, the `use aura_std::{...}` list: add +`CostSum` after `ConstantCost` and `VolSlippageCost` after `Sub` (alphabetical +within the existing list). Resulting additions only — `RollingMax`, `RollingMin`, +`Sub`, `LinComb`, `Recorder`, `ConstantCost` are already imported. + +- [ ] **Step 4: Add the consts, `CostConfig`, and interned port names** + +In `crates/aura-cli/src/main.rs`, after the `STAGE1_R_STOP_K` const (line 2566) +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; + +/// 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 +} +``` + +- [ ] **Step 5: Widen the `stage1_r_graph` `cost:` param type** + +In `crates/aura-cli/src/main.rs:2624`, change the `cost` parameter type: + +```rust + cost: Option<(CostConfig, mpsc::Sender<(Timestamp, Vec)>, mpsc::Sender<(Timestamp, Vec)>)>, +``` + +- [ ] **Step 6: Hoist the vol proxy into the single `price` feed** + +In `crates/aura-cli/src/main.rs`, replace the current `let price` + `g.feed` +(lines 2672-2676) with the hoisted vol proxy and a single multi-target feed: + +```rust + // 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); + 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); +``` + +- [ ] **Step 7: Replace the cost block with the `CostSum`-aggregated block** + +In `crates/aura-cli/src/main.rs`, replace the entire current +`if let Some((cost_per_trade, tx_net, tx_cost)) = cost { ... }` block (lines +2698-2732) with: + +```rust + 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)) + .bind("weights[1]", Scalar::f64(1.0)) + .bind("weights[2]", Scalar::f64(-1.0)) + .bind("weights[3]", Scalar::f64(-1.0)), + ); + 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(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 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(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]")); + } +``` + +- [ ] **Step 8: Thread `run_stage1_r` (signature + cost bundle)** + +In `crates/aura-cli/src/main.rs:2956`, change the signature and the bundle: + +```rust +fn run_stage1_r( + data: RunData, + trace: Option<&str>, + const_cost: Option, + slip_vol_mult: Option, +) -> RunReport { +``` + +Then replace line 2969 (`let cost_bundle = cost.map(|c| (c, tx_net, tx_cost));`): + +```rust + 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 + }; +``` + +(The doc comment at 2952-2955 may keep its prose; the senders `tx_net`/`tx_cost` +and the `stage1_r_graph(...)` call at 2970 are otherwise unchanged.) + +- [ ] **Step 9: Thread the production call site (`run_dispatch`)** + +In `crates/aura-cli/src/main.rs:3145`, change: + +```rust + (HarnessKind::Stage1R, data) => run_stage1_r(data, trace, args.cost, args.slip_vol_mult), +``` + +- [ ] **Step 10: Thread the test call site** + +In `crates/aura-cli/src/main.rs:4373`, the unit test +`run_stage1_r_synthetic_folds_an_r_block` calls `run_stage1_r(RunData::Synthetic, +None, None)`; add the fourth arg: + +```rust + let report = run_stage1_r(RunData::Synthetic, None, None, None); +``` + +(Confirm the exact current call by reading the line; add one trailing `None`.) + +- [ ] **Step 11: Add the `RunArgs` field and parse the flag** + +In `crates/aura-cli/src/main.rs:3059-3064`, add the field to `RunArgs`: + +```rust + slip_vol_mult: Option, +``` + +In `parse_run_args`, add a local beside `let mut cost: Option = None;` (3081): + +```rust + let mut slip_vol_mult: Option = None; +``` + +Add a parse arm after the `--cost-per-trade` arm (after line 3122): + +```rust + "--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; + } +``` + +Add the field to the `Ok(RunArgs { ... })` literal at 3135: + +```rust + Ok(RunArgs { harness, data, trace, cost, slip_vol_mult }) +``` + +Extend the `parse_run_args` usage string (3073) and the doc (3067) to include +`[--slip-vol-mult ]` after `[--cost-per-trade ]`. + +- [ ] **Step 12: Extend the `USAGE` const** + +In `crates/aura-cli/src/main.rs:3156`, in the `aura run` clause, add +`[--slip-vol-mult ]` after `[--cost-per-trade ]`. + +- [ ] **Step 13: Compile gate** + +Run: `cargo build -p aura-cli` +Expected: 0 errors (all call sites threaded; the two signature changes resolve). + +- [ ] **Step 14: Run the new composition test, verify it passes** + +Run: `cargo test -p aura-cli --test cli_run stage1_r_both_costs_compose` +Expected: PASS — both flags accepted, net_both < net_flat, net_r_equity persisted. + +- [ ] **Step 15: Verify the no-cost golden floor is unchanged** + +Run: `cargo test -p aura-cli --test cli_run stage1_r_single_run_output_golden` +Expected: PASS — the C18 no-cost golden is byte-identical (no cost nodes wired, +no `net_r_equity` tap). + +--- + +### Task 4: Node-level exact-sum composition tests (aura-engine) + +Pins the cycle's correctness property precisely: `CostSum` over the real +`ConstantCost` + `VolSlippageCost` nodes folds to the exact per-trade cost sum, +and the aggregate agrees with the in-graph net seam. Test-only (the nodes exist +from Tasks 1-2); it drives the real nodes, not algebraic stand-ins. + +**Files:** +- Test: `crates/aura-engine/tests/stage1_r_e2e.rs` + +- [ ] **Step 1: Add the node-driver helpers and the composition tests** + +In `crates/aura-engine/tests/stage1_r_e2e.rs`, extend the `aura_std` import +(line 28-30) to include `CostSum` and `VolSlippageCost`. Then append, beside +`const_cost_node_stream` (273-295): + +```rust +/// 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 (LinComb over the executor +/// + the AGGREGATE cost) equals the post-run `summarize_r` net total. The aggregate +/// is the single cost stream the run-path's 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", + ); +} +``` + +- [ ] **Step 2: Run the composition tests** + +Run: `cargo test -p aura-engine --test stage1_r_e2e cost_sum` +Expected: PASS — `cost_sum_composes_constant_and_vol_slippage_exactly` and +`cost_sum_of_one_is_identity_for_vol_slippage` pass. + +Run: `cargo test -p aura-engine --test stage1_r_e2e aggregate_net_r_equity` +Expected: PASS — the in-graph-vs-post-run aggregate agreement holds. + +--- + +### Self-review (orchestrator, pre-handoff) + +1. **Spec coverage:** VolSlippageCost (Task 1), CostSum (Task 2), run-path wiring + + CLI flag + composition E2E + golden floor (Task 3), exact-sum + aggregate + agreement + identity (Task 4). Every spec section has a task. +2. **Placeholder scan:** no TBD/TODO/"similar to"/"add appropriate". +3. **Type consistency:** `cost_in_r`/`cum_cost_in_r`/`open_cost_in_r` field order + matches across both nodes, `CostSum`, `COST_FIELDS`, `COST_SUM_PORTS`, and the + cost `Recorder` cols; `CostConfig`/`slip_vol_mult`/`SLIP_VOL_LENGTH` consistent + across `stage1_r_graph`/`run_stage1_r`/`RunArgs`/`parse_run_args`. +4. **Step granularity:** each step is a single file edit or one command. +5. **No commit steps:** none present. +6. **Pin/replacement contiguity:** the CLI test asserts on JSON keys + (`net_expectancy_r`), not a verbatim substring of an edited body — no split-pin + risk. The golden test (Task 3 Step 15) is an existing pin, unchanged. +7. **Compile-gate vs deferred-caller:** the two signature changes + (`stage1_r_graph` cost type, `run_stage1_r` arity) and ALL their call sites + (`run_dispatch` 3145, the 6 `None`-passing graph sites which type-unify, and + the test site 4373) are threaded inside Task 3 before its Step-13 build gate — + no caller deferred past the gate. +8. **Verification-command filters resolve:** `vol_slippage`/`cost_sum`/ + `stage1_r_both_costs_compose`/`stage1_r_single_run_output_golden`/ + `aggregate_net_r_equity` each name a real test added or existing in this plan.