diff --git a/crates/aura-cli/src/campaign_run.rs b/crates/aura-cli/src/campaign_run.rs index 996d90c..90c007a 100644 --- a/crates/aura-cli/src/campaign_run.rs +++ b/crates/aura-cli/src/campaign_run.rs @@ -19,7 +19,7 @@ use std::path::{Path, PathBuf}; use std::sync::{mpsc, Arc}; use aura_campaign::{CampaignOutcome, CellOutcome, CellSpec, ExecFault, MemberFault, MemberRunner}; -use aura_core::{Cell, ParamSpec, Scalar, ScalarKind, Timestamp}; +use aura_core::{Cell, ParamSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp}; use aura_engine::{ blueprint_from_json, f64_field, summarize, summarize_r, ColumnarTrace, FamilySelection, RunReport, @@ -28,8 +28,9 @@ use aura_ingest::{instrument_geometry, open_columns_window, unix_ms_to_epoch_ns} use aura_registry::{CampaignRunRecord, WriteKind}; use aura_research::{ campaign_to_json, content_id_of, parse_campaign, parse_process, validate_campaign, - validate_process, CampaignDoc, DocRef, + validate_process, CampaignDoc, CostSpec, DocRef, }; +use aura_std::{CarryCost, ConstantCost, VolSlippageCost}; use crate::project::Env; use crate::research_docs::{ @@ -61,6 +62,40 @@ fn stop_rule_for_regime(regime: Option) -> aura_compo } } +/// The one CostSpec -> cost-node binding (#234), beside its risk sibling +/// `stop_rule_for_regime` and shared the same way: `CliMemberRunner::run_member` +/// (via `run_blueprint_member`) and `persist_campaign_traces`'s drift-alarm +/// re-run both bind through here, so the persist-side re-run structurally +/// cannot diverge from the run-side cost model (the #219 divergence class, +/// cost edition). The bound knob names are the builders' own `ParamSpec` names +/// — the `CostSpec` serde vocabulary conforms to them. +pub(crate) fn cost_nodes_for(specs: &[CostSpec]) -> Vec { + specs + .iter() + .map(|s| { + let (knob, v) = cost_knob(s); + match s { + CostSpec::Constant { .. } => ConstantCost::builder().bind(knob, Scalar::f64(v)), + CostSpec::VolSlippage { .. } => VolSlippageCost::builder().bind(knob, Scalar::f64(v)), + CostSpec::Carry { .. } => CarryCost::builder().bind(knob, Scalar::f64(v)), + } + }) + .collect() +} + +/// The one CostSpec-variant -> (knob name, value) mapping, shared by +/// `cost_nodes_for`'s bind above and `run_blueprint_member`'s manifest stamp +/// (main.rs): the stamp key must equal the bind key for reproduce to +/// re-derive a `CostSpec` from a stored manifest, so both sites read the +/// name off this single function rather than each carrying its own literal. +pub(crate) fn cost_knob(spec: &CostSpec) -> (&'static str, f64) { + match spec { + CostSpec::Constant { cost_per_trade } => ("cost_per_trade", *cost_per_trade), + CostSpec::VolSlippage { slip_vol_mult } => ("slip_vol_mult", *slip_vol_mult), + CostSpec::Carry { carry_per_cycle } => ("carry_per_cycle", *carry_per_cycle), + } +} + /// Phrase an [`ExecFault`] for stderr — Debug-leak-free, path-addressed /// (stage index + block id), the `doc_fault_prose`/`ref_fault_prose` register. pub(crate) fn exec_fault_prose(f: &ExecFault) -> String { @@ -232,6 +267,9 @@ struct CliMemberRunner<'a> { /// into per-member binding resolution (#231: rebind per campaign without /// touching the blueprint's content id). bindings: BTreeMap, + /// The campaign's cost model (#234), threaded into every member run via + /// `cost_nodes_for` (empty = zero costs, today's graph). + cost: Vec, } impl MemberRunner for CliMemberRunner<'_> { @@ -313,6 +351,7 @@ impl MemberRunner for CliMemberRunner<'_> { self.env, stop, &binding, + &self.cost, ); report.manifest.instrument = Some(cell.instrument.clone()); Ok(report) @@ -482,6 +521,7 @@ pub(crate) fn run_campaign_returning( env, server: Arc::new(data_server::DataServer::new(env.data_path())), bindings: campaign.data.bindings.clone(), + cost: campaign.cost.clone(), }; let outcome = aura_campaign::execute( campaign_id, @@ -862,16 +902,32 @@ 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, None) - .bootstrap_with_cells(&point) - .expect("the member's point re-bootstraps (it already ran this realization)"); + // The campaign's OWN cost model (#234), bound the same way + // `CliMemberRunner::run_member` binds it (via `cost_nodes_for`): + // empty = no leg, so a cost-less campaign's re-run is unchanged. + // A non-empty model MUST be threaded here too — the recorded + // member ran net (Task 4's `run_blueprint_member`); leaving this + // re-run gross would trip the C1 drift alarm below on every + // legitimate costed campaign, not just on a real divergence. + let (tx_cost, rx_cost) = mpsc::channel(); + let (tx_net, _rx_net) = mpsc::channel(); + let cost_leg = (!campaign.cost.is_empty()).then(|| crate::CostLeg { + nodes: cost_nodes_for(&campaign.cost), + tx_cost, + tx_net, + }); + let mut h = + crate::wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, false, geo.pip_size, &binding, cost_leg) + .bootstrap_with_cells(&point) + .expect("the member's point re-bootstraps (it already ran this realization)"); h.run(sources); - // Drain ALL FOUR channels (`run_signal_r` leaves req undrained; the - // trace path must not): eq/ex/req feed the taps, r feeds the metrics. + // Drain ALL FIVE channels (`run_signal_r` leaves req undrained; the + // trace path must not): eq/ex/req feed the taps, r+cost feed the metrics. let eq_rows: Vec<(Timestamp, Vec)> = rx_eq.try_iter().collect(); 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(); + let cost_rows: Vec<(Timestamp, Vec)> = rx_cost.try_iter().collect(); // The C1 drift alarm: metrics equality against the recorded // member. The reduce-mode fold shares its arithmetic with this @@ -879,7 +935,7 @@ pub(crate) fn persist_campaign_traces( // emits exactly the rows `summarize_r`'s ledger reads), so equality // is bit-exact. let mut rerun_metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0)); - rerun_metrics.r = Some(summarize_r(&r_rows, &[])); + rerun_metrics.r = Some(summarize_r(&r_rows, &cost_rows)); if rerun_metrics != member_report.metrics { return Err(format!( "trace re-run diverged from the recorded member (C1 violation): cell \ @@ -1225,4 +1281,24 @@ mod tests { assert_eq!(tap_channel("r_equity"), Some(TapChannel::REquity)); assert_eq!(tap_channel("net_r_equity"), None); } + + #[test] + /// #234: the one CostSpec -> builder binding maps each component to its + /// shipped cost node with the knob BOUND — a bound component adds no open + /// param, so the wrapped param_space stays cost-invariant (the probe/member + /// space equivalence the campaign path relies on). + fn cost_nodes_for_maps_each_component_to_its_bound_builder() { + let nodes = cost_nodes_for(&[ + aura_research::CostSpec::Constant { cost_per_trade: 2.0 }, + aura_research::CostSpec::VolSlippage { slip_vol_mult: 0.5 }, + aura_research::CostSpec::Carry { carry_per_cycle: 0.1 }, + ]); + let labels: Vec = nodes.iter().map(|n| n.label()).collect(); + assert_eq!(labels, ["ConstantCost", "VolSlippageCost", "CarryCost"]); + assert!( + nodes.iter().all(|n| n.params().is_empty()), + "every component must be fully bound (no open param leaks into param_space)" + ); + assert!(cost_nodes_for(&[]).is_empty(), "an empty model binds no nodes"); + } } diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index 7506a79..9b8264c 100644 --- a/crates/aura-cli/src/main.rs +++ b/crates/aura-cli/src/main.rs @@ -1053,6 +1053,39 @@ fn stop_rule_from_params(params: &[(String, Scalar)]) -> StopRule { } } +/// Re-derive the cost model a member was minted under from its manifest params +/// (`cost[k].`, stamped by `run_blueprint_member`) — the #233 stop-regime +/// pattern: like the stop, the cost model rides OUTSIDE the wrapped +/// param_space, so `point_from_params` cannot recover it. Components are read +/// in index order; the builder knob name discriminates the variant (each +/// shipped cost node has exactly one, distinctly named knob). No `cost[0].*` +/// param = the empty (gross) model — every pre-cost member widens to it, +/// mirroring `stop_rule_from_params`'s default arm. +fn cost_specs_from_params(params: &[(String, Scalar)]) -> Vec { + let mut specs = Vec::new(); + for k in 0.. { + let prefix = format!("cost[{k}]."); + let Some((knob, v)) = params + .iter() + .find_map(|(n, s)| n.strip_prefix(&prefix).map(|rest| (rest, *s))) + else { + break; + }; + specs.push(match knob { + "cost_per_trade" => aura_research::CostSpec::Constant { cost_per_trade: v.as_f64() }, + "slip_vol_mult" => aura_research::CostSpec::VolSlippage { slip_vol_mult: v.as_f64() }, + "carry_per_cycle" => aura_research::CostSpec::Carry { carry_per_cycle: v.as_f64() }, + other => { + // Corrupted-on-disk manifest data — the reproduce path's clean + // exit register (`aura:` + exit 1), like point_from_params. + eprintln!("aura: manifest cost param cost[{k}].{other} names no cost component"); + std::process::exit(1); + } + }); + } + specs +} + /// Look up a persisted family by id, or exit 1 (unknown id / registry load failure) — /// the single place `reproduce_family` and `reproduce_family_in` resolve a family, so /// the two exit-1 error phrasings can't drift out of sync between the call sites. @@ -1115,6 +1148,10 @@ fn reproduce_family_in( let (tx_r, _) = mpsc::channel(); let (tx_req, _) = mpsc::channel(); let stop = stop_rule_from_params(&stored.manifest.params); + // The member's cost model, re-derived from its manifest (the #233 + // stop pattern): a costed family re-runs under the exact components it + // was minted with, so `net_expectancy_r` reproduces bit-identically. + let cost = cost_specs_from_params(&stored.manifest.params); // The member's binding, re-derived from the stored blueprint's own // input roles (name defaults — family manifests carry no overrides). let binding = binding::resolve_binding(&hash, reload().input_roles(), &BTreeMap::new()) @@ -1188,6 +1225,7 @@ fn reproduce_family_in( env, stop, &binding, + &cost, ); outcomes.push((label, rerun.metrics == stored.metrics)); } @@ -1588,12 +1626,23 @@ fn run_blueprint_member( env: &project::Env, stop: StopRule, binding: &binding::ResolvedBinding, + cost: &[aura_research::CostSpec], ) -> RunReport { 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 mut h = wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, true, pip, binding, None) + // The doc's cost model as an optional wrap leg (#234): empty = no leg, + // exactly the pre-cost graph. The net curve is a !reduce trace concern; + // its sender is wired but unread here (the r_equity tap precedent above). + let (tx_cost, rx_cost) = mpsc::channel(); + let (tx_net, _rx_net) = mpsc::channel(); + let cost_leg = (!cost.is_empty()).then(|| CostLeg { + nodes: campaign_run::cost_nodes_for(cost), + tx_cost, + tx_net, + }); + let mut h = wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, true, pip, binding, cost_leg) .bootstrap_with_cells(point) .expect("member bootstraps (point kind-checked against param_space)"); h.run(sources); @@ -1605,11 +1654,27 @@ fn run_blueprint_member( named.push(("stop_length".to_string(), Scalar::i64(length))); named.push(("stop_k".to_string(), Scalar::f64(k))); } + // Stamp the cost model the member ran under, beside the stop knobs (#234, + // the #233 pattern): one `cost[k].` param per component, in + // component order — the knob name discriminates the variant (each shipped + // cost node has exactly one, distinctly named knob). The name is read off + // `campaign_run::cost_knob`, the same function `cost_nodes_for` binds + // through, so the stamp key cannot drift from the bind key: the manifest + // carries enough to re-derive the exact model later. `reproduce_family_in` + // reads this stamp back via `cost_specs_from_params` (the #233 stop-regime + // pattern), so a costed family reproduces net, not gross. + for (k, spec) in cost.iter().enumerate() { + let (knob, v) = campaign_run::cost_knob(spec); + named.push((format!("cost[{k}].{knob}"), Scalar::f64(v))); + } let mut manifest = sim_optimal_manifest(named, window, seed, pip); manifest.broker = r_sma_broker_label(pip); manifest.topology_hash = Some(topo.to_string()); manifest.project = env.provenance(); let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); + // The member's cost rows (empty when no cost model) join the R reduction: + // net_expectancy_r diverges from gross by exactly the modelled costs. + let cost_rows: Vec<(Timestamp, Vec)> = rx_cost.try_iter().collect(); let (total_pips, max_drawdown) = rx_eq .try_iter() .next() @@ -1618,7 +1683,7 @@ fn run_blueprint_member( 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, &[])); + m.r = Some(summarize_r(&r_rows, &cost_rows)); RunReport { manifest, metrics: m } } @@ -1725,7 +1790,7 @@ fn blueprint_sweep_family( .sweep(|point| { // fresh per-member graph (Composite is !Clone, reload per member) run through // the shared reduce-mode member path — the same fn reproduction re-runs. - run_blueprint_member(reload(doc), point, &space, data.run_sources(env, &binding.columns()), window, 0, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, &binding) + run_blueprint_member(reload(doc), point, &space, data.run_sources(env, &binding.columns()), window, 0, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, &binding, &[]) }) // render the sweep terminal's BindError to a message (the fn's String error contract), // so `UnknownKnob("nope")` still surfaces verbatim at the IO wrapper. @@ -1758,7 +1823,7 @@ fn blueprint_sweep_over( binder.sweep_with_lattice(|point| { let sources = data.windowed_sources(from, to, env, &binding.columns()); let window = window_of(&sources).expect("non-empty in-sample window"); - run_blueprint_member(reload(doc), point, &space, sources, window, 0, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, binding) + run_blueprint_member(reload(doc), point, &space, sources, window, 0, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, binding, &[]) }) } @@ -1776,7 +1841,7 @@ fn run_oos_blueprint( let pip = data.pip_size(); let sources = data.windowed_sources(from, to, env, &binding.columns()); let window = window_of(&sources).expect("non-empty out-of-sample window"); - let report = run_blueprint_member(reload, params, space, sources, window, 0, pip, topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, binding); + let report = run_blueprint_member(reload, params, space, sources, window, 0, pip, topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, binding, &[]); (Vec::new(), report) } @@ -1903,7 +1968,7 @@ fn blueprint_mc_family( let family = monte_carlo(&base_point, &seeds, |seed, _base| { let sources = synthetic_walk_sources(seed); let window = window_of(&sources).expect("non-empty synthetic walk"); - run_blueprint_member(reload(doc), &[], &space, sources, window, seed, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, &binding) + run_blueprint_member(reload(doc), &[], &space, sources, window, seed, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, &binding, &[]) }); // Silent-vacuous MC guard (refuse-don't-guess, C10): with >= 2 seeds, if every draw's // metrics are bit-identical to the first, no seed reached a distinguishable realization — @@ -3964,6 +4029,184 @@ mod tests { ); } + /// #234: a member run under a constant cost model nets the hand-computed + /// per-trade cost — `net_expectancy_r ≈ expectancy_r − mean(cost_per_trade + /// / |entry − stop|)` over summarize_r's ledger (closed rows + the + /// window-end open row; the CostRunner R-normalization contract), while + /// every gross field stays byte-identical to the cost-less run; and the + /// manifest stamps the component for reproduce to re-derive. + #[test] + fn run_blueprint_member_joins_a_constant_cost_model_into_net_metrics() { + let env = project::Env::std(); + let data = DataSource::Synthetic; + let doc = blueprint_to_json(&load_closed_r_sma()).expect("serializes"); + let reload = || blueprint_from_json(&doc, &|t| std_vocabulary(t)).expect("loads"); + let space = blueprint_axis_probe(&doc, &env).param_space(); + let binding = binding::resolve_binding("costnet", reload().input_roles(), &BTreeMap::new()) + .expect("the price role resolves"); + let stop = StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }; + let window = data.full_window(&env); + let pip = data.pip_size(); + const CPT: f64 = 0.0005; // price units; the synthetic stream trades near 1.0 + let run = |cost: &[aura_research::CostSpec]| { + run_blueprint_member( + reload(), + &[], + &space, + data.run_sources(&env, &binding.columns()), + window, + 0, + pip, + "topo", + &env, + stop, + &binding, + cost, + ) + }; + let gross = run(&[]); + let netted = run(&[aura_research::CostSpec::Constant { cost_per_trade: CPT }]); + + // The trade geometry, independently: a non-reduce cost-less wrap over + // the same realization, draining the dense R record whose ledger + // summarize_r reads. + 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 flat = wrap_r(reload(), tx_eq, tx_ex, tx_r, tx_req, stop, false, pip, &binding, None) + .compile_with_params(&[]) + .expect("wraps"); + let mut h = Harness::bootstrap(flat).expect("bootstraps"); + h.run(data.run_sources(&env, &binding.columns())); + let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); + + // summarize_r's ledger: closed rows, plus the final row if open — each + // charged cost_per_trade / |entry − stop| (0 when the latched distance + // is 0). PM record cols: closed=0, entry_price=6, stop_price=7, + // open=11 (the aura-analysis r_col contract). + let mut costs: Vec = Vec::new(); + for (i, (_, row)) in r_rows.iter().enumerate() { + let is_last = i == r_rows.len() - 1; + if row[0].as_bool() || (is_last && row[11].as_bool()) { + let latched = (row[6].as_f64() - row[7].as_f64()).abs(); + costs.push(if latched > 0.0 { CPT / latched } else { 0.0 }); + } + } + let g = gross.metrics.r.as_ref().expect("gross member carries R metrics"); + let n = netted.metrics.r.as_ref().expect("netted member carries R metrics"); + assert_eq!( + costs.len() as u64, g.n_trades, + "the independent ledger walk must see the member's trades" + ); + assert!(costs.iter().any(|&c| c > 0.0), "at least one trade must charge (non-vacuous)"); + let mean_cost = costs.iter().sum::() / costs.len() as f64; + + assert_eq!(gross.metrics.total_pips, netted.metrics.total_pips, "gross pips unchanged"); + assert_eq!(g.expectancy_r, n.expectancy_r, "gross R stays byte-identical under cost"); + assert_eq!(g.n_trades, n.n_trades); + assert_eq!(g.net_expectancy_r, g.expectancy_r, "cost-less: net == gross"); + assert!( + (n.net_expectancy_r - (g.expectancy_r - mean_cost)).abs() < 1e-12, + "net = gross − mean per-trade cost: net {} gross {} mean_cost {mean_cost}", + n.net_expectancy_r, + g.expectancy_r + ); + // The manifest stamps the component (the reproduce re-derivation carrier)... + assert!( + netted + .manifest + .params + .iter() + .any(|(k, v)| k == "cost[0].cost_per_trade" && *v == Scalar::f64(CPT)), + "member manifest stamps the cost component: {:?}", + netted.manifest.params + ); + // ...and a cost-less member stamps nothing (content/label stability). + assert!( + !gross.manifest.params.iter().any(|(k, _)| k.starts_with("cost[")), + "a cost-less member stamps no cost params" + ); + } + + /// #234: `campaign_run::persist_campaign_traces`'s C1 drift alarm re-runs a + /// costed member in `!reduce` mode (fresh per-cycle taps + a `CostLeg` + /// bound the same way `cost_nodes_for` binds it) and compares the result + /// against the recorded reduce-mode member. This pins that cross-mode + /// equivalence holds UNDER a non-empty cost model: the `!reduce` + + /// `CostLeg` wiring (this test, verbatim) and the reduce-mode + /// `run_blueprint_member` path must net to the exact same `RunMetrics` + /// over the same synthetic realization — the equality the drift alarm + /// structurally depends on to not false-positive on every legitimate + /// costed campaign. + #[test] + fn persist_side_nonreduce_rerun_matches_reduce_mode_net_metrics_under_cost() { + let env = project::Env::std(); + let data = DataSource::Synthetic; + let doc = blueprint_to_json(&load_closed_r_sma()).expect("serializes"); + let reload = || blueprint_from_json(&doc, &|t| std_vocabulary(t)).expect("loads"); + let space = blueprint_axis_probe(&doc, &env).param_space(); + let binding = + binding::resolve_binding("persistnet", reload().input_roles(), &BTreeMap::new()) + .expect("the price role resolves"); + let stop = StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }; + let window = data.full_window(&env); + let pip = data.pip_size(); + let cost = [aura_research::CostSpec::Constant { cost_per_trade: 0.0005 }]; + + // The recorded member: reduce-mode, cost-bound — `run_blueprint_member`, + // the exact path `CliMemberRunner::run_member` calls. + let recorded = run_blueprint_member( + reload(), + &[], + &space, + data.run_sources(&env, &binding.columns()), + window, + 0, + pip, + "topo", + &env, + stop, + &binding, + &cost, + ); + + // The persist-side re-run, verbatim structure: `!reduce` + a `CostLeg` + // built through the SAME `campaign_run::cost_nodes_for`, then + // `summarize` + `summarize_r(&r_rows, &cost_rows)` — exactly + // `persist_campaign_traces`'s C1 drift-alarm computation. + 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 cost_leg = + Some(CostLeg { nodes: campaign_run::cost_nodes_for(&cost), tx_cost, tx_net }); + let flat = wrap_r(reload(), tx_eq, tx_ex, tx_r, tx_req, stop, false, pip, &binding, cost_leg) + .compile_with_params(&[]) + .expect("the persist-side wrap builds"); + let mut h = Harness::bootstrap(flat).expect("the persist-side harness bootstraps"); + h.run(data.run_sources(&env, &binding.columns())); + let eq_rows: Vec<(Timestamp, Vec)> = rx_eq.try_iter().collect(); + let ex_rows: Vec<(Timestamp, Vec)> = rx_ex.try_iter().collect(); + let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); + let cost_rows: Vec<(Timestamp, Vec)> = rx_cost.try_iter().collect(); + let mut rerun_metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0)); + rerun_metrics.r = Some(summarize_r(&r_rows, &cost_rows)); + + assert!( + rerun_metrics.r.as_ref().is_some_and(|r| r.n_trades > 0), + "the fixture must actually close at least one costed trade" + ); + assert_eq!( + rerun_metrics, recorded.metrics, + "the persist path's !reduce + CostLeg re-run must net to the exact same \ + RunMetrics as the reduce-mode member under the same cost model — the \ + C1 drift alarm `persist_campaign_traces` relies on" + ); + } + /// 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` @@ -4001,10 +4244,10 @@ mod tests { let binding = binding::resolve_binding(signal_a.name(), signal_a.input_roles(), &BTreeMap::new()) .expect("the price role resolves"); let report_a = run_blueprint_member( - signal_a, &[], &space, make_source(), window, 0, pip_a, "topo", &env, stop, &binding, + signal_a, &[], &space, make_source(), window, 0, pip_a, "topo", &env, stop, &binding, &[], ); let report_b = run_blueprint_member( - r_meanrev_signal(Some(3), Some(0.0)), &[], &space, make_source(), window, 0, pip_b, "topo", &env, stop, &binding, + r_meanrev_signal(Some(3), Some(0.0)), &[], &space, make_source(), window, 0, pip_b, "topo", &env, stop, &binding, &[], ); // Guard: the run must have actually traded, else the invariant is vacuous. assert!( @@ -4636,6 +4879,7 @@ mod tests { &env, non_default, &binding, + &[], ); // persist exactly as the sweep/campaign paths do: store the blueprint, append the family. @@ -4659,6 +4903,68 @@ mod tests { let _ = std::fs::remove_dir_all(&dir); } + /// #234: a family minted under a cost model must reproduce bit-identically — + /// reproduce re-derives the components from the member manifest (the #233 + /// stop-regime pattern). Without the re-derivation the re-run joins an + /// empty cost slice, its net_expectancy_r reverts to gross, and the member + /// reports DIVERGED. All three variants ride along so every knob name + /// round-trips (the vol_slippage component also exercises the reduce-mode + /// vol proxy). + #[test] + fn reproduce_family_re_derives_a_costed_member_bit_identically() { + let dir = std::env::temp_dir().join(format!("aura-repro-cost-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("temp dir"); + let reg = Registry::open(dir.join("runs.jsonl")); + + let env = project::Env::std(); + let data = DataSource::Synthetic; + let doc = blueprint_to_json(&load_closed_r_sma()).expect("serializes"); + let reload = || blueprint_from_json(&doc, &|t| std_vocabulary(t)).expect("loads"); + let topo = topology_hash(&reload()); + let space = blueprint_axis_probe(&doc, &env).param_space(); + let pip = data.pip_size(); + let window = data.full_window(&env); + let binding = binding::resolve_binding("costrepro", reload().input_roles(), &BTreeMap::new()) + .expect("the price role resolves"); + let cost = vec![ + aura_research::CostSpec::Constant { cost_per_trade: 0.0005 }, + aura_research::CostSpec::VolSlippage { slip_vol_mult: 0.5 }, + aura_research::CostSpec::Carry { carry_per_cycle: 0.0001 }, + ]; + let report = run_blueprint_member( + reload(), + &[], + &space, + data.run_sources(&env, &binding.columns()), + window, + 0, + pip, + &topo, + &env, + StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, + &binding, + &cost, + ); + // Non-vacuity: the cost model must actually bite, else a DIVERGED + // verdict could never be observed and this pin proves nothing. + let r = report.metrics.r.as_ref().expect("member carries R metrics"); + assert_ne!(r.net_expectancy_r, r.expectancy_r, "the cost model must move net off gross"); + + let canonical = blueprint_to_json(&reload()).unwrap(); + reg.put_blueprint(&topo, &canonical).expect("store blueprint"); + let id = reg.append_family("costrepro", FamilyKind::Sweep, &[report]).expect("append"); + + let rep = reproduce_family_in(®, &id, &data, &env); + assert_eq!(rep.outcomes.len(), 1, "one member reproduced"); + assert!( + rep.outcomes.iter().all(|(_, ok)| *ok), + "the costed member re-derives bit-identically: {:?}", + rep.outcomes + ); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn reproduce_family_re_derives_every_mc_member_bit_identically() { let dir = std::env::temp_dir().join(format!("aura-repro-mc-{}", std::process::id())); diff --git a/crates/aura-cli/tests/research_docs.rs b/crates/aura-cli/tests/research_docs.rs index 361b4ab..00164c5 100644 --- a/crates/aura-cli/tests/research_docs.rs +++ b/crates/aura-cli/tests/research_docs.rs @@ -2359,6 +2359,184 @@ fn campaign_persist_non_default_regime_does_not_false_fail_c1() { ); } +/// Property (#234 Task 5, quality-review Minor finding): a costed campaign's +/// trace-persist re-run must not false-fail the C1 drift alarm, exercised +/// through the REAL production path — `campaign run` -> `run_campaign_returning` +/// -> `CliMemberRunner` (net-mode member run) -> `persist_campaign_traces` +/// (non-reduce re-run binding a `CostLeg` via `cost_nodes_for`) — not the +/// hand-mirrored arithmetic pinned by +/// `persist_side_nonreduce_rerun_matches_reduce_mode_net_metrics_under_cost` in +/// `main.rs`'s unit tests. Mirrors the sibling non-default-regime e2e above +/// (same #219-class divergence risk, cost edition): a costed campaign +/// document, `persist_taps` non-empty so the re-run actually executes, must +/// exit 0 with its nominee's cost stamp on disk and no C1 violation. Gated on +/// the local GER40 archive; skips on a data-less host. +#[test] +fn campaign_run_real_e2e_costed_campaign_persists_without_drift_alarm() { + let _fixture = project_lock(); + let dir = built_project(); + let runs_dir = dir.join("runs"); + std::fs::remove_dir_all(&runs_dir).ok(); + let _cleanup = ScratchGuard(vec![ + ScratchPath::Dir(runs_dir.clone()), + ScratchPath::File(dir.join("cost-persist.process.json")), + ScratchPath::File(dir.join("cost-persist.campaign.json")), + ]); + let bp_id = seed_blueprint(dir, "campaign-persist-cost-seed"); + let proc_id = register_process_doc(dir, "cost-persist.process.json", SWEEP_ONLY_PROCESS_DOC); + let base = campaign_doc_json( + &bp_id, + &proc_id, + (1725148800000, 1727740799999), + "\"equity\", \"r_equity\"", + "\"family_table\"", + ); + let with_cost = base.replacen( + "\"seed\": 7,", + "\"seed\": 7,\n \"cost\": [ { \"constant\": { \"cost_per_trade\": 0.0005 } } ],", + 1, + ); + assert_ne!(with_cost, base, "replacen must actually match the fixture's seed field"); + write_doc(dir, "cost-persist.campaign.json", &with_cost); + let (out, code) = run_code_in(dir, &["campaign", "run", "cost-persist.campaign.json"]); + + // Skip on a data-less machine: the member-data refusal, never a panic. + if code == Some(1) + && (out.contains("no recorded geometry") || out.contains("no data for instrument")) + { + eprintln!("skip: no local GER40 data for the campaign cost-persist e2e"); + return; + } + + assert!( + !out.contains("C1 violation"), + "the costed persist-side re-run must bind the SAME cost model the nominee ran \ + under, not false-fail C1: {out}" + ); + assert_eq!(code, Some(0), "a deterministic costed campaign persists cleanly: {out}"); + + // Non-vacuity: every emitted member actually stamped the cost component + // it ran under (else the C1 comparison above never engaged a non-empty + // CostLeg and this test would pass for the wrong reason). + let member_lines: Vec<&str> = + out.lines().filter(|l| l.starts_with("{\"family_id\":")).collect(); + assert!(!member_lines.is_empty(), "family_table member lines emitted: {out}"); + for line in &member_lines { + let v: serde_json::Value = serde_json::from_str(line).expect("member line parses as JSON"); + let params = v["report"]["manifest"]["params"] + .as_array() + .expect("manifest.params is an array"); + assert!( + params.iter().any(|p| p[0].as_str() == Some("cost[0].cost_per_trade")), + "the costed member stamps its cost component: {line}" + ); + } + + // The nominee's taps actually landed under the costed re-run. + let record_line = out + .lines() + .find(|l| l.starts_with("{\"campaign_run\":")) + .expect("the always-on final campaign_run line"); + let v: serde_json::Value = + serde_json::from_str(record_line).expect("campaign_run line parses as JSON"); + let trace_name = v["campaign_run"]["trace_name"] + .as_str() + .expect("persist_taps non-empty => the record carries trace_name"); + let cell_key = format!("{}-GER40-w0", &bp_id[..8]); + let cell_dir = runs_dir.join("traces").join(trace_name).join(&cell_key); + assert!( + cell_dir.join("r_equity.json").is_file(), + "the costed nominee's r_equity tap persists at {}", + cell_dir.display() + ); +} + +/// Property (#234): a `vol_slippage` cost component — the structurally +/// distinct leg from `constant` (it reactivates the vol-proxy RollingMax/ +/// RollingMin/Sub chain feeding a `VolSlippageCost` node, not a flat per-close +/// charge) — nets a real campaign member through the SAME production path as +/// its `constant` sibling +/// (`campaign_run_real_e2e_costed_campaign_persists_without_drift_alarm`): +/// `campaign run` -> `CliMemberRunner` (net-mode) -> `persist_campaign_traces` +/// (non-reduce re-run binding a `CostLeg`). The re-run must bind the identical +/// vol-slippage model the nominee ran under, so it must not false-fail the C1 +/// drift alarm; the emitted member must stamp `cost[0].slip_vol_mult`; and the +/// vol-driven charge must actually move net off gross (non-vacuity — a +/// zero-effect charge would let this pass even if the vol proxy were never +/// wired). Gated on the local GER40 archive; skips on a data-less host. +#[test] +fn campaign_run_real_e2e_vol_slippage_cost_persists_without_drift_alarm() { + let _fixture = project_lock(); + let dir = built_project(); + let runs_dir = dir.join("runs"); + std::fs::remove_dir_all(&runs_dir).ok(); + let _cleanup = ScratchGuard(vec![ + ScratchPath::Dir(runs_dir.clone()), + ScratchPath::File(dir.join("volcost-persist.process.json")), + ScratchPath::File(dir.join("volcost-persist.campaign.json")), + ]); + let bp_id = seed_blueprint(dir, "campaign-persist-volcost-seed"); + let proc_id = + register_process_doc(dir, "volcost-persist.process.json", SWEEP_ONLY_PROCESS_DOC); + let base = campaign_doc_json( + &bp_id, + &proc_id, + (1725148800000, 1727740799999), + "\"r_equity\"", + "\"family_table\"", + ); + let with_cost = base.replacen( + "\"seed\": 7,", + "\"seed\": 7,\n \"cost\": [ { \"vol_slippage\": { \"slip_vol_mult\": 0.5 } } ],", + 1, + ); + assert_ne!(with_cost, base, "replacen must actually match the fixture's seed field"); + write_doc(dir, "volcost-persist.campaign.json", &with_cost); + let (out, code) = run_code_in(dir, &["campaign", "run", "volcost-persist.campaign.json"]); + + // Skip on a data-less machine: the member-data refusal, never a panic. + if code == Some(1) + && (out.contains("no recorded geometry") || out.contains("no data for instrument")) + { + eprintln!("skip: no local GER40 data for the campaign vol-slippage-persist e2e"); + return; + } + + assert!( + !out.contains("C1 violation"), + "the vol-slippage persist-side re-run must bind the SAME cost model the nominee \ + ran under, not false-fail C1: {out}" + ); + assert_eq!(code, Some(0), "a deterministic vol-slippage-costed campaign persists cleanly: {out}"); + + let member_lines: Vec<&str> = + out.lines().filter(|l| l.starts_with("{\"family_id\":")).collect(); + assert!(!member_lines.is_empty(), "family_table member lines emitted: {out}"); + let mut any_net_moved = false; + for line in &member_lines { + let v: serde_json::Value = serde_json::from_str(line).expect("member line parses as JSON"); + let params = v["report"]["manifest"]["params"] + .as_array() + .expect("manifest.params is an array"); + assert!( + params.iter().any(|p| p[0].as_str() == Some("cost[0].slip_vol_mult")), + "the costed member stamps its vol_slippage component: {line}" + ); + if let Some(r) = v["report"]["metrics"]["r"].as_object() { + let gross = r["expectancy_r"].as_f64().expect("expectancy_r is a number"); + let net = r["net_expectancy_r"].as_f64().expect("net_expectancy_r is a number"); + if gross != net { + any_net_moved = true; + } + } + } + assert!( + any_net_moved, + "at least one member's vol-slippage charge must move net off gross \ + (else the vol proxy never engaged): {out}" + ); +} + /// RED (#212 — the trace-dir collision facet of the same #219 record gap): two /// risk regimes over the SAME (strategy, instrument, window) cell must persist /// their traces into DISTINCT directories. `campaign_cell_key` names dirs