Brummel/Aura
1
0
Fork 0
Code Issues 33 Pull Requests Actions Packages Projects Releases Wiki Activity

plan: 0082 vol-slippage cost + cost-graph composition

Browse Source 
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
This commit is contained in:
Brummel
2026-06-28 15:59:56 +02:00
parent 372eae9cac
commit 31ef46314a
1 changed files with 965 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plans/0082-vol-slippage-cost.md
+965
View File
@@ -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<usize> {
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<AnyColumn> {
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].<field>`).
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<usize> {
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<AnyColumn> {
(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<String> = 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].<field>` `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<Vec<String>> = 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<f64>, // --cost-per-trade
slip_vol_mult: Option<f64>, // --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<Scalar>)>, mpsc::Sender<(Timestamp, Vec<Scalar>)>)>,
```
- [ ] **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<f64>,
slip_vol_mult: Option<f64>,
) -> 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<f64>,
```
In `parse_run_args`, add a local beside `let mut cost: Option<f64> = None;` (3081):
```rust
let mut slip_vol_mult: Option<f64> = 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 <f64>]` after `[--cost-per-trade <f64>]`.
- [ ] **Step 12: Extend the `USAGE` const**
In `crates/aura-cli/src/main.rs:3156`, in the `aura run` clause, add
`[--slip-vol-mult <f64>]` after `[--cost-per-trade <f64>]`.
- [ ] **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<Scalar>)],
k: f64,
vol: f64,
) -> Vec<(Timestamp, Vec<Scalar>)> {
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<Scalar>)]]) -> Vec<(Timestamp, Vec<Scalar>)> {
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<AnyColumn> =
(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.