From 3fe4684190f623592ba7ff1656bf9eb890ca032e Mon Sep 17 00:00:00 2001 From: Brummel Date: Sun, 28 Jun 2026 14:41:45 +0200 Subject: [PATCH] =?UTF-8?q?feat(0081):=20cost-model=20graph=20cycle=201=20?= =?UTF-8?q?=E2=80=94=20ConstantCost=20node=20+=20net-R=20seam?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lands the first cycle of the "Cost-model graph (in R)" milestone: cost is now a composable in-graph node, not a post-run scalar (C10). - ConstantCost (aura-std): a flat round-trip cost per trade, emitted in R as a 3-field record {cost_in_r, cum_cost_in_r, open_cost_in_r} isomorphic to PositionManagement's R-triple. R-pure: cost_in_r = cost_per_trade / |entry-stop|; notional cancels, no equity held. - summarize_r folds the cost stream into net_expectancy_r, replacing its scalar round_trip_cost (one home for cost, no double-count). Positional co-temporal join; an empty stream is the gross-R baseline. - net_r_equity tap (stage1_r_graph): LinComb(4) over [cum_realized_r, unrealized_r, -cum_cost_in_r, -open_cost_in_r] -> Recorder, a sibling of r_equity; --cost-per-trade on the run path wires the node + tap + fold. - A no-cost run is byte-identical: no net_r_equity tap, net == gross, the C18 golden unchanged. Tests: exact-subsumption (the node-stream net == mean(r_i - cost_i) over ALL trades incl. the window-end open one), the in-graph net_r_equity tap + net < gross E2E, the no-cost golden held. Full suite green; clippy clean. Implement-time decisions ratified (recorded on #148): the cost stream is recorded 3-wide (the node's full output; summarize_r reads col 0/2, col 1 feeds the in-graph net curve) — resolving a 2-wide/3-wide inconsistency in the plan's Step 3; Trade drops its `latched` field (cost now arrives in R), so r_col ENTRY_PRICE/STOP_PRICE become #[allow(dead_code)] layout contract. Scoped to the run path (non-reduce); sweep/walkforward/mc pass None — cost on the reduce-mode sweep path and OOS-pooling cost are a deferred follow-up. Known pre-existing concern (not this cycle): aura-std lib.rs module doc still names "realistic broker profiles", retired by the C10 rework. refs #148 --- crates/aura-analysis/src/lib.rs | 148 ++++++++++---- crates/aura-cli/src/main.rs | 142 +++++++++---- crates/aura-cli/tests/cli_run.rs | 26 +++ crates/aura-composites/tests/risk_executor.rs | 8 +- crates/aura-engine/tests/stage1_r_e2e.rs | 109 +++++++++- .../tests/streaming_reduction_equivalence.rs | 4 +- crates/aura-std/src/constant_cost.rs | 189 ++++++++++++++++++ crates/aura-std/src/lib.rs | 2 + 8 files changed, 539 insertions(+), 89 deletions(-) create mode 100644 crates/aura-std/src/constant_cost.rs diff --git a/crates/aura-analysis/src/lib.rs b/crates/aura-analysis/src/lib.rs index 30d90a5..91b5f81 100644 --- a/crates/aura-analysis/src/lib.rs +++ b/crates/aura-analysis/src/lib.rs @@ -135,7 +135,13 @@ mod r_col { pub const CLOSED: usize = 0; pub const REALIZED_R: usize = 1; pub const DIRECTION: usize = 4; + // entry/stop define the latched distance (1R). Since the net-of-cost fold reads the + // cost-in-R straight from the cost stream (the aura-std `ConstantCost` producer recovers + // the distance), this crate's reductions no longer touch these two columns; they stay in + // the documented layout contract and are exercised by the test fixtures. + #[allow(dead_code)] pub const ENTRY_PRICE: usize = 6; + #[allow(dead_code)] pub const STOP_PRICE: usize = 7; pub const CONVICTION_AT_ENTRY: usize = 9; pub const SIZE: usize = 10; @@ -143,6 +149,17 @@ mod r_col { pub const UNREALIZED_R: usize = 12; } +// Cost-model record column indices — the lockstep contract with aura-std's +// `ConstantCost` output schema (`cost_in_r` / `cum_cost_in_r` / `open_cost_in_r`). +// Read positionally across the crate boundary as a type-erased `Scalar` vector +// (C4 SoA), like `r_col`. `summarize_r` folds the per-close cost (col 0) and the +// window-end open trade's would-be cost (col 2); the running sum (col 1) is not +// folded, so it is intentionally not named here. +mod cost_col { + pub const COST_IN_R: usize = 0; + pub const OPEN_COST_IN_R: usize = 2; +} + /// Van Tharp's conventional trade-count cap for the n-normalized SQN ("SQN /// score"): capping n stops the single-number objective rewarding sheer turnover. const SQN_CAP: u64 = 100; @@ -151,23 +168,34 @@ const SQN_CAP: u64 = 100; /// The trade ledger is the rows where `closed_this_cycle`; a position still open on the /// last row is force-closed at its `unrealized_r` (a window-end trade — never silently /// folded as unrealised MtM). Empty input -> a well-defined all-zero `RMetrics`. -pub fn summarize_r(record: &[(Timestamp, Vec)], round_trip_cost: f64) -> RMetrics { +pub fn summarize_r( + record: &[(Timestamp, Vec)], + cost: &[(Timestamp, Vec)], +) -> RMetrics { // Collect one entry per trade: its realised R, the entry-conviction |bias|, and the - // latched R-distance (|entry - stop|, the frozen R-denominator) recovered from the - // record. The ledger is the closed rows; a position still open on the last row is - // force-closed at its unrealized R (a window-end trade). + // per-trade cost-in-R the cost-model graph supplied (read positionally from the + // co-temporal `cost` stream). The ledger is the closed rows; a position still open on + // the last row is force-closed at its unrealized R (a window-end trade). struct Trade { r: f64, bias_abs: f64, - latched: f64, + cost: f64, } + // Positional join: the cost stream is co-temporal 1:1 with the record (both recorded + // per cycle off the same executor cadence — a cost node 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 2). + // 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). let mut trades: Vec = Vec::new(); - for (_, row) in record { + for (i, (_, row)) in record.iter().enumerate() { if row[r_col::CLOSED].as_bool() { + let c = cost.get(i).map(|(_, cr)| cr[cost_col::COST_IN_R].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, }); } } @@ -175,10 +203,14 @@ pub fn summarize_r(record: &[(Timestamp, Vec)], round_trip_cost: f64) -> if let Some((_, last)) = record.last() && last[r_col::OPEN].as_bool() { + let c = cost + .get(record.len() - 1) + .map(|(_, cr)| cr[cost_col::OPEN_COST_IN_R].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; } @@ -241,12 +273,9 @@ pub fn summarize_r(record: &[(Timestamp, Vec)], round_trip_cost: f64) -> (0.0, 0.0) } }; - // net-of-cost: subtract one round-trip spread (price units) per trade, expressed in R - // by dividing by that trade's latched R-distance (a zero distance contributes no cost). - let net_sum: f64 = trades - .iter() - .map(|t| t.r - if t.latched > 0.0 { round_trip_cost / t.latched } else { 0.0 }) - .sum(); + // 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; // conviction terciles: sort by conviction_at_entry ascending, split into three contiguous // near-equal-count buckets (floor boundaries i*n/3), E[R] per bucket. < 3 trades -> 0s. @@ -604,9 +633,26 @@ mod tests { v[r_col::OPEN] = Scalar::bool(false); (Timestamp(0), v) } + + // A 14-wide PositionManagement dense row carrying BOTH the closed-trade geometry + // (entry/stop/conviction) AND the open-at-end fields (open/unrealized_r) — the one + // fixture the cost-stream subsumption test needs (`r_row_full` forces closed=true, so + // it cannot build a window-end open row). Only the columns summarize_r reads are set; + // the rest default to f64(0.0) (never read), mirroring the existing row helpers. + fn row14(closed: bool, r: f64, entry: f64, stop: f64, conv: f64, open: bool, unreal: f64) -> Vec { + let mut v = vec![Scalar::f64(0.0); 14]; + v[r_col::CLOSED] = Scalar::bool(closed); + v[r_col::REALIZED_R] = Scalar::f64(r); + v[r_col::ENTRY_PRICE] = Scalar::f64(entry); + v[r_col::STOP_PRICE] = Scalar::f64(stop); + v[r_col::CONVICTION_AT_ENTRY] = Scalar::f64(conv); + v[r_col::OPEN] = Scalar::bool(open); + v[r_col::UNREALIZED_R] = Scalar::f64(unreal); + v + } #[test] fn summarize_r_is_zero_on_empty() { - let m = summarize_r(&[], 0.0); + let m = summarize_r(&[], &[]); assert_eq!(m.n_trades, 0); assert_eq!(m.expectancy_r, 0.0); assert_eq!(m.max_r_drawdown, 0.0); @@ -624,7 +670,7 @@ mod tests { r_row(true, -1.0, true, 0.0), r_row(true, 1.0, false, 0.0), // last row not open ]; - let m = summarize_r(&rec, 0.0); + let m = summarize_r(&rec, &[]); assert_eq!(m.n_trades, 3); assert!((m.expectancy_r - (2.0 / 3.0)).abs() < 1e-9); assert!((m.win_rate - (2.0 / 3.0)).abs() < 1e-9); @@ -635,7 +681,7 @@ mod tests { fn summarize_r_force_closes_open_position_at_window_end() { // one closed +1, then last row open with unrealized -0.5 -> a window-end trade. let rec = vec![r_row(true, 1.0, true, 0.0), r_row(false, 0.0, true, -0.5)]; - let m = summarize_r(&rec, 0.0); + let m = summarize_r(&rec, &[]); assert_eq!(m.n_trades, 2); assert_eq!(m.n_open_at_end, 1); assert!((m.expectancy_r - 0.25).abs() < 1e-9); // (1 + -0.5)/2 @@ -644,7 +690,7 @@ mod tests { fn summarize_r_max_drawdown_on_by_trade_curve() { // cum: +3, +1 (dd 2), +4 -> max dd = 2. let rec = vec![r_row(true, 3.0, false, 0.0), r_row(true, -2.0, false, 0.0), r_row(true, 3.0, false, 0.0)]; - assert!((summarize_r(&rec, 0.0).max_r_drawdown - 2.0).abs() < 1e-9); + assert!((summarize_r(&rec, &[]).max_r_drawdown - 2.0).abs() < 1e-9); } #[test] @@ -657,17 +703,17 @@ mod tests { r_row(true, 1.0, false, 0.0), r_row(true, -1.0, false, 0.0), ]; - let m = summarize_r(&rec, 0.0); + let m = summarize_r(&rec, &[]); assert!((m.sqn - 1.0).abs() < 1e-9, "sqn = sqrt(n)*mean/sd; got {}", m.sqn); } #[test] fn summarize_r_sqn_zero_when_under_two_trades_or_zero_variance() { // n < 2 -> 0 - assert_eq!(summarize_r(&[r_row(true, 2.0, false, 0.0)], 0.0).sqn, 0.0); + assert_eq!(summarize_r(&[r_row(true, 2.0, false, 0.0)], &[]).sqn, 0.0); // identical R -> zero variance -> 0 let flat = vec![r_row(true, 1.0, false, 0.0), r_row(true, 1.0, false, 0.0), r_row(true, 1.0, false, 0.0)]; - assert_eq!(summarize_r(&flat, 0.0).sqn, 0.0); + assert_eq!(summarize_r(&flat, &[]).sqn, 0.0); } #[test] @@ -678,7 +724,7 @@ mod tests { let rec: Vec<_> = (0..144) .map(|i| r_row(true, if i % 2 == 0 { 1.0 } else { 2.0 }, false, 0.0)) .collect(); - let m = summarize_r(&rec, 0.0); + let m = summarize_r(&rec, &[]); assert_eq!(m.n_trades, 144); assert!(m.sqn_normalized > 0.0, "sqn_normalized nonzero; got {}", m.sqn_normalized); assert!(m.sqn_normalized < m.sqn, "capped SQN100 < raw sqn for n > 100"); @@ -699,7 +745,7 @@ mod tests { r_row(true, 1.0, false, 0.0), r_row(true, -1.0, false, 0.0), ]; - let m = summarize_r(&rec, 0.0); + let m = summarize_r(&rec, &[]); assert!((m.sqn_normalized - m.sqn).abs() < 1e-12, "below cap, SQN100 == sqn"); assert!((m.sqn_normalized - 1.0).abs() < 1e-9, "sqn_normalized = 1.0 here"); } @@ -716,7 +762,7 @@ mod tests { r_row_full(1.0, 100.0, 99.0, 0.9), r_row_full(2.0, 100.0, 99.0, 1.0), ]; - let m = summarize_r(&calibrated, 0.0); + let m = summarize_r(&calibrated, &[]); assert!((m.conviction_terciles_r[0] - (-1.5)).abs() < 1e-9, "low tercile: {:?}", m.conviction_terciles_r); assert!((m.conviction_terciles_r[2] - 1.5).abs() < 1e-9, "high tercile: {:?}", m.conviction_terciles_r); assert!(m.conviction_terciles_r[2] > m.conviction_terciles_r[0], "high conviction must out-earn low when calibrated"); @@ -731,29 +777,61 @@ mod tests { r_row_full(-1.0, 100.0, 99.0, 0.9), r_row_full(-2.0, 100.0, 99.0, 1.0), ]; - let a = summarize_r(&anti, 0.0); + let a = summarize_r(&anti, &[]); assert!(a.conviction_terciles_r[2] < a.conviction_terciles_r[0], "anti-calibrated must invert: {:?}", a.conviction_terciles_r); } #[test] fn summarize_r_conviction_terciles_zero_under_three_trades() { let rec = vec![r_row_full(2.0, 100.0, 99.0, 0.5), r_row_full(-1.0, 100.0, 99.0, 0.5)]; - assert_eq!(summarize_r(&rec, 0.0).conviction_terciles_r, [0.0; 3]); + assert_eq!(summarize_r(&rec, &[]).conviction_terciles_r, [0.0; 3]); } #[test] fn summarize_r_net_of_cost_subtracts_round_trip_per_trade() { - // one trade: R=+2, entry 100, stop 90 -> latched 10. round_trip_cost 1.0 (price - // units) -> cost in R = 1/10 = 0.1. gross E[R]=2.0, net = 1.9. + // one closed trade: R=+2, entry 100, stop 90 -> latched 10. A ConstantCost(1.0) + // node (1.0 price units) emits cost_in_r = 1.0/10 = 0.1 on the closed row (col 0). + // gross E[R]=2.0, net = 1.9. The empty stream is the gross baseline (net == gross). let rec = vec![r_row_full(2.0, 100.0, 90.0, 0.5)]; - let gross = summarize_r(&rec, 0.0); - let net = summarize_r(&rec, 1.0); + let gross = summarize_r(&rec, &[]); + // producer 3-wide row [cost_in_r, cum_cost_in_r, open_cost_in_r]: closed -> 0.1 / 0.1 / 0. + let cost = vec![(Timestamp(0), vec![Scalar::f64(0.1), Scalar::f64(0.1), Scalar::f64(0.0)])]; + let net = summarize_r(&rec, &cost); assert!((gross.expectancy_r - 2.0).abs() < 1e-9); assert!((gross.net_expectancy_r - 2.0).abs() < 1e-9, "zero cost -> net == gross"); assert!((net.expectancy_r - 2.0).abs() < 1e-9, "cost never changes gross expectancy"); assert!((net.net_expectancy_r - 1.9).abs() < 1e-9, "net = 2 - 1/10; got {}", net.net_expectancy_r); } + /// Property: `summarize_r` folding the cost stream reproduces the retired scalar + /// `round_trip_cost` argument EXACTLY — `net_expectancy_r == mean(rᵢ − costᵢ)` over + /// BOTH the closed trade (cost from the stream's col 0) and the window-end open trade + /// (cost from col 2), with the cost-in-R taken straight from the co-temporal stream a + /// `ConstantCost` node emits (not recomputed from latched). Exercises a nonzero cost, + /// so it proves exact subsumption — not merely the cost=0 baseline. + #[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; + let record = vec![ + (Timestamp(0), row14(true, 1.0, 100.0, 96.0, 1.0, false, 0.0)), + (Timestamp(1), row14(false, 0.0, 100.0, 98.0, 1.0, true, 0.5)), + ]; + // Cost stream the ConstantCost(2.0) node emits, co-temporal — the producer's + // 3-wide row [cost_in_r, cum_cost_in_r, open_cost_in_r]. + let cost = vec![ + // closed row: 0.5 charged on close (col 0), cum 0.5, no open cost. + (Timestamp(0), vec![Scalar::f64(c / 4.0), Scalar::f64(c / 4.0), Scalar::f64(0.0)]), + // open row: no close cost, cum unchanged, would-be open cost 1.0 (col 2). + (Timestamp(1), vec![Scalar::f64(0.0), Scalar::f64(c / 4.0), Scalar::f64(c / 2.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); + } + // One PositionManagement dense record row for the derive tests: only the columns // derive_position_events reads (closed, direction, size, open) are set; the rest // default to 0. Distinct per-row timestamps (the derive keys events on the row's @@ -923,14 +1001,14 @@ mod tests { // two closed trades (R = +2.0, then -1.0) over a minimal PositionManagement // record; trade_rs must carry [2.0, -1.0] in trade order. let rec = pm_record_two_closed_trades(); // helper below - let m = summarize_r(&rec, 0.0); + let m = summarize_r(&rec, &[]); assert_eq!(m.trade_rs, vec![2.0, -1.0]); assert_eq!(m.n_trades, 2); } #[test] fn summarize_r_empty_record_has_empty_trade_rs() { - let m = summarize_r(&[], 0.0); + let m = summarize_r(&[], &[]); assert!(m.trade_rs.is_empty()); assert_eq!(m.n_trades, 0); } @@ -945,7 +1023,7 @@ mod tests { #[test] fn summarize_r_includes_open_trade_and_matches_r_metrics_from_rs() { let rec = pm_record_closed_then_open_at_end(); - let m = summarize_r(&rec, 0.0); + let m = summarize_r(&rec, &[]); assert_eq!(m.trade_rs, vec![2.0, 0.5]); assert_eq!(m.n_trades, 2); assert_eq!(m.n_open_at_end, 1); @@ -969,7 +1047,7 @@ mod tests { // (trade_rs is an in-memory conduit, excluded from equality) — this is what // keeps serialize->deserialize round-trips equal (trade_rs is serde-skipped, // so it deserializes empty). - let a = summarize_r(&pm_record_two_closed_trades(), 0.0); + let a = summarize_r(&pm_record_two_closed_trades(), &[]); let mut b = a.clone(); b.trade_rs = Vec::new(); assert_eq!(a, b); @@ -979,7 +1057,7 @@ mod tests { fn populated_trade_rs_is_absent_from_serialized_json() { // the in-memory conduit must never reach the wire: a populated trade_rs is // dropped by #[serde(skip)], so the C18 on-disk shape is byte-unperturbed (#139). - let m = summarize_r(&pm_record_two_closed_trades(), 0.0); + let m = summarize_r(&pm_record_two_closed_trades(), &[]); assert_eq!(m.trade_rs, vec![2.0, -1.0], "precondition: trade_rs is populated"); let json = serde_json::to_string(&m).expect("RMetrics serializes"); assert!(!json.contains("trade_rs"), "trade_rs must not reach the wire: {json}"); diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index bd7ca3c..1bdf36e 100644 --- a/crates/aura-cli/src/main.rs +++ b/crates/aura-cli/src/main.rs @@ -29,8 +29,9 @@ use aura_registry::{ FamilyMember, Generalization, NameKind, PlateauMode, Registry, RunTraces, TraceStore, WriteKind, }; use aura_std::{ - Add, Bias, Delay, Ema, GatedRecorder, Gt, Latch, LinComb, LongOnly, Mul, Recorder, RollingMax, - RollingMin, SeriesReducer, SimBroker, Sma, Sqrt, Sub, PM_FIELD_NAMES, PM_RECORD_KINDS, + Add, Bias, ConstantCost, Delay, Ema, GatedRecorder, Gt, Latch, LinComb, LongOnly, Mul, Recorder, + RollingMax, RollingMin, SeriesReducer, SimBroker, Sma, Sqrt, Sub, PM_FIELD_NAMES, + PM_RECORD_KINDS, }; use std::sync::mpsc::{self, Receiver}; use std::sync::LazyLock; @@ -1199,7 +1200,7 @@ fn stage1_r_sweep_family(trace: Option<&str>, data: &DataSource, grid: &Stage1RG let (tx_ex, _) = mpsc::channel(); let (tx_r, _) = mpsc::channel(); let (tx_req, _) = mpsc::channel(); - let bp = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, true); + let bp = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, true, None); let space = bp.param_space(); // resolve the open stop slots' exact path-qualified names from the live param-space // (the composite prefix is never hand-synced — match by the stable suffix). @@ -1233,7 +1234,7 @@ fn stage1_r_sweep_family(trace: Option<&str>, data: &DataSource, grid: &Stage1RG let (tx_r, rx_r) = mpsc::channel(); let (tx_req, rx_req) = mpsc::channel(); let reduce = trace.is_none(); - let mut h = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, reduce) + let mut h = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, reduce, None) .bootstrap_with_cells(point) .expect("stage1-r grid points are kind-checked against param_space"); h.run(data.run_sources()); @@ -1263,7 +1264,7 @@ fn stage1_r_sweep_family(trace: Option<&str>, data: &DataSource, grid: &Stage1RG let bias_sign_flips = rx_ex.try_iter().next().map(|(_, row)| row[2].as_i64() as u64).unwrap_or(0); let mut m = RunMetrics { total_pips, max_drawdown, bias_sign_flips, r: None }; - m.r = Some(summarize_r(&r_rows, 0.0)); + m.r = Some(summarize_r(&r_rows, &[])); m } else { // trace path (--trace set): raw recorders, persist, summarize — today's code. @@ -1272,10 +1273,10 @@ fn stage1_r_sweep_family(trace: Option<&str>, data: &DataSource, grid: &Stage1RG let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); let req_rows: Vec<(Timestamp, Vec)> = rx_req.try_iter().collect(); if let Some(name) = trace { - persist_traces_r(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows, &req_rows); + persist_traces_r(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows, &req_rows, &[]); } let mut m = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0)); - m.r = Some(summarize_r(&r_rows, 0.0)); + m.r = Some(summarize_r(&r_rows, &[])); m }; RunReport { manifest, metrics } @@ -1291,7 +1292,7 @@ fn stage1_r_space() -> Vec { let (tx_ex, _) = mpsc::channel(); let (tx_r, _) = mpsc::channel(); let (tx_req, _) = mpsc::channel(); - stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, true).param_space() + stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, true, None).param_space() } /// Windowed reduce-mode stage1-r sweep over `[from, to]` — the in-sample leg of the @@ -1305,7 +1306,7 @@ fn stage1_r_sweep_over(from: Timestamp, to: Timestamp, data: &DataSource, grid: let (tx_ex, _) = mpsc::channel(); let (tx_r, _) = mpsc::channel(); let (tx_req, _) = mpsc::channel(); - let bp = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, true); + let bp = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, true, None); let space = bp.param_space(); let stop_length_axis = space.iter().map(|p| p.name.clone()) .find(|n| n.ends_with(STOP_LENGTH_SUFFIX)).expect("open stage1-r vol-stop exposes a stop_length axis"); @@ -1321,7 +1322,7 @@ fn stage1_r_sweep_over(from: Timestamp, to: Timestamp, data: &DataSource, grid: let (tx_ex, rx_ex) = mpsc::channel(); let (tx_r, rx_r) = mpsc::channel(); let (tx_req, _rx_req) = mpsc::channel(); - let mut h = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, true) + let mut h = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, true, None) .bootstrap_with_cells(point) .expect("stage1-r grid points are kind-checked against param_space"); let sources = data.windowed_sources(from, to); @@ -1337,7 +1338,7 @@ fn stage1_r_sweep_over(from: Timestamp, to: Timestamp, data: &DataSource, grid: .map(|(_, row)| (row[0].as_f64(), row[1].as_f64())).unwrap_or((0.0, 0.0)); let bias_sign_flips = rx_ex.try_iter().next().map(|(_, row)| row[2].as_i64() as u64).unwrap_or(0); let mut m = RunMetrics { total_pips, max_drawdown, bias_sign_flips, r: None }; - m.r = Some(summarize_r(&r_rows, 0.0)); + m.r = Some(summarize_r(&r_rows, &[])); RunReport { manifest, metrics: m } }) .map(|(fam, lat)| (fam, Some(lat))) @@ -1356,7 +1357,7 @@ fn run_oos_r( let (tx_ex, rx_ex) = mpsc::channel(); let (tx_r, rx_r) = mpsc::channel(); let (tx_req, rx_req) = mpsc::channel(); - let bp = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, false); + let bp = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, false, None); let space = bp.param_space(); let mut h = bp.bootstrap_with_cells(params) .expect("chosen params pre-validated by the in-sample GridSpace::new"); @@ -1373,12 +1374,12 @@ fn run_oos_r( let mut manifest = sim_optimal_manifest(named, window, 0, pip); manifest.broker = stage1_r_broker_label(pip); if let Some(name) = trace { - persist_traces_r(&format!("{name}/oos{}", from.0), &manifest, &eq_rows, &ex_rows, &req_rows); + persist_traces_r(&format!("{name}/oos{}", from.0), &manifest, &eq_rows, &ex_rows, &req_rows, &[]); } let equity = f64_field(&eq_rows, 0); let exposure = f64_field(&ex_rows, 0); let mut metrics = summarize(&equity, &exposure); - metrics.r = Some(summarize_r(&r_rows, 0.0)); + metrics.r = Some(summarize_r(&r_rows, &[])); (equity, RunReport { manifest, metrics }) } @@ -1434,7 +1435,7 @@ fn stage1_breakout_sweep_family(trace: Option<&str>, data: &DataSource, grid: &S let bias_sign_flips = rx_ex.try_iter().next().map(|(_, row)| row[2].as_i64() as u64).unwrap_or(0); let mut m = RunMetrics { total_pips, max_drawdown, bias_sign_flips, r: None }; - m.r = Some(summarize_r(&r_rows, 0.0)); + m.r = Some(summarize_r(&r_rows, &[])); m } else { let eq_rows: Vec<(Timestamp, Vec)> = rx_eq.try_iter().collect(); @@ -1442,10 +1443,10 @@ fn stage1_breakout_sweep_family(trace: Option<&str>, data: &DataSource, grid: &S let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); let req_rows: Vec<(Timestamp, Vec)> = rx_req.try_iter().collect(); if let Some(name) = trace { - persist_traces_r(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows, &req_rows); + persist_traces_r(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows, &req_rows, &[]); } let mut m = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0)); - m.r = Some(summarize_r(&r_rows, 0.0)); + m.r = Some(summarize_r(&r_rows, &[])); m }; points.push(SweepPoint { params: vec![], report: RunReport { manifest, metrics } }); @@ -1506,7 +1507,7 @@ fn stage1_meanrev_sweep_family(trace: Option<&str>, data: &DataSource, grid: &St let bias_sign_flips = rx_ex.try_iter().next().map(|(_, row)| row[2].as_i64() as u64).unwrap_or(0); let mut m = RunMetrics { total_pips, max_drawdown, bias_sign_flips, r: None }; - m.r = Some(summarize_r(&r_rows, 0.0)); + m.r = Some(summarize_r(&r_rows, &[])); m } else { let eq_rows: Vec<(Timestamp, Vec)> = rx_eq.try_iter().collect(); @@ -1514,10 +1515,10 @@ fn stage1_meanrev_sweep_family(trace: Option<&str>, data: &DataSource, grid: &St let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); let req_rows: Vec<(Timestamp, Vec)> = rx_req.try_iter().collect(); if let Some(name) = trace { - persist_traces_r(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows, &req_rows); + persist_traces_r(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows, &req_rows, &[]); } let mut m = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0)); - m.r = Some(summarize_r(&r_rows, 0.0)); + m.r = Some(summarize_r(&r_rows, &[])); m }; points.push(SweepPoint { params: vec![], report: RunReport { manifest, metrics } }); @@ -2620,6 +2621,7 @@ fn stage1_r_graph( slow_len: Option, stop_open: bool, reduce: bool, + cost: Option<(f64, mpsc::Sender<(Timestamp, Vec)>, mpsc::Sender<(Timestamp, Vec)>)>, ) -> Composite { let mut g = GraphBuilder::new("stage1_r"); // SMA-cross signal → Bias (the same signal the pip sample uses). @@ -2693,6 +2695,41 @@ fn stage1_r_graph( g.connect(exec.output("cum_realized_r"), r_equity.input("term[0]")); g.connect(exec.output("unrealized_r"), r_equity.input("term[1]")); g.connect(r_equity.output("value"), req.input("col[0]")); + if let Some((cost_per_trade, tx_net, tx_cost)) = cost { + let cost_node = g.add( + ConstantCost::builder().bind("cost_per_trade", Scalar::f64(cost_per_trade)), + ); + g.connect(exec.output("closed_this_cycle"), cost_node.input("closed")); + g.connect(exec.output("open"), cost_node.input("open")); + g.connect(exec.output("entry_price"), cost_node.input("entry_price")); + g.connect(exec.output("stop_price"), cost_node.input("stop_price")); + // net_r_equity = cum_realized_r + unrealized_r - cum_cost_in_r - open_cost_in_r + 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 `summarize_r` folds: the `ConstantCost` node's full + // [cost_in_r, cum_cost_in_r, open_cost_in_r] schema. `summarize_r`'s `cost_col` + // reads col 0 (per-close) and col 2 (window-end open) — the lockstep contract + // with aura-std's ConstantCost output — so the stream is recorded 3-wide. + let cost_rec = g.add(Recorder::builder( + vec![ScalarKind::F64, ScalarKind::F64, ScalarKind::F64], + Firing::Any, + tx_cost, + )); + g.connect(cost_node.output("cost_in_r"), cost_rec.input("col[0]")); + g.connect(cost_node.output("cum_cost_in_r"), cost_rec.input("col[1]")); + g.connect(cost_node.output("open_cost_in_r"), cost_rec.input("col[2]")); + } } g.build().expect("stage1_r wiring resolves") } @@ -2912,9 +2949,11 @@ fn stage1_meanrev_graph( /// `aura run --harness stage1-r [--real [--from][--to]] [--trace ]`: build the /// dual-tap stage1-r harness, run it on synthetic or real M1 data, fold the pip taps via /// `summarize` and the dense R-record via `summarize_r`, and attach the R block as -/// `RunMetrics.r = Some(..)`. Pure/deterministic (C1). round_trip_cost is 0.0 (Stage-1 is -/// frictionless; a --cost knob is a follow-on). -fn run_stage1_r(data: RunData, trace: Option<&str>) -> RunReport { +/// `RunMetrics.r = Some(..)`. Pure/deterministic (C1). `cost` is the optional flat +/// round-trip cost per trade (price units): `Some(c)` wires the `ConstantCost` node and +/// the `net_r_equity` tap and folds the cost stream into `net_expectancy_r`; `None` is the +/// frictionless gross-R baseline (net == gross, no extra tap, byte-unchanged). +fn run_stage1_r(data: RunData, trace: Option<&str>, cost: Option) -> RunReport { if let Some(n) = trace && let Err(e) = TraceStore::open("runs").ensure_name_free(n, WriteKind::Run) { @@ -2925,7 +2964,10 @@ fn run_stage1_r(data: RunData, trace: Option<&str>) -> RunReport { let (tx_ex, rx_ex) = mpsc::channel(); let (tx_r, rx_r) = mpsc::channel(); let (tx_req, rx_req) = mpsc::channel(); - let flat = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, Some(2), Some(4), false, false) + 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"); let mut h = Harness::bootstrap(flat).expect("valid stage1-r harness"); @@ -2948,6 +2990,11 @@ fn run_stage1_r(data: RunData, trace: Option<&str>) -> RunReport { let ex_rows: Vec<(Timestamp, Vec)> = rx_ex.try_iter().collect(); let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); let req_rows: Vec<(Timestamp, Vec)> = rx_req.try_iter().collect(); + // The cost taps drain empty when `cost` is `None` (no cost node wired): an empty + // `net_rows` suppresses the `net_r_equity` trace and an empty `cost_rows` folds to + // net == gross — the byte-unchanged frictionless baseline. + let net_rows: Vec<(Timestamp, Vec)> = rx_net.try_iter().collect(); + let cost_rows: Vec<(Timestamp, Vec)> = rx_cost.try_iter().collect(); let mut manifest = sim_optimal_manifest( vec![ @@ -2963,28 +3010,35 @@ fn run_stage1_r(data: RunData, trace: Option<&str>) -> RunReport { ); manifest.broker = stage1_r_broker_label(pip_size); if let Some(name) = trace { - persist_traces_r(name, &manifest, &eq_rows, &ex_rows, &req_rows); + 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, 0.0)); + metrics.r = Some(summarize_r(&r_rows, &cost_rows)); RunReport { manifest, metrics } } -/// Persist a stage1-r run's three taps: equity (off the SimBroker), exposure (off the -/// Bias), and r_equity = cum_realized_r + unrealized_r (off the RiskExecutor). Separate -/// from the two-tap `persist_traces` so the pip handlers stay byte-unchanged on disk. +/// Persist a stage1-r run's taps: equity (off the SimBroker), exposure (off the Bias), +/// r_equity = cum_realized_r + unrealized_r (off the RiskExecutor), and — only on a cost +/// run — net_r_equity (gross r_equity minus the cost-in-R taps). Separate from the two-tap +/// `persist_traces` so the pip handlers stay byte-unchanged on disk; the `net_r_equity` tap +/// is emitted only when `net_rows` is non-empty, so a no-cost run's on-disk trace set is +/// byte-unchanged too. 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 taps = 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); @@ -3000,21 +3054,23 @@ enum HarnessKind { Stage1R, } -/// The parsed `aura run` invocation: a harness, a data source, and an optional trace name. +/// The parsed `aura run` invocation: a harness, a data source, an optional trace name, and +/// an optional flat round-trip cost per trade (`--cost-per-trade`, stage1-r only this cycle). struct RunArgs { harness: HarnessKind, data: RunData, trace: Option, + cost: Option, } /// Parse the `run` tail: `[--harness ] [--real [--from ] [--to ]] -/// [--trace ]` in any order, each flag at most once. `--harness sma` is the default -/// (`--harness macd` / `--harness stage1-r` select the others); `--from`/`--to` are legal -/// only with `--real`. Pure (no I/O, no exit) so the grammar is unit-testable; `main` does -/// the side effects. +/// [--trace ] [--cost-per-trade ]` in any order, each flag at most once. +/// `--harness sma` is the default (`--harness macd` / `--harness stage1-r` select the +/// others); `--from`/`--to` are legal only with `--real`. Pure (no I/O, no exit) so the +/// grammar is unit-testable; `main` does the side effects. fn parse_run_args(rest: &[&str]) -> Result { let usage = || { - "usage: aura run [--harness ] [--real [--from ] [--to ]] [--trace ]" + "usage: aura run [--harness ] [--real [--from ] [--to ]] [--trace ] [--cost-per-trade ]" .to_string() }; let mut harness: Option = None; @@ -3022,6 +3078,7 @@ fn parse_run_args(rest: &[&str]) -> Result { let mut from: Option = None; let mut to: Option = None; let mut trace: Option = None; + let mut cost: Option = None; let mut tail = rest; while let Some((flag, t)) = tail.split_first() { match *flag { @@ -3058,6 +3115,11 @@ fn parse_run_args(rest: &[&str]) -> Result { trace = Some((*value).to_string()); tail = t; } + "--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; + } _ => return Err(usage()), } } @@ -3070,7 +3132,7 @@ fn parse_run_args(rest: &[&str]) -> Result { Some(s) => RunData::Real { symbol: s, from, to }, None => RunData::Synthetic, }; - Ok(RunArgs { harness, data, trace }) + Ok(RunArgs { harness, data, trace, cost }) } /// Route a parsed `run` invocation to its harness handler. The compile-time `match` IS @@ -3080,7 +3142,7 @@ fn run_dispatch(args: RunArgs) -> Result { Ok(match (args.harness, args.data) { (HarnessKind::Sma, RunData::Synthetic) => run_sample(trace), (HarnessKind::Macd, RunData::Synthetic) => run_macd(trace), - (HarnessKind::Stage1R, data) => run_stage1_r(data, trace), + (HarnessKind::Stage1R, data) => run_stage1_r(data, trace, args.cost), (HarnessKind::Sma, RunData::Real { symbol, from, to }) => { run_sample_real(&symbol, from, to, trace) } @@ -3091,7 +3153,7 @@ fn run_dispatch(args: RunArgs) -> Result { } const USAGE: &str = - "usage: aura run [--harness ] [--real [--from ] [--to ]] [--trace ] | aura chart [--tap ] [--panels] | aura graph | aura sweep [--strategy ] [--real [--from ] [--to ]] [--name |--trace ] | aura mc [--name |--trace ] | aura mc --strategy stage1-r [--real [--from ] [--to ]] [--fast ] [--slow ] [--stop-length ] [--stop-k ] [--block-len ] [--resamples ] [--seed ] | aura walkforward [--strategy ] [--real [--from ] [--to ]] [--name |--trace ] [--fast ] [--slow ] [--stop-length ] [--stop-k ] [--select ] | aura generalize [--strategy stage1-r] --real --fast --slow --stop-length --stop-k [--from ] [--to ] [--metric ] [--name ] | aura runs families | aura runs family [rank ]"; + "usage: aura run [--harness ] [--real [--from ] [--to ]] [--trace ] [--cost-per-trade ] | aura chart [--tap ] [--panels] | aura graph | aura sweep [--strategy ] [--real [--from ] [--to ]] [--name |--trace ] | aura mc [--name |--trace ] | aura mc --strategy stage1-r [--real [--from ] [--to ]] [--fast ] [--slow ] [--stop-length ] [--stop-k ] [--block-len ] [--resamples ] [--seed ] | aura walkforward [--strategy ] [--real [--from ] [--to ]] [--name |--trace ] [--fast ] [--slow ] [--stop-length ] [--stop-k ] [--select ] | aura generalize [--strategy stage1-r] --real --fast --slow --stop-length --stop-k [--from ] [--to ] [--metric ] [--name ] | aura runs families | aura runs family [rank ]"; fn main() { // Restore the default SIGPIPE disposition. Rust's runtime sets SIGPIPE to SIG_IGN @@ -4308,7 +4370,7 @@ mod tests { fn run_stage1_r_synthetic_folds_an_r_block() { // the stage1-r harness scores the SMA-cross signal in R: one shell-callable run // yields a RunReport whose metrics.r is Some with a finite SQN and >= 1 trade. - let report = run_stage1_r(RunData::Synthetic, None); + let report = run_stage1_r(RunData::Synthetic, None, None); let r = report.metrics.r.as_ref().expect("stage1-r run must populate metrics.r"); assert!(r.n_trades >= 1, "expected >= 1 trade, got {}", r.n_trades); assert!(r.sqn.is_finite(), "SQN must be finite, got {}", r.sqn); diff --git a/crates/aura-cli/tests/cli_run.rs b/crates/aura-cli/tests/cli_run.rs index 2ac1bd2..2e8bf7a 100644 --- a/crates/aura-cli/tests/cli_run.rs +++ b/crates/aura-cli/tests/cli_run.rs @@ -1620,6 +1620,32 @@ fn stage1_r_trace_persists_r_equity_and_charts_it() { let _ = std::fs::remove_dir_all(&dir); } +/// Property (Task 3, the run-path cost seam): `aura run --harness stage1-r +/// --cost-per-trade ` charges a flat round-trip cost in R end to end — the run +/// persists a fourth `net_r_equity` tap beside r_equity and the report's +/// `net_expectancy_r` is STRICTLY below the gross `expectancy_r`. Pins the whole +/// flag→ConstantCost node→net_r_equity tap→summarize_r net fold loop (a no-cost run +/// leaves net == gross, the held golden); a regression that dropped the cost node or +/// left the net fold reading an empty stream would collapse net back onto gross here. +#[test] +fn stage1_r_cost_run_persists_net_r_equity_and_charges_cost() { + let dir = temp_cwd("stage1-r-cost"); + let run = Command::new(BIN) + .current_dir(&dir) + .args(["run", "--harness", "stage1-r", "--cost-per-trade", "2", "--trace", "c1"]) + .output() + .unwrap(); + assert!(run.status.success(), "exit: {:?}; stderr: {}", run.status, + String::from_utf8_lossy(&run.stderr)); + assert!(dir.join("runs/traces/c1/net_r_equity.json").exists(), "net_r_equity trace persisted"); + let s = String::from_utf8(run.stdout).expect("utf-8 stdout"); + let v: serde_json::Value = serde_json::from_str(s.trim()).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}"); + let _ = std::fs::remove_dir_all(&dir); +} + /// Property (the spec's dual-yardstick headline): `aura run --harness stage1-r` scores /// ONE signal in BOTH yardsticks at once — the single emitted `RunReport` carries the pip /// `metrics.total_pips` (off the SimBroker branch) AND the nested `r` block (off the diff --git a/crates/aura-composites/tests/risk_executor.rs b/crates/aura-composites/tests/risk_executor.rs index d7538ff..0425c22 100644 --- a/crates/aura-composites/tests/risk_executor.rs +++ b/crates/aura-composites/tests/risk_executor.rs @@ -94,7 +94,7 @@ fn run_executor(prices: &[f64], stop_distance: f64, risk_budget: f64) -> Vec<(Ti #[test] fn risk_executor_bootstraps_and_folds_to_expected_rmetric() { let ledger = run_executor(&[100.0, 102.0, 105.0], 10.0, 1.0); - let m = summarize_r(&ledger, 0.0); + let m = summarize_r(&ledger, &[]); assert_eq!(m.n_open_at_end, 1, "the open position must be counted"); assert_eq!(m.n_trades, 1); assert!((m.expectancy_r - 0.5).abs() < 1e-9, "window-end R = (105-100)/10; got {}", m.expectancy_r); @@ -138,7 +138,7 @@ fn risk_executor_r_invariant_under_risk_budget() { #[test] fn folded_rmetrics_survives_runmetrics_serde_round_trip() { let ledger = run_executor(&[100.0, 104.0, 108.0, 110.0, 102.0, 96.0, 94.0], 5.0, 1.0); - let folded = summarize_r(&ledger, 0.0); + let folded = summarize_r(&ledger, &[]); assert!(folded.n_trades >= 1, "the run must produce at least one trade to fold"); let m = RunMetrics { total_pips: 0.0, max_drawdown: 0.0, bias_sign_flips: 0, r: Some(folded) }; let json = serde_json::to_string(&m).expect("serialize a run's RunMetrics with an r block"); @@ -175,7 +175,7 @@ fn risk_executor_vol_stop_arm_bootstraps_and_folds() { path.iter().enumerate().map(|(i, &p)| (Timestamp(i as i64), Scalar::f64(p))).collect(); h.run(vec![Box::new(VecSource::new(stream))]); let ledger: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); - let m = summarize_r(&ledger, 0.0); + let m = summarize_r(&ledger, &[]); assert!(m.n_trades >= 1, "the vol-stop executor must fold at least one trade"); assert!(m.sqn.is_finite(), "SQN must be finite, got {}", m.sqn); } @@ -207,7 +207,7 @@ fn fold_executor(exec: aura_engine::Composite, params: Vec) -> aura_engi path.iter().enumerate().map(|(i, &p)| (Timestamp(i as i64), Scalar::f64(p))).collect(); h.run(vec![Box::new(VecSource::new(stream))]); let ledger: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); - summarize_r(&ledger, 0.0) + summarize_r(&ledger, &[]) } /// Property (#137): `risk_executor_vol_open` exposes the vol-stop's EWMA length and diff --git a/crates/aura-engine/tests/stage1_r_e2e.rs b/crates/aura-engine/tests/stage1_r_e2e.rs index 6323fd2..fc0eabe 100644 --- a/crates/aura-engine/tests/stage1_r_e2e.rs +++ b/crates/aura-engine/tests/stage1_r_e2e.rs @@ -27,7 +27,7 @@ use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp}; use aura_engine::{PositionAction, RMetrics, derive_position_events, summarize_r}; -use aura_std::{FixedStop, PM_FIELD_NAMES, PM_RECORD_KINDS, PM_WIDTH, PositionManagement}; +use aura_std::{ConstantCost, FixedStop, PM_FIELD_NAMES, PM_RECORD_KINDS, PM_WIDTH, PositionManagement}; // The indices `report.rs::summarize_r` reads the dense record by. Re-declared here // (the consumer's `r_col` module is private to `aura-engine`) so the guard test can @@ -41,6 +41,11 @@ const STOP_PRICE: usize = 7; const CONVICTION_AT_ENTRY: usize = 9; const SIZE: usize = 10; const DIRECTION: usize = 4; +const CUM_REALIZED_R: usize = 13; +// `ConstantCost`'s 3-wide output schema [cost_in_r, cum_cost_in_r, open_cost_in_r] — the +// run-path's in-graph `net_r_equity` tap reads cum (1) + open (2); summarize_r reads 0/2. +const CUM_COST_IN_R: usize = 1; +const OPEN_COST_IN_R: usize = 2; /// Property: the real producer->consumer seam composes. Driving `FixedStop` -> /// `PositionManagement` directly (node-by-node, not a bootstrapped graph) over a @@ -77,7 +82,7 @@ fn synthetic_long_then_stop_produces_a_sane_rmetric() { )); } } - let m = summarize_r(&ledger, 0.0); + let m = summarize_r(&ledger, &[]); assert!(m.n_trades >= 1, "expected at least one trade (entry then stop), got {m:?}"); assert!(m.expectancy_r.is_finite()); // entered ~100, stopped at the 95 level (price gapped to 94 through it) -> a loss. @@ -118,7 +123,26 @@ fn run_chain_ledger(stop: &mut dyn Node, path: &[(f64, f64)]) -> Vec<(Timestamp, /// over a `(bias, price)` path. This is the actual producer->consumer seam, /// node-by-node; the resulting `RMetrics` is what a recorded Stage-1 run would report. fn run_chain(stop: &mut dyn Node, path: &[(f64, f64)]) -> RMetrics { - summarize_r(&run_chain_ledger(stop, path), 0.0) + summarize_r(&run_chain_ledger(stop, path), &[]) +} + +/// The co-temporal cost stream a `ConstantCost(c)` node would emit over `ledger`: per row, +/// the round-trip cost `c` (price units) expressed in R via that row's latched distance +/// (`|entry_price - stop_price|`), placed in BOTH the closed-cost (col 0) and the open-cost +/// (col 2) slots of the producer's 3-wide `[cost_in_r, cum_cost_in_r, open_cost_in_r]` row. +/// A zero-latched (flat) row contributes no cost. `summarize_r` reads col 0 for a closed +/// trade and col 2 for the window-end open trade, so this reproduces the retired scalar +/// `round_trip_cost` argument exactly — the cost recovered from the producer's own +/// `entry_price`/`stop_price` columns, end-to-end. Col 1 (cum) is unread by `summarize_r`. +fn const_cost_stream(ledger: &[(Timestamp, Vec)], c: f64) -> Vec<(Timestamp, Vec)> { + ledger + .iter() + .map(|(ts, row)| { + let latched = (row[ENTRY_PRICE].as_f64() - row[STOP_PRICE].as_f64()).abs(); + let cir = if latched > 0.0 { c / latched } else { 0.0 }; + (*ts, vec![Scalar::f64(cir), Scalar::f64(0.0), Scalar::f64(cir)]) + }) + .collect() } /// A constant-long-bias path (`bias = +1` every cycle) over `prices`. @@ -204,8 +228,8 @@ fn clean_stop_folds_to_exactly_minus_one_r() { fn net_of_cost_charges_one_round_trip_per_trade_through_the_recovered_latched_dist() { let mut stop = FixedStop::new(10.0); let ledger = run_chain_ledger(&mut stop, &long_path(&[100.0, 105.0, 110.0])); - let gross = summarize_r(&ledger, 0.0); - let net = summarize_r(&ledger, 2.0); + let gross = summarize_r(&ledger, &[]); + let net = summarize_r(&ledger, &const_cost_stream(&ledger, 2.0)); assert_eq!(gross.n_trades, 1); assert!((gross.expectancy_r - 1.0).abs() < 1e-9, "gross window-end R = (110-100)/10; got {}", gross.expectancy_r); // gross is untouched by the cost; only net_expectancy_r absorbs it. @@ -229,10 +253,10 @@ fn net_of_cost_charges_one_round_trip_per_trade_through_the_recovered_latched_di fn net_of_cost_is_charged_per_r_so_a_wider_stop_dilutes_it() { let mut tight = FixedStop::new(5.0); let tight_l = run_chain_ledger(&mut tight, &long_path(&[100.0, 102.0, 105.0])); - let (tg, tn) = (summarize_r(&tight_l, 0.0), summarize_r(&tight_l, 2.0)); + let (tg, tn) = (summarize_r(&tight_l, &[]), summarize_r(&tight_l, &const_cost_stream(&tight_l, 2.0))); let mut wide = FixedStop::new(20.0); let wide_l = run_chain_ledger(&mut wide, &long_path(&[100.0, 102.0, 105.0])); - let (wg, wn) = (summarize_r(&wide_l, 0.0), summarize_r(&wide_l, 2.0)); + let (wg, wn) = (summarize_r(&wide_l, &[]), summarize_r(&wide_l, &const_cost_stream(&wide_l, 2.0))); let tight_cost = tg.expectancy_r - tn.net_expectancy_r; // 2.0/5 = 0.4 let wide_cost = wg.expectancy_r - wn.net_expectancy_r; // 2.0/20 = 0.1 assert!((tight_cost - 0.4).abs() < 1e-9, "tight cost = 2/5; got {tight_cost}"); @@ -240,6 +264,75 @@ fn net_of_cost_is_charged_per_r_so_a_wider_stop_dilutes_it() { assert!(tight_cost > wide_cost, "a tighter stop must pay more R per fixed price-unit cost"); } +/// Drive the REAL `aura_std::ConstantCost(c)` node over a recorded PM `ledger`, +/// node-by-node off the producer's own `closed`/`open`/`entry_price`/`stop_price` +/// columns — the actual cost producer the run-path graph taps (not an algebraic stand-in +/// like `const_cost_stream`). Returns the node's 3-wide `[cost_in_r, cum_cost_in_r, +/// open_cost_in_r]` rows, co-temporal 1:1 with `ledger`. Fresh single-element columns per +/// cycle (the node's lookback is 1), matching the node's own unit-test driving pattern. +fn const_cost_node_stream( + ledger: &[(Timestamp, Vec)], + c: f64, +) -> Vec<(Timestamp, Vec)> { + let mut node = ConstantCost::new(c); + 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 + ]; + 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(); + 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() +} + +/// Property (Task 3, the run-path net seam): **the in-graph `net_r_equity` tap and the +/// post-run `summarize_r` net fold agree on the same recorded streams.** The run-path +/// graph taps a per-cycle `net_r_equity = cum_realized_r + unrealized_r − cum_cost_in_r − +/// open_cost_in_r` (a `LinComb` over the executor + `ConstantCost` outputs), while +/// `summarize_r` independently folds the same PM records + cost stream into +/// `net_expectancy_r`. The FINAL `net_r_equity` sample must equal the post-run net total +/// (`net_expectancy_r × n_trades` = Σ over trades of `r − cost`) — a divergence would mean +/// the charted net-R-equity series and the reported net metric disagree. The path carries +/// one clean stopped loser AND a window-end open trade, so both the cumulative-close-cost +/// (`cum_cost_in_r`) and the window-end open-cost (`open_cost_in_r`) terms bite. The cost is +/// produced by the real `ConstantCost` node, the same producer the run-path wires. +#[test] +fn net_r_equity_final_sample_agrees_with_summarize_r_net_total() { + let mut stop = FixedStop::new(10.0); + // open long @100, stop out at the 90 level (bias->0): a clean -1R closed loser; then a + // fresh long @100 held up to 105, open at window end (+0.5R). Closed-cost charges + // cum_cost_in_r; the open trade charges open_cost_in_r — both net terms exercised. + 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 cost = const_cost_node_stream(&ledger, 2.0); + let m = summarize_r(&ledger, &cost); + assert_eq!(m.n_trades, 2, "one closed loser + one window-end open trade"); + assert_eq!(m.n_open_at_end, 1, "the second long is still open at window end"); + // post-run net total = mean net R × trade count = Σ (r − cost) over all trades. + let post_run_net_total = m.net_expectancy_r * m.n_trades as f64; + // in-graph final net_r_equity sample = the LinComb(1,1,−1,−1) at the last cycle. + let (_, last_pm) = ledger.last().unwrap(); + let (_, last_cost) = cost.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 net_r_equity {net_eq_final} must equal post-run net total {post_run_net_total}", + ); +} + /// Property (#130): **`sqn` and `sqn_normalized` (SQN100) are computed from the /// producer's *recorded* dense records, not hand-built rows, and below the /// 100-trade cap the two are identical.** Every `summarize_r` test that touches @@ -400,7 +493,7 @@ fn derived_event_table_agrees_with_r_metrics_over_one_ledger() { let mut stop = FixedStop::new(5.0); let ledger = run_chain_ledger(&mut stop, &long_path(&[100.0, 104.0, 108.0, 110.0, 102.0, 96.0, 94.0])); - let m: RMetrics = summarize_r(&ledger, 0.0); + let m: RMetrics = summarize_r(&ledger, &[]); let events = derive_position_events(&ledger, 7); let closes = events.iter().filter(|e| e.action == PositionAction::Close).count() as u64; diff --git a/crates/aura-engine/tests/streaming_reduction_equivalence.rs b/crates/aura-engine/tests/streaming_reduction_equivalence.rs index e8d4a04..47657fa 100644 --- a/crates/aura-engine/tests/streaming_reduction_equivalence.rs +++ b/crates/aura-engine/tests/streaming_reduction_equivalence.rs @@ -60,7 +60,7 @@ fn gated_recorder_then_summarize_r_equals_raw_summarize_r() { .collect(); // RAW: summarize_r over the whole dense record. - let raw = summarize_r(&full, 0.0); + let raw = summarize_r(&full, &[]); // FOLDED: drive GatedRecorder (gate = CLOSED), collect its emitted rows, fold. let (tx, rx) = mpsc::channel(); @@ -75,7 +75,7 @@ fn gated_recorder_then_summarize_r_equals_raw_summarize_r() { } g.finalize(); let folded_rows: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); - let folded = summarize_r(&folded_rows, 0.0); + let folded = summarize_r(&folded_rows, &[]); assert_eq!(folded, raw, "GatedRecorder+summarize_r must equal raw summarize_r"); } diff --git a/crates/aura-std/src/constant_cost.rs b/crates/aura-std/src/constant_cost.rs new file mode 100644 index 0000000..fea7ff3 --- /dev/null +++ b/crates/aura-std/src/constant_cost.rs @@ -0,0 +1,189 @@ +//! `ConstantCost` — a flat round-trip cost charged once per closed trade, in R. +//! The first cost node of the C10 cost-model graph: 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 { + assert!(cost_per_trade >= 0.0, "ConstantCost cost_per_trade must be >= 0"); + 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 full geometry record is present this cycle + // (all four fields fire together; before warm-up none of them do). + 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); + if closed_w.is_empty() || open_w.is_empty() || entry_w.is_empty() || stop_w.is_empty() { + return None; + } + let closed = closed_w[0]; + let open = open_w[0]; + let entry = entry_w[0]; + let stop = stop_w[0]; + let latched = (entry - stop).abs(); + // A zero latched distance is a flat/degenerate cycle (no valid 1R denominator): + // it contributes no cost rather than dividing by zero — matching summarize_r's guard. + 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 { + format!("ConstantCost({})", self.cost_per_trade) + } +} + +#[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 label_carries_the_cost() { + // Parameterised sibling convention: the identifying param is in the label, + // so two differently-priced cost nodes disambiguate in a trace. + assert_eq!(ConstantCost::new(2.0).label(), "ConstantCost(2)"); + assert_eq!(ConstantCost::new(0.5).label(), "ConstantCost(0.5)"); + } + + #[test] + #[should_panic(expected = "cost_per_trade must be >= 0")] + fn new_panics_on_negative_cost() { + let _ = ConstantCost::new(-1.0); + } + + #[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()) + ); + } +} diff --git a/crates/aura-std/src/lib.rs b/crates/aura-std/src/lib.rs index 5985119..800742d 100644 --- a/crates/aura-std/src/lib.rs +++ b/crates/aura-std/src/lib.rs @@ -18,6 +18,7 @@ mod add; mod and; mod bias; +mod constant_cost; mod delay; mod ema; mod eqconst; @@ -43,6 +44,7 @@ mod sub; pub use add::Add; pub use and::And; pub use bias::Bias; +pub use constant_cost::ConstantCost; pub use delay::Delay; pub use ema::Ema; pub use eqconst::EqConst;