diff --git a/crates/aura-cli/src/campaign_run.rs b/crates/aura-cli/src/campaign_run.rs index 0f26ca2..996d90c 100644 --- a/crates/aura-cli/src/campaign_run.rs +++ b/crates/aura-cli/src/campaign_run.rs @@ -862,7 +862,7 @@ pub(crate) fn persist_campaign_traces( let (tx_ex, rx_ex) = mpsc::channel(); let (tx_r, rx_r) = mpsc::channel(); let (tx_req, rx_req) = mpsc::channel(); - let mut h = crate::wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, false, geo.pip_size, &binding) + let mut h = crate::wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, false, geo.pip_size, &binding, None) .bootstrap_with_cells(&point) .expect("the member's point re-bootstraps (it already ran this realization)"); h.run(sources); diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index a068dc4..7506a79 100644 --- a/crates/aura-cli/src/main.rs +++ b/crates/aura-cli/src/main.rs @@ -13,8 +13,8 @@ mod scaffold; mod verb_sugar; use render::{ChartData, ChartMeta, ChartMode, ReduceKind, Series}; -use aura_core::{zip_params, Cell, Firing, ParamSpec, Scalar, ScalarKind, Timestamp}; -use aura_composites::{risk_executor, StopRule}; +use aura_core::{zip_params, Cell, Firing, ParamSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp}; +use aura_composites::{cost_graph, risk_executor, StopRule}; use aura_engine::{ blueprint_from_json, blueprint_to_json, f64_field, join_on_ts, monte_carlo, param_stability, r_metrics_from_rs, summarize, summarize_r, walk_forward, window_of, BindError, BlueprintNode, @@ -30,8 +30,8 @@ use aura_registry::{ DEFLATION_BLOCK_LEN, DEFLATION_N_RESAMPLES, }; use aura_std::{ - GatedRecorder, LinComb, Recorder, - SeriesReducer, SimBroker, PM_FIELD_NAMES, + cost_port, GatedRecorder, LinComb, Recorder, RollingMax, RollingMin, + SeriesReducer, SimBroker, Sub, GEOMETRY_WIDTH, PM_FIELD_NAMES, PM_RECORD_KINDS, }; // `std_vocabulary` is now only reached through `project::Env::resolve` in production @@ -60,12 +60,8 @@ use aura_std::{Add, Gt, Latch, Mul, Sqrt}; use aura_engine::ColumnarTrace; #[cfg(test)] use aura_std::{Bias, Ema, Sma}; -// `RollingMax`/`RollingMin`/`Sub` are now only reached from the test-only -// `r_breakout_signal`/`r_meanrev_signal` carves: #221's `wrap_r` cost-graph -// removal dropped the last production caller (the vol-slippage proxy), mirroring -// `Delay`/`Scale` above. #[cfg(test)] -use aura_std::{RollingMax, RollingMin, Sub}; +use aura_std::{ConstantCost, VolSlippageCost}; use std::sync::mpsc; use std::sync::LazyLock; use std::collections::{BTreeMap, HashSet}; @@ -1130,7 +1126,7 @@ fn reproduce_family_in( eprintln!("aura: {}", binding::synthetic_refusal(&hash, &binding)); std::process::exit(1); } - let space = wrap_r(reload(), tx_eq, tx_ex, tx_r, tx_req, stop, true, SYNTHETIC_PIP_SIZE, &binding).param_space(); + let space = wrap_r(reload(), tx_eq, tx_ex, tx_r, tx_req, stop, true, SYNTHETIC_PIP_SIZE, &binding, None).param_space(); let point = point_from_params(&space, &stored.manifest.params); // A MonteCarlo member carries no tuning params (the params-join is empty), so its // reproduce line would print a BLANK member label; the seed IS its realization @@ -1241,6 +1237,24 @@ pub(crate) const R_SMA_STOP_LENGTH: i64 = 3; /// `StopRule::Vol` and its manifest record, like [`R_SMA_STOP_LENGTH`]. pub(crate) const R_SMA_STOP_K: f64 = 2.0; +/// Short-horizon realized-range window for vol-scaled slippage. Deliberately +/// distinct from `R_SMA_STOP_LENGTH` (3): scaling slippage by the stop's own +/// vol would collapse cost-in-R to a constant (spec 0082). Short enough to warm +/// within the synthetic smoke fixture so the run path exercises non-zero slippage. +const SLIP_VOL_LENGTH: i64 = 5; + +/// The optional cost leg of the R scaffolding (#234): the resolved cost-node +/// builders (from `campaign_run::cost_nodes_for`, fully bound — they add no +/// open param, so the wrapped `param_space` is cost-invariant) plus the two +/// recording channels — the aggregate 3-field cost record (`tx_cost`, the +/// `summarize_r` join input) and the `net_r_equity` curve (`tx_net`, recorded +/// only in `!reduce` trace mode). +struct CostLeg { + nodes: Vec, + tx_cost: mpsc::Sender<(Timestamp, Vec)>, + tx_net: mpsc::Sender<(Timestamp, Vec)>, +} + /// Which data a `run` drives a harness on: the built-in synthetic stream, or real M1 /// close bars for a vetted symbol over an optional window. #[derive(Debug)] @@ -1304,6 +1318,7 @@ fn wrap_r( reduce: bool, pip_size: f64, binding: &binding::ResolvedBinding, + cost: Option, ) -> Composite { let mut g = GraphBuilder::new("r_sma"); // The strategy signal → Bias, nested as a serializable roles→`bias` leg. @@ -1382,6 +1397,95 @@ fn wrap_r( g.connect(exec.output("unrealized_r"), r_equity.input("term[1]")); g.connect(r_equity.output("value"), req.input("col[0]")); } + // The optional cost leg (#234 — the #221-deleted wiring rebuilt as a + // campaign-documented feature): a `cost_graph` fed by the executor's four + // geometry outputs, its aggregate record recorded co-temporally with the R + // record (the `summarize_r` positional-join input), and — in `!reduce` + // trace mode — the LinComb(4) net-equity curve into `tx_net`. Absent leg + // (`None`) == today's graph, byte-identical. + if let Some(leg) = cost { + // Components taking a `volatility` extra input (the vol_slippage + // shape), discovered from each builder's own schema past the geometry + // prefix — no second vocabulary of "which node needs the proxy". + let vol_slots: Vec = leg + .nodes + .iter() + .enumerate() + .filter(|(_, n)| { + n.schema().inputs[GEOMETRY_WIDTH..].iter().any(|p| p.name == "volatility") + }) + .map(|(k, _)| k) + .collect(); + // The short-horizon realized-range vol proxy, fed from the close role, + // shared by every vol-scaled component (the #221-deleted arm, verbatim + // wiring; built in BOTH modes — the reduce-mode member run charges + // slippage too). + let vol_range = if vol_slots.is_empty() { + None + } else { + 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")); + g.feed(close, [vhi.input("series"), vlo.input("series")]); + Some(vrange) + }; + // One cost_graph composite fans the shared PM-geometry into the + // components and sums their per-field charges (C10). + let cg = g.add(cost_graph(leg.nodes)); + g.connect(exec.output("closed_this_cycle"), cg.input("closed")); + g.connect(exec.output("open"), cg.input("open")); + g.connect(exec.output("entry_price"), cg.input("entry_price")); + g.connect(exec.output("stop_price"), cg.input("stop_price")); + for k in vol_slots { + let vrange = vol_range.as_ref().expect("proxy is built whenever a vol slot exists"); + g.connect(vrange.output("value"), cg.input(cost_port(k, "volatility"))); + } + if reduce { + // The aggregate cost record, gated on the SAME closed flag as the R + // GatedRecorder and flushed once at finalize — so the cost rows stay + // positionally 1:1 with the gated R rows (`summarize_r`'s join). The + // gate rides as an APPENDED col 3: cols 0..2 keep the aura-analysis + // `cost_col` contract (cost_in_r = 0, open_cost_in_r = 2). + let crec = g.add(GatedRecorder::builder( + vec![ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::Bool], + 3, + Firing::Any, + leg.tx_cost, + )); + g.connect(cg.output("cost_in_r"), crec.input("col[0]")); + g.connect(cg.output("cum_cost_in_r"), crec.input("col[1]")); + g.connect(cg.output("open_cost_in_r"), crec.input("col[2]")); + g.connect(exec.output("closed_this_cycle"), crec.input("col[3]")); + } else { + // The full per-cycle aggregate cost record (col 0 per-close, col 2 + // window-end — what summarize_r folds on the trace path). + let crec = g.add(Recorder::builder( + vec![ScalarKind::F64, ScalarKind::F64, ScalarKind::F64], + Firing::Any, + leg.tx_cost, + )); + g.connect(cg.output("cost_in_r"), crec.input("col[0]")); + g.connect(cg.output("cum_cost_in_r"), crec.input("col[1]")); + g.connect(cg.output("open_cost_in_r"), crec.input("col[2]")); + // net_r_equity = cum_realized_r + unrealized_r − Σcum_cost_in_r + // − Σopen_cost_in_r (the #221-deleted LinComb(4), weights 1,1,-1,-1). + 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(cg.output("cum_cost_in_r"), net_eq.input("term[2]")); + g.connect(cg.output("open_cost_in_r"), net_eq.input("term[3]")); + let net_rec = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, leg.tx_net)); + g.connect(net_eq.output("value"), net_rec.input("col[0]")); + } + } g.build().expect("r_sma wiring resolves") } @@ -1445,7 +1549,7 @@ fn run_signal_r( // receiver alive so the sink's sends do not fail, but do not drain it. let (tx_req, _rx_req) = mpsc::channel(); let (sources, window, pip_size) = resolve_run_data(&data, env, &binding); - let wrapped = wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, false, pip_size, &binding); + let wrapped = wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, false, pip_size, &binding, None); let flat = wrapped .compile_with_params(params) .expect("signal binds + wraps to a valid harness"); @@ -1489,7 +1593,7 @@ fn run_blueprint_member( let (tx_ex, rx_ex) = mpsc::channel(); let (tx_r, rx_r) = mpsc::channel(); let (tx_req, _rx_req) = mpsc::channel(); - let mut h = wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, true, pip, binding) + let mut h = wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, true, pip, binding, None) .bootstrap_with_cells(point) .expect("member bootstraps (point kind-checked against param_space)"); h.run(sources); @@ -1540,7 +1644,7 @@ fn blueprint_axis_probe(doc: &str, env: &project::Env) -> Composite { let (tx_ex, _) = mpsc::channel(); let (tx_r, _) = mpsc::channel(); let (tx_req, _) = mpsc::channel(); - wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, true, SYNTHETIC_PIP_SIZE, &probe) + wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, true, SYNTHETIC_PIP_SIZE, &probe, None) } /// `aura sweep --list-axes`: one `:` line per open @@ -3582,6 +3686,7 @@ mod tests { false, SYNTHETIC_PIP_SIZE, &binding, + None, ) .compile_with_params(&[]) .expect("closed hl_channel wraps to a valid harness"); @@ -3671,6 +3776,7 @@ mod tests { false, SYNTHETIC_PIP_SIZE, &binding, + None, ) .compile_with_params(&[]) .expect("r-meanrev signal wraps to a valid harness"); @@ -3694,6 +3800,170 @@ mod tests { assert_eq!(*bias.last().unwrap(), 1.0, "the down-fade long must hold to the end: {bias:?}"); } + /// #234: the optional cost leg records an aggregate cost stream positionally + /// 1:1 with the R record (`summarize_r`'s positional-join contract) plus a + /// per-cycle net_r_equity curve, in non-reduce trace mode. The cost-less + /// side (`cost: None` == today's graph, byte-identical) is pinned by the + /// suite's untouched equivalence anchors. + #[test] + fn wrap_r_cost_leg_records_cost_rows_co_temporal_with_the_r_record() { + let signal = load_closed_r_sma(); + let binding = binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new()) + .expect("the price role resolves"); + 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_cost, rx_cost) = mpsc::channel(); + let (tx_net, rx_net) = mpsc::channel(); + let leg = CostLeg { + nodes: vec![ConstantCost::builder().bind("cost_per_trade", Scalar::f64(0.0005))], + tx_cost, + tx_net, + }; + let flat = wrap_r( + signal, + tx_eq, + tx_ex, + tx_r, + tx_req, + StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, + false, + SYNTHETIC_PIP_SIZE, + &binding, + Some(leg), + ) + .compile_with_params(&[]) + .expect("costed wrap builds"); + let mut h = Harness::bootstrap(flat).expect("costed harness bootstraps"); + let src: Vec> = vec![Box::new(VecSource::new(r_sma_prices()))]; + h.run(src); + let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); + let cost_rows: Vec<(Timestamp, Vec)> = rx_cost.try_iter().collect(); + let net_rows: Vec<(Timestamp, Vec)> = rx_net.try_iter().collect(); + assert!(!r_rows.is_empty(), "the fixture must warm the executor"); + assert_eq!( + cost_rows.len(), + r_rows.len(), + "the cost stream is positionally 1:1 with the R record (co-temporality)" + ); + assert_eq!(net_rows.len(), cost_rows.len(), "the net curve emits per cost row"); + assert!( + cost_rows.iter().any(|(_, row)| row[0].as_f64() != 0.0), + "at least one close must charge a non-zero cost (non-vacuous)" + ); + } + + /// #234: the reduce-mode cost branch (a `GatedRecorder` with the appended + /// `closed_this_cycle` gate column) stays co-temporal with the gated R + /// record too — the same 1:1 join contract as the trace-mode leg above, + /// but over the member/sweep run path `summarize_r` actually consumes. + #[test] + fn wrap_r_cost_leg_in_reduce_mode_stays_co_temporal_with_the_gated_r_record() { + let signal = load_closed_r_sma(); + let binding = binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new()) + .expect("the price role resolves"); + 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_cost, rx_cost) = mpsc::channel(); + let (tx_net, _rx_net) = mpsc::channel(); + let leg = CostLeg { + nodes: vec![ConstantCost::builder().bind("cost_per_trade", Scalar::f64(0.0005))], + tx_cost, + tx_net, + }; + let flat = wrap_r( + signal, + tx_eq, + tx_ex, + tx_r, + tx_req, + StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, + true, + SYNTHETIC_PIP_SIZE, + &binding, + Some(leg), + ) + .compile_with_params(&[]) + .expect("costed reduce-mode wrap builds"); + let mut h = Harness::bootstrap(flat).expect("costed reduce-mode harness bootstraps"); + let src: Vec> = vec![Box::new(VecSource::new(r_sma_prices()))]; + h.run(src); + let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); + let cost_rows: Vec<(Timestamp, Vec)> = rx_cost.try_iter().collect(); + assert!(!r_rows.is_empty(), "the fixture must warm and close at least one trade"); + assert_eq!( + cost_rows.len(), + r_rows.len(), + "the reduce-mode gated cost stream is positionally 1:1 with the gated R record" + ); + assert!( + cost_rows.iter().any(|(_, row)| row[0].as_f64() != 0.0), + "at least one gated close must charge a non-zero cost (non-vacuous)" + ); + } + + /// #234: a cost component that declares an extra `volatility` port (the + /// `VolSlippageCost` shape) is wired to a live realized-range proxy built + /// from the close role (`RollingMax`/`RollingMin`/`Sub` over + /// `SLIP_VOL_LENGTH`), discovered purely from the component's own schema + /// past the geometry prefix (`vol_slots`) — never a disconnected or + /// hardwired-zero input. If that wiring regresses, `ctx.f64_in(GEOMETRY_WIDTH)` + /// inside the component stays permanently empty and every charge is exactly + /// 0.0 (`vol_not_yet_warm_emits_zero_cost_co_temporally`'s withhold case), so a + /// non-zero charge over a genuinely moving price series is a direct witness + /// that the proxy reached the component. The two tests above use + /// `ConstantCost`, which never touches this wiring at all. + #[test] + fn wrap_r_cost_leg_wires_the_vol_slippage_proxy_from_the_close_role() { + let signal = load_closed_r_sma(); + let binding = binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new()) + .expect("the price role resolves"); + 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_cost, rx_cost) = mpsc::channel(); + let (tx_net, _rx_net) = mpsc::channel(); + let leg = CostLeg { + nodes: vec![VolSlippageCost::builder().bind("slip_vol_mult", Scalar::f64(0.5))], + tx_cost, + tx_net, + }; + let flat = wrap_r( + signal, + tx_eq, + tx_ex, + tx_r, + tx_req, + StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, + false, + SYNTHETIC_PIP_SIZE, + &binding, + Some(leg), + ) + .compile_with_params(&[]) + .expect("vol-slippage-costed wrap builds"); + let mut h = Harness::bootstrap(flat).expect("vol-slippage-costed harness bootstraps"); + let src: Vec> = vec![Box::new(VecSource::new(r_sma_prices()))]; + h.run(src); + let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); + let cost_rows: Vec<(Timestamp, Vec)> = rx_cost.try_iter().collect(); + assert!(!r_rows.is_empty(), "the fixture must warm and close at least one trade"); + assert_eq!( + cost_rows.len(), + r_rows.len(), + "the vol-slippage cost stream stays positionally 1:1 with the R record" + ); + assert!( + cost_rows.iter().any(|(_, row)| row[0].as_f64() != 0.0), + "the volatility proxy must actually feed the component — a disconnected \ + proxy leaves cost_in_r at exactly 0.0 on every close: {cost_rows:?}" + ); + } + /// Property: the pip a run resolves and stamps into its manifest must be the /// pip the in-graph `SimBroker` divides by — the resolved (real, per-instrument) /// pip has to reach the graph, not just the manifest label. The `SimBroker`