diff --git a/docs/plans/0081-cost-model-constant-cost.md b/docs/plans/0081-cost-model-constant-cost.md new file mode 100644 index 0000000..97a8c32 --- /dev/null +++ b/docs/plans/0081-cost-model-constant-cost.md @@ -0,0 +1,671 @@ +# Cost-model graph (in R) — cycle 1: ConstantCost + net-R seam — Implementation Plan + +> **Parent spec:** `docs/specs/0081-cost-model-constant-cost.md` +> +> **For agentic workers:** REQUIRED SUB-SKILL: use the `implement` skill to run +> this plan. Steps use `- [ ]` checkboxes for tracking. + +**Goal:** Add one in-graph cost node (`ConstantCost`) and the net-R seam so +`aura run --harness stage1-r --cost-per-trade C` yields a `net_r_equity` trace and a +cost-adjusted `net_expectancy_r`, with a no-cost run byte-identical to today. + +**Architecture:** `ConstantCost` (aura-std) emits a 3-field cost-in-R record +`{cost_in_r, cum_cost_in_r, open_cost_in_r}` isomorphic to PM's `{realized_r, +cum_realized_r, unrealized_r}`. Two consumers read that one stream: an in-graph +`net_r_equity` sink (`LinComb(4,[1,1,-1,-1])` + `Recorder`, the `r_equity` idiom plus +two cost terms) and the post-run `summarize_r`, whose scalar `round_trip_cost: f64` is +replaced by consuming the co-temporal cost stream. The cost layer is opt-in on the +run path (non-reduce); a no-cost run is unchanged. + +**Tech Stack:** aura-std (new node), aura-analysis (`summarize_r`), aura-cli +(`stage1_r_graph`, `run_stage1_r`, `persist_traces_r`, `RunArgs`/`parse_run_args`/ +`run_dispatch`). Cross-crate seam guarded by the existing `r_col` lockstep test. + +**Scope note (cycle-1 minimal cut, recorded on #148 J/K):** the run path only +(non-reduce / trace mode, where `net_r_equity` and the full cost stream wire cleanly). +`stage1_r_graph`'s sweep/walkforward/mc callers pass `None` (no cost): cost in the +reduce-mode sweep path + the OOS-pooling cost in `r_metrics_from_rs` are a deferred +follow-up cycle. Acceptance evidence is the JSON `net_expectancy_r` field + the +chartable `net_r_equity` trace — no human-readable pretty-printer is added (the spec's +console mock is illustrative; `aura run` emits JSON). + +**Files this plan creates or modifies:** + +- Create: `crates/aura-std/src/constant_cost.rs` — the `ConstantCost` node (4 PM-geometry + inputs → 3-field cost-in-R record, one `cost_per_trade` F64 param). +- Modify: `crates/aura-std/src/lib.rs:18-70` — register `mod constant_cost;` + re-export. +- Modify: `crates/aura-analysis/src/lib.rs:154-287` — `summarize_r` signature + net fold; + rewrite its cost unit test `:744-755`. +- Modify (migrate call sites, compile gate): all `summarize_r(...)` callers across + `crates/aura-analysis/src/lib.rs`, `crates/aura-composites/tests/risk_executor.rs`, + `crates/aura-engine/tests/stage1_r_e2e.rs`, + `crates/aura-engine/tests/streaming_reduction_equivalence.rs`, + `crates/aura-cli/src/main.rs`. +- Modify: `crates/aura-cli/src/main.rs` — `stage1_r_graph:2614-2698` (+ its 7 callers), + `run_stage1_r:2917-2971`, `persist_traces_r:2976-2992`, `RunArgs:3004-3008`, + `parse_run_args:3015-3074`, `run_dispatch:3078-3091`, `USAGE:3093`. +- Test: `crates/aura-engine/tests/stage1_r_e2e.rs` — rewrite the two cost properties + (`:194-241`) onto the node; add the in-graph↔post-run agreement assertion. +- Test: `crates/aura-cli/tests/cli_run.rs` — add a `net_r_equity` persistence + cost run + test (sibling of `:1599`); the no-cost C18 golden `:1686` stays byte-identical. + +--- + +## Task 1: `ConstantCost` node in aura-std + +**Files:** +- Create: `crates/aura-std/src/constant_cost.rs` +- Modify: `crates/aura-std/src/lib.rs:18-70` +- Test: `crates/aura-std/src/constant_cost.rs` (`#[cfg(test)] mod tests`) + +- [ ] **Step 1: Write the node + its failing unit tests (one diff, a new file)** + +Create `crates/aura-std/src/constant_cost.rs` with the node AND its tests. (RED here is +the test module referencing a node that does not yet compile cleanly; Step 2 verifies, +Step 3 is the node body. For a single new file the test and impl co-exist in the diff — +the orchestrator accepts RED on the new-file's first failing run, per the project's +RED-in-single-diff allowance.) + +```rust +//! `ConstantCost` — a flat round-trip cost charged once per closed trade, in R. +//! The first cost node of the C10 cost-model graph (#148): an ordinary downstream +//! node (C9) that reads the executor's trade-geometry and emits a cost-in-R record +//! isomorphic to `PositionManagement`'s R-triple — `cost_in_r` (charged on a close) +//! / `cum_cost_in_r` (its running sum) / `open_cost_in_r` (the open trade's would-be +//! cost, for the window-end trade `summarize_r` synthesises). R-pure: with flat-1R +//! sizing the cost is `cost_per_trade / |entry - stop|`; notional cancels (C10). + +use aura_core::{ + Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, + ScalarKind, +}; + +/// A flat per-trade cost in price units (`cost_per_trade`), emitted in R. Inputs are +/// four executor-exposed fields: `closed` (closed_this_cycle), `open`, `entry_price`, +/// `stop_price`. Emits `None` until the executor has produced a record this cycle. +pub struct ConstantCost { + cost_per_trade: f64, + cum: f64, + out: [Cell; 3], +} + +impl ConstantCost { + pub fn new(cost_per_trade: f64) -> Self { + Self { cost_per_trade, cum: 0.0, out: [Cell::from_f64(0.0); 3] } + } + + /// The param-generic recipe: one `cost_per_trade` F64 knob. + pub fn builder() -> PrimitiveBuilder { + PrimitiveBuilder::new( + "ConstantCost", + 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() }, + ], + 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: "cost_per_trade".into(), kind: ScalarKind::F64 }], + }, + |p| Box::new(ConstantCost::new(p[0].f64())), + ) + } +} + +impl Node for ConstantCost { + fn lookbacks(&self) -> Vec { + vec![1, 1, 1, 1] + } + + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { + // Withhold until the executor's geometry is present this cycle (price warmed up). + let ew = ctx.f64_in(2); + if ew.is_empty() { + return None; + } + let entry = ew[0]; + let stop = ctx.f64_in(3).get(0).unwrap_or(0.0); + let closed = ctx.bool_in(0).get(0).unwrap_or(false); + let open = ctx.bool_in(1).get(0).unwrap_or(false); + let latched = (entry - stop).abs(); + let per = if latched > 0.0 { self.cost_per_trade / 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 { + "ConstantCost".to_string() + } +} + +#[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 + ] + } + + #[test] + fn no_geometry_yet_withholds() { + let mut c = ConstantCost::new(2.0); + let inputs = cols(); // entry column empty + assert_eq!(c.eval(Ctx::new(&inputs, Timestamp(0))), None); + } + + #[test] + fn closed_charges_cost_per_trade_over_latched() { + let mut c = ConstantCost::new(2.0); + let mut inputs = cols(); + inputs[0].push(Scalar::bool(true)).unwrap(); // closed + inputs[1].push(Scalar::bool(false)).unwrap(); // not open + inputs[2].push(Scalar::f64(100.0)).unwrap(); // entry + inputs[3].push(Scalar::f64(96.0)).unwrap(); // stop -> latched 4.0 + // cost_in_r = 2.0/4.0 = 0.5; cum = 0.5; open_cost_in_r = 0.0 + assert_eq!( + c.eval(Ctx::new(&inputs, Timestamp(0))), + Some([Cell::from_f64(0.5), Cell::from_f64(0.5), Cell::from_f64(0.0)].as_slice()) + ); + } + + #[test] + fn open_emits_would_be_cost_not_charged_to_cum() { + let mut c = ConstantCost::new(2.0); + let mut inputs = cols(); + inputs[0].push(Scalar::bool(false)).unwrap(); // not closed + 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 + // cost_in_r = 0 (no close); cum stays 0; open_cost_in_r = 0.5 + assert_eq!( + c.eval(Ctx::new(&inputs, Timestamp(0))), + Some([Cell::from_f64(0.0), Cell::from_f64(0.0), Cell::from_f64(0.5)].as_slice()) + ); + } + + #[test] + fn zero_latched_contributes_no_cost() { + let mut c = ConstantCost::new(2.0); + 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 + 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 = ConstantCost::new(2.0); + 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(); // 0.5 + 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.0 -> 1.0; cum 1.5 + assert_eq!( + c.eval(Ctx::new(&b, Timestamp(1))), + Some([Cell::from_f64(1.0), Cell::from_f64(1.5), Cell::from_f64(0.0)].as_slice()) + ); + } +} +``` + +- [ ] **Step 2: Register the module in aura-std** + +In `crates/aura-std/src/lib.rs`, add the `mod` line in the alphabetical block (between +`mod bias;` at `:20` and `mod delay;` at `:21`): + +```rust +mod constant_cost; +``` + +and the re-export in the `pub use` block (between `pub use bias::Bias;` at `:45` and +`pub use delay::Delay;` at `:46`): + +```rust +pub use constant_cost::ConstantCost; +``` + +- [ ] **Step 3: Run the node tests to verify they pass** + +Run: `cargo test -p aura-std constant_cost` +Expected: PASS — `closed_charges_cost_per_trade_over_latched`, +`open_emits_would_be_cost_not_charged_to_cum`, `zero_latched_contributes_no_cost`, +`cum_accumulates_across_closes`, `no_geometry_yet_withholds` all green. + +- [ ] **Step 4: Workspace builds clean with the new node** + +Run: `cargo build -p aura-std` +Expected: builds, 0 warnings (the node has no dead code; `Cell::from_f64(0.0); 3` +relies on `Cell: Copy` — if `Cell` is not `Copy`, use +`[Cell::from_f64(0.0), Cell::from_f64(0.0), Cell::from_f64(0.0)]` instead). + +--- + +## Task 2: `summarize_r` folds the cost stream (signature change + all call sites) + +This is a single compile unit: changing `summarize_r`'s signature breaks every caller +in the workspace, so **all** call sites are migrated in this task (a `cargo build +--workspace` 0-errors gate is unsatisfiable until they are). The fold's net token form +is preserved so a no-cost call is byte-identical. + +**Files:** +- Modify: `crates/aura-analysis/src/lib.rs:154-287` (signature, `Trade`, collection loop, + net fold) and `:744-755` (its cost unit test). +- Modify (migrate callers): `crates/aura-analysis/src/lib.rs` (19 test sites), + `crates/aura-composites/tests/risk_executor.rs` (4), `crates/aura-engine/tests/stage1_r_e2e.rs` (9), + `crates/aura-engine/tests/streaming_reduction_equivalence.rs` (2), + `crates/aura-cli/src/main.rs` (9 production). +- Test: the new exact-subsumption test in `crates/aura-analysis/src/lib.rs`. + +- [ ] **Step 1: Write the failing exact-subsumption test** + +Add to `crates/aura-analysis/src/lib.rs`'s `#[cfg(test)] mod tests`. It builds a tiny PM +record (one closed trade + one window-end open trade) and the co-temporal cost stream a +`ConstantCost(C)` node would emit, then asserts `net_expectancy_r` equals the +hand-computed `mean(rᵢ − C/latchedᵢ)` over BOTH trades — proving exact subsumption of +the old scalar (not just at C=0). (Helper `row14(...)` building a 14-wide PM row and the +cost row builder follow the existing test fixtures in this module; mirror the closest +existing `summarize_r` test's row construction at `:609-755`.) + +```rust +#[test] +fn summarize_r_folds_cost_stream_exactly_like_the_scalar_over_all_trades() { + // Trade 1: closed, entry 100 stop 96 (latched 4), realized R = 1.0. + // Trade 2: still open at window end, entry 100 stop 98 (latched 2), unrealized R = 0.5. + let c = 2.0_f64; + // PM record: (col0 closed, col1 realized_r, col4 dir, col6 entry, col7 stop, + // col9 conviction, col11 open, col12 unrealized_r). Use the module's row helper. + let record = vec![ + (Timestamp(0), pm_row(/*closed*/ true, /*r*/ 1.0, /*entry*/ 100.0, /*stop*/ 96.0, + /*conv*/ 1.0, /*open*/ false, /*unreal*/ 0.0)), + (Timestamp(1), pm_row(/*closed*/ false, /*r*/ 0.0, /*entry*/ 100.0, /*stop*/ 98.0, + /*conv*/ 1.0, /*open*/ true, /*unreal*/ 0.5)), + ]; + // Cost stream the ConstantCost(2.0) node emits, co-temporal: [cost_in_r, open_cost_in_r]. + let cost = vec![ + (Timestamp(0), vec![Scalar::f64(c / 4.0), Scalar::f64(0.0)]), // closed: 0.5 charged + (Timestamp(1), vec![Scalar::f64(0.0), Scalar::f64(c / 2.0)]), // open: would-be 1.0 + ]; + let m = summarize_r(&record, &cost); + // gross E[R] = mean(1.0, 0.5) = 0.75; net = mean(1.0-0.5, 0.5-1.0) = mean(0.5,-0.5) = 0.0 + assert!((m.expectancy_r - 0.75).abs() < 1e-12); + assert!((m.net_expectancy_r - 0.0).abs() < 1e-12); +} +``` + +(If no `pm_row` helper exists, add one to the test module: +`fn pm_row(closed: bool, r: f64, entry: f64, stop: f64, conv: f64, open: bool, unreal: f64) +-> Vec` returning a 14-element vector with the documented `r_col` positions set +and the rest `Scalar::f64(0.0)`/`Scalar::bool(false)`/`Scalar::i64(0)`/`Scalar::ts(Timestamp(0))` +per `PM_RECORD_KINDS`.) + +- [ ] **Step 2: Run the new test to verify it fails to compile** + +Run: `cargo test -p aura-analysis summarize_r_folds_cost_stream` +Expected: FAIL — compile error `this function takes ... arguments` / `expected f64, found +&[...]` at the call (the signature still takes `round_trip_cost: f64`). + +- [ ] **Step 3: Change `summarize_r`'s signature, `Trade`, the collection loop, and the net fold** + +In `crates/aura-analysis/src/lib.rs`: + +Signature `:154`: +```rust +pub fn summarize_r( + record: &[(Timestamp, Vec)], + cost: &[(Timestamp, Vec)], +) -> RMetrics { +``` + +`Trade` struct `:159-163` — add a per-trade `cost` term: +```rust + struct Trade { + r: f64, + bias_abs: f64, + latched: f64, + cost: f64, + } +``` + +Collection loop `:165-184` — **positional** join: the cost stream is co-temporal 1:1 +with the record (both recorded per cycle off the same executor cadence — `ConstantCost` +emits exactly when PM emits, so `record[i]` and `cost[i]` are the same cycle). A closed +row `i` reads `cost[i]`'s `cost_in_r` (col 0); the window-end open last row reads its +`open_cost_in_r` (col 1). An empty `cost` slice ⇒ every `.get(i)` is `None` ⇒ cost `0.0` +(the gross-R baseline). Positional avoids the shared-timestamp collision a `ts`-keyed map +would have (C4: same `ts` can be two cycles): +```rust + let mut trades: Vec = Vec::new(); + for (i, (_, row)) in record.iter().enumerate() { + if row[r_col::CLOSED].as_bool() { + let c = cost.get(i).map(|(_, cr)| cr[0].as_f64()).unwrap_or(0.0); + trades.push(Trade { + r: row[r_col::REALIZED_R].as_f64(), + bias_abs: row[r_col::CONVICTION_AT_ENTRY].as_f64(), + latched: (row[r_col::ENTRY_PRICE].as_f64() - row[r_col::STOP_PRICE].as_f64()).abs(), + cost: c, + }); + } + } + let mut n_open_at_end = 0u64; + if let Some((_, last)) = record.last() + && last[r_col::OPEN].as_bool() + { + let c = cost.get(record.len() - 1).map(|(_, cr)| cr[1].as_f64()).unwrap_or(0.0); + trades.push(Trade { + r: last[r_col::UNREALIZED_R].as_f64(), + bias_abs: last[r_col::CONVICTION_AT_ENTRY].as_f64(), + latched: (last[r_col::ENTRY_PRICE].as_f64() - last[r_col::STOP_PRICE].as_f64()).abs(), + cost: c, + }); + n_open_at_end = 1; + } +``` +(The empty-input early return at `:186-202` and `n == 0` are unchanged.) + +Net fold `:244-250` — subtract the per-trade cost the node supplied (empty stream ⇒ +every `t.cost == 0.0` ⇒ `net_sum == Σ t.r`, byte-identical to the old cost=0 result): +```rust + // net-of-cost: subtract the cost-model's per-trade cost-in-R (from the cost stream; + // an empty stream is the gross-R baseline — every trade's cost is 0.0). + let net_sum: f64 = trades.iter().map(|t| t.r - t.cost).sum(); + let net_expectancy_r = net_sum / n as f64; +``` +`r_metrics_from_rs` (`:301-353`) is **unchanged** — it carries no cost (the pooled-OOS +copy stays `net_expectancy_r = expectancy_r` under the cost=0 invariant `:349`). + +- [ ] **Step 4: Migrate the cost unit test `summarize_r_net_of_cost_subtracts_round_trip_per_trade` (`:744-755`)** + +Rewrite it to drive cost via the stream (the node's emission), preserving its assertion +that a per-trade round-trip cost is charged in R. Replace the `summarize_r(&record, 1.0)` +call with a constructed cost stream that emits `1.0/latched` on each closed row (and the +window-end `open_cost_in_r` if the fixture leaves a position open), and assert the same +net result the old scalar produced. Keep the fixture's trades; only the cost-injection +mechanism changes. + +- [ ] **Step 5: Migrate every other `summarize_r` call site to the new signature** + +The cost-0 sites take the empty stream; verify with grep first (Step 6). Mechanical +transform `summarize_r(, 0.0)` → `summarize_r(, &[])` at: +- `crates/aura-analysis/src/lib.rs`: `:609, :627, :638, :647, :660, :667, :670, :681, :702, :719, :734, :741, :749` (the `,0.0` sites; the cross-reducer test `:948` too). +- `crates/aura-composites/tests/risk_executor.rs`: `:97, :141, :178, :210`. +- `crates/aura-engine/tests/stage1_r_e2e.rs`: `:80, :121, :207, :403` and the first call at `:232` and `:235`. +- `crates/aura-engine/tests/streaming_reduction_equivalence.rs`: `:63, :78`. +- `crates/aura-cli/src/main.rs` (production): `:1266, :1278, :1340, :1381, :1437, :1448, :1509, :1520, :2969` → `summarize_r(, &[])`. + +The **non-zero** cost test sites — `crates/aura-analysis/src/lib.rs:750` (`,1.0`, handled +in Step 4) and `crates/aura-engine/tests/stage1_r_e2e.rs:208` (`,2.0`) and the second +calls at `:232`/`:235` (`,2.0`) — are rewritten in Task-2 Step 4 / Task-3 Step 7 to +construct a `2.0`-cost stream aligned to their trade record (build +`[(ts, [2.0/latched_at_ts, open_cost_at_ts]), …]`) and assert the same net. + +- [ ] **Step 6: Verify no `summarize_r(... , )` call survives** + +Run: `git grep -nE 'summarize_r\([^)]*,\s*[0-9]' -- 'crates/**/*.rs'` +Expected: no output (every call now passes a `&[...]` cost stream; the `, 0.0`/`, 1.0`/ +`, 2.0` second arguments are gone). A hit is an un-migrated site — fix it. + +- [ ] **Step 7: Build the workspace — the compile gate** + +Run: `cargo build --workspace --all-targets` +Expected: `0 errors` (every `summarize_r` caller migrated). + +- [ ] **Step 8: Run the analysis + cross-crate suites** + +Run: `cargo test -p aura-analysis` +Expected: PASS incl. `summarize_r_folds_cost_stream_exactly_like_the_scalar_over_all_trades`, +the migrated cost test, and `summarize_r_includes_open_trade_and_matches_r_metrics_from_rs` +(cross-reducer equality holds at cost=0). + +--- + +## Task 3: Run-path wiring — `net_r_equity` tap, cost stream, `--cost-per-trade` + +**Files:** +- Modify: `crates/aura-cli/src/main.rs` — `stage1_r_graph:2614-2698` (+ its 7 call sites), + `run_stage1_r:2917-2971`, `persist_traces_r:2976-2992`, `RunArgs:3004-3008`, + `parse_run_args:3015-3074`, `run_dispatch:3078-3091`, `USAGE:3093`. +- Test: `crates/aura-engine/tests/stage1_r_e2e.rs` (in-graph↔post-run agreement), + `crates/aura-cli/tests/cli_run.rs` (net_r_equity persistence; no-cost golden held). + +- [ ] **Step 1: Write the failing CLI cost-run test** + +Add to `crates/aura-cli/tests/cli_run.rs` (sibling of +`stage1_r_trace_persists_r_equity_and_charts_it:1599`). Run `aura run --harness stage1-r +--cost-per-trade 2 --trace `, assert (a) `runs/traces//net_r_equity.json` exists, +(b) the run JSON has `net_expectancy_r` strictly less than `expectancy_r`. (Mirror the +existing test's harness invocation + `runs/traces/.../r_equity.json` assertion.) + +```rust +#[test] +fn stage1_r_cost_run_persists_net_r_equity_and_charges_cost() { + let tmp = tempdir().unwrap(); + let out = run_cli(&tmp, &["run", "--harness", "stage1-r", "--cost-per-trade", "2", "--trace", "c1"]); + assert!(tmp.path().join("runs/traces/c1/net_r_equity.json").exists(), + "net_r_equity trace persisted"); + let v: serde_json::Value = serde_json::from_str(&out).unwrap(); + let gross = v["metrics"]["r"]["expectancy_r"].as_f64().unwrap(); + let net = v["metrics"]["r"]["net_expectancy_r"].as_f64().unwrap(); + assert!(net < gross, "cost charged: net {net} < gross {gross}"); +} +``` + +(Use the test module's existing CLI-invocation helper and temp-dir pattern; match the +`run_cli` / output-capture shape already used at `:1599-1621`.) + +- [ ] **Step 2: Run it — fails (no flag, no tap yet)** + +Run: `cargo test -p aura-cli stage1_r_cost_run_persists_net_r_equity` +Expected: FAIL — `--cost-per-trade` is an unknown flag (`parse_run_args` returns the +usage error) / the run exits non-zero. + +- [ ] **Step 3: Add the cost taps to `stage1_r_graph`** + +Add a `cost` parameter (the bundle) to `stage1_r_graph` and wire the node + net tap when +present and not in reduce mode. New signature (append after `reduce: bool`): +```rust + cost: Option<(f64, mpsc::Sender<(Timestamp, Vec)>, mpsc::Sender<(Timestamp, Vec)>)>, +``` +Inside the existing `if !reduce { … }` block (after the `r_equity` wiring at `:2695`), +add: +```rust + 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 + 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(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]")); + 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 the post-run fold reads: [cost_in_r, open_cost_in_r] + let cost_rec = g.add( + Recorder::builder(vec![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("open_cost_in_r"), cost_rec.input("col[1]")); + } +``` +Add `use aura_std::ConstantCost;` to the imports if not already pulled via the existing +`aura_std::{…}` use (LinComb/Recorder are already imported). + +- [ ] **Step 4: Thread `None` through every non-run `stage1_r_graph` caller** + +`stage1_r_graph` is called from the sweep/walkforward/mc grid builders. Enumerate them +(line numbers may have drifted — grep is the authoritative channel): + +Run: `git grep -n 'stage1_r_graph(' -- crates/aura-cli/src/main.rs` +Expected (recon estimate): `:1202, :1236, :1294, :1308, :1324, :1359` (grid builders) plus +`:2928` (the run caller, updated in Step 5). Append `, None` to each grid-builder call +(cycle-1 scope: no cost on the reduce-mode sweep path). A call this step misses is a hard +compile error at the Step-9/10 gate. + +- [ ] **Step 5: Thread cost through `run_stage1_r`** + +Change `run_stage1_r`'s signature to `fn run_stage1_r(data: RunData, trace: Option<&str>, +cost: Option) -> RunReport`. Create the two extra channels, pass the bundle when +cost is set, drain the streams, fold cost into `summarize_r`, and persist `net_r_equity`: +```rust + let (tx_eq, rx_eq) = mpsc::channel(); + let (tx_ex, rx_ex) = mpsc::channel(); + let (tx_r, rx_r) = mpsc::channel(); + 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 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"); +``` +After `h.run(sources)` and the existing drains, add: +```rust + let net_rows: Vec<(Timestamp, Vec)> = rx_net.try_iter().collect(); + let cost_rows: Vec<(Timestamp, Vec)> = rx_cost.try_iter().collect(); +``` +Persist + fold (replace `:2966-2969`): +```rust + if let Some(name) = trace { + persist_traces_r(name, &manifest, &eq_rows, &ex_rows, &req_rows, &net_rows); + } + let mut metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0)); + metrics.r = Some(summarize_r(&r_rows, &cost_rows)); +``` +(When `cost` is `None`, `cost_bundle` is `None` ⇒ no cost node / no `net_rows` / `cost_rows` +empty ⇒ `summarize_r(&r_rows, &[])` ⇒ net == gross, and `persist_traces_r` gets an empty +`net_rows`.) + +- [ ] **Step 6: Add the `net_r_equity` trace to `persist_traces_r`** + +Add a `net_rows: &[(Timestamp, Vec)]` parameter and emit the tap only when +non-empty (so a no-cost run's on-disk trace set is byte-unchanged): +```rust +fn persist_traces_r( + name: &str, + manifest: &RunManifest, + eq_rows: &[(Timestamp, Vec)], + ex_rows: &[(Timestamp, Vec)], + req_rows: &[(Timestamp, Vec)], + net_rows: &[(Timestamp, Vec)], +) { + let mut taps = vec![ + ColumnarTrace::from_rows("equity", &[ScalarKind::F64], eq_rows), + ColumnarTrace::from_rows("exposure", &[ScalarKind::F64], ex_rows), + ColumnarTrace::from_rows("r_equity", &[ScalarKind::F64], req_rows), + ]; + if !net_rows.is_empty() { + taps.push(ColumnarTrace::from_rows("net_r_equity", &[ScalarKind::F64], net_rows)); + } + if let Err(e) = TraceStore::open("runs").write(name, manifest, &taps) { + eprintln!("aura: trace persist failed: {e}"); + std::process::exit(2); + } +} +``` + +- [ ] **Step 7: Rewrite the e2e cost properties + add the in-graph↔post-run agreement assertion** + +In `crates/aura-engine/tests/stage1_r_e2e.rs:194-241`, rewrite the two scalar-cost +properties (`net_of_cost_charges_one_round_trip_per_trade_through_the_recovered_latched_dist` +and `net_of_cost_is_charged_per_r_so_a_wider_stop_dilutes_it`) to construct a cost stream +(the `ConstantCost` emission for the test's trades) and pass it to `summarize_r` — same +assertions, cost now sourced from the node/stream not the scalar. Add an assertion that +the final `net_r_equity` sample equals `cum_realized_r + unrealized_r − total_cost` (the +post-run net total) — the in-graph↔post-run agreement. (If building the full graph is +heavy here, assert the algebraic identity on the constructed records directly.) + +- [ ] **Step 8: Add `--cost-per-trade` to `RunArgs` / `parse_run_args` / `run_dispatch` / `USAGE`** + +`RunArgs` `:3004-3008`: +```rust +struct RunArgs { + harness: HarnessKind, + data: RunData, + trace: Option, + cost: Option, +} +``` +`parse_run_args` — declare `let mut cost: Option = None;` near `:3024`, add a flag +arm (alongside `--trace` at `:3056`): +```rust + "--cost-per-trade" if cost.is_none() => { + let (value, t) = t.split_first().ok_or_else(usage)?; + cost = Some(value.parse().map_err(|_| usage())?); + tail = t; + } +``` +and add `cost` to the `Ok(RunArgs { … })` at `:3073`. Update the `usage` string `:3017` +to include `[--cost-per-trade ]`. + +`run_dispatch` `:3083` — thread it into the stage1-r arm (other harness arms ignore cost +this cycle): +```rust + (HarnessKind::Stage1R, data) => run_stage1_r(data, trace, args.cost), +``` +Update the top-level `USAGE` const `:3093` to add `[--cost-per-trade ]` to the +`aura run` synopsis. + +Thread the **other** `run_stage1_r` caller too — the test at `crates/aura-cli/src/main.rs:4311` +calls `run_stage1_r(...)` and must pass the new arg or the build breaks. Find both callers +and update them: `git grep -n 'run_stage1_r(' -- crates/aura-cli/src/main.rs` (expect the +dispatch `:3083` and the test `:4311`); the test caller passes `None` (no cost) unless it +specifically asserts a cost run. + +- [ ] **Step 9: Run the run-path tests** + +Run: `cargo test -p aura-cli stage1_r_cost_run_persists_net_r_equity` +Expected: PASS — `net_r_equity.json` persisted; `net < gross`. + +Run: `cargo test -p aura-cli stage1_r_single_run_output_golden` +Expected: PASS, byte-identical — the no-cost golden at `cli_run.rs:1686` +(`"net_expectancy_r":1.2710005136982836`) is unchanged (no-cost run ⇒ `summarize_r(&r_rows, +&[])` ⇒ net == gross, no `net_r_equity` tap). + +- [ ] **Step 10: Full regression gate** + +Run: `cargo test --workspace` +Expected: PASS, the full suite green (the 665-test baseline plus the new tests), no +golden moved. + +Run: `cargo clippy --workspace --all-targets -- -D warnings` +Expected: clean.