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

feat(cli,engine): multi-column opening on every real-data path (#231 task 4)

Browse Source 
The six Close-only open sites route through the resolved binding's
column set via open_columns (probe_window, open_real_source,
run_sources/windowed_sources, the campaign member open, the trace
re-run open) — a strategy declaring high/low/close roles gets exactly
those columns, merged in the canonical C4 order. Synthetic data refuses
any binding beyond {close} with the honest --real remedy (the walk
generates a close series only); the sweep/mc family builders ride their
established Err contract, run/walkforward/reproduce refuse exit-1.

Proof at two layers: an engine e2e composes an OHLC blueprint from data
and runs it over four inline VecSources to hand-computed rows (the
merge-order pin), and an archive-gated CLI e2e sweeps a high/low
blueprint over GER40 opening the High and Low columns.

Verified: engine OHLC e2e green, full workspace suite green, clippy -D
warnings clean.

refs #231
This commit is contained in:
Brummel
2026-07-11 03:37:14 +02:00
parent 99a7f6603c
commit 1d14ebc1cd
4 changed files with 428 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aura-cli/src/campaign_run.rs
+22 -20
View File
@@ -24,7 +24,7 @@ use aura_engine::{
blueprint_from_json, f64_field, summarize, summarize_r, ColumnarTrace, FamilySelection,
RunReport,
};
use aura_ingest::{instrument_geometry, unix_ms_to_epoch_ns, M1Field, M1FieldSource};
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,
@@ -263,12 +263,14 @@ impl MemberRunner for CliMemberRunner<'_> {
})?;
let from = unix_ms_to_epoch_ns(window_ms.0);
let to = unix_ms_to_epoch_ns(window_ms.1);
let source = match M1FieldSource::open_window(
// One source per resolved binding column, canonical order (the same
// order `wrap_r` declares the roles in — the shared plan).
let sources = match open_columns_window(
&self.server,
&cell.instrument,
Some(from),
Some(to),
M1Field::Close,
&binding.columns(),
) {
Some(s) => s,
// No archived file overlaps the window at all.
@@ -279,10 +281,11 @@ impl MemberRunner for CliMemberRunner<'_> {
});
}
};
// A window that overlaps a file but holds zero matching bars yields a
// source whose first peek is None (open_window's documented contract)
// — the same no-data condition.
if aura_engine::Source::peek(&source).is_none() {
// A window that overlaps a file but holds zero matching bars yields
// sources whose first peek is None (open_window's documented contract)
// — the same no-data condition (all columns decode the same bars, so
// probing the first source covers the set).
if aura_engine::Source::peek(sources[0].as_ref()).is_none() {
return Err(MemberFault::NoData {
instrument: cell.instrument.clone(),
window_ms,
@@ -298,7 +301,7 @@ impl MemberRunner for CliMemberRunner<'_> {
signal,
&point,
&space,
vec![Box::new(source)],
sources,
(from, to),
0,
geo.pip_size,
@@ -823,23 +826,23 @@ pub(crate) fn persist_campaign_traces(
cell_rec.instrument, from.0, to.0
)
};
let source = M1FieldSource::open_window(
server,
&cell_rec.instrument,
Some(from),
Some(to),
M1Field::Close,
)
.ok_or_else(no_data)?;
if aura_engine::Source::peek(&source).is_none() {
return Err(no_data());
}
let signal = blueprint_from_json(blueprint_json, &|t| env.resolve(t))
.expect("stored blueprint passed the referential gate; reload is infallible");
// Same name-default binding the member ran under (campaign
// overrides thread through in the DataSection task).
let binding =
crate::binding::resolve_binding(&cell_rec.strategy, signal.input_roles(), &BTreeMap::new())?;
let sources = open_columns_window(
server,
&cell_rec.instrument,
Some(from),
Some(to),
&binding.columns(),
)
.ok_or_else(no_data)?;
if aura_engine::Source::peek(sources[0].as_ref()).is_none() {
return Err(no_data());
}
// The cell's OWN regime (#219/#212), not the hardcoded default: the
// recorded member ran under `cell_rec.regime`, bound through the same
// `stop_rule_for_regime` helper `CliMemberRunner::run_member` uses, so
@@ -853,7 +856,6 @@ pub(crate) fn persist_campaign_traces(
let mut h = crate::wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, false, geo.pip_size, &binding)
.bootstrap_with_cells(&point)
.expect("the member's point re-bootstraps (it already ran this realization)");
let sources: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(source)];
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.
crates/aura-cli/src/main.rs
+113 -45
View File
@@ -516,11 +516,15 @@ fn pip_or_refuse(
}
}
/// Probe the full data window: open a single-pass probe `M1FieldSource`, drain it
/// for the first/last timestamp, and return `(first, last)`. Refuses (via
/// `no_real_data`) when the symbol/window yields no source or no bars. Shared by
/// `open_real_source` (which needs the manifest window from a probe separate from
/// the run source) and `DataSource::full_window`.
/// Probe the full data window: open a single-pass probe source through the
/// shared opener, drain it for the first/last timestamp, and return
/// `(first, last)`. Refuses (via `no_real_data`) when the symbol/window yields
/// no source or no bars. Shared by `open_real_source` (which needs the manifest
/// window from a probe separate from the run sources) and
/// `DataSource::full_window`. The probe drains the CLOSE column regardless of
/// the strategy's binding: the bounds are field-independent (every archived bar
/// carries all six fields at one timestamp), and file-level absence is
/// field-independent too.
fn probe_window(
server: &std::sync::Arc<data_server::DataServer>,
symbol: &str,
@@ -528,28 +532,32 @@ fn probe_window(
to_ms: Option<i64>,
env: &project::Env,
) -> (Timestamp, Timestamp) {
let mut probe = aura_ingest::M1FieldSource::open(server, symbol, from_ms, to_ms, aura_ingest::M1Field::Close)
.unwrap_or_else(|| no_real_data(symbol, env));
let first = aura_engine::Source::peek(&probe).unwrap_or_else(|| no_real_data(symbol, env));
let mut probe = aura_ingest::open_columns(server, symbol, from_ms, to_ms, &[aura_ingest::M1Field::Close])
.unwrap_or_else(|| no_real_data(symbol, env))
.pop()
.expect("open_columns yields one source per requested field");
let first = aura_engine::Source::peek(probe.as_ref()).unwrap_or_else(|| no_real_data(symbol, env));
let mut last = first;
while let Some((t, _)) = aura_engine::Source::next(&mut probe) {
while let Some((t, _)) = aura_engine::Source::next(&mut *probe) {
last = t;
}
(first, last)
}
/// Open a real M1-close source for a recorded symbol over an optional window, returning
/// the run source paired with its manifest `window` and per-instrument `pip_size`.
/// Open the real M1 sources for a recorded symbol over an optional window — one
/// source per resolved binding column, in canonical order — returning the run
/// sources paired with the manifest `window` and the per-instrument `pip_size`.
/// Single home of the real-source construction the single-run handlers share — the
/// sidecar-pip lookup, the `DataServer` `has_symbol` refusal, the probe-window pass, and
/// the run-source `open` (each refusal an stderr + exit 1). Pre-data refusals keep the
/// the run-source open (each refusal an stderr + exit 1). Pre-data refusals keep the
/// pip honest by construction. Used by `resolve_run_data`.
fn open_real_source(
symbol: &str,
from_ms: Option<i64>,
to_ms: Option<i64>,
env: &project::Env,
) -> (Box<dyn aura_engine::Source>, (Timestamp, Timestamp), f64) {
fields: &[aura_ingest::M1Field],
) -> (Vec<Box<dyn aura_engine::Source>>, (Timestamp, Timestamp), f64) {
// Per-instrument pip from the recorded sidecar; resolved BEFORE bar-data access
// so an instrument with no geometry refuses without touching the archive.
let server = std::sync::Arc::new(data_server::DataServer::new(env.data_path()));
@@ -559,12 +567,9 @@ fn open_real_source(
}
// Manifest window: drain a separate probe (single-pass Source) for first/last ts.
let window = probe_window(&server, symbol, from_ms, to_ms, env);
let source: Box<dyn aura_engine::Source> =
match aura_ingest::M1FieldSource::open(&server, symbol, from_ms, to_ms, aura_ingest::M1Field::Close) {
Some(s) => Box::new(s),
None => no_real_data(symbol, env),
};
(source, window, pip)
let sources = aura_ingest::open_columns(&server, symbol, from_ms, to_ms, fields)
.unwrap_or_else(|| no_real_data(symbol, env));
(sources, window, pip)
}
impl DataSource {
@@ -625,28 +630,31 @@ impl DataSource {
}
}
/// A fresh full-window source per member (single-pass). Synthetic: showcase.
fn run_sources(&self, env: &project::Env) -> Vec<Box<dyn aura_engine::Source>> {
/// A fresh full-window source set per member (single-pass): the synthetic
/// showcase close stream, or one real source per resolved binding column
/// in canonical order (callers guard the synthetic arm to `{close}`).
fn run_sources(&self, env: &project::Env, fields: &[aura_ingest::M1Field]) -> Vec<Box<dyn aura_engine::Source>> {
match self {
DataSource::Synthetic => vec![Box::new(VecSource::new(showcase_prices()))],
DataSource::Real { server, symbol, from_ms, to_ms, .. } => vec![Box::new(
aura_ingest::M1FieldSource::open(server, symbol, *from_ms, *to_ms, aura_ingest::M1Field::Close)
.unwrap_or_else(|| no_real_data(symbol, env)),
)],
DataSource::Real { server, symbol, from_ms, to_ms, .. } => {
aura_ingest::open_columns(server, symbol, *from_ms, *to_ms, fields)
.unwrap_or_else(|| no_real_data(symbol, env))
}
}
}
/// A fresh windowed source for an IS/OOS sub-window (walk-forward). Synthetic:
/// `walkforward_window_source`. Real: `open_window` (ns-native `Timestamp`).
/// A fresh windowed source set for an IS/OOS sub-window (walk-forward).
/// Synthetic: `walkforward_window_source`. Real: one source per resolved
/// binding column over the ns-native window.
fn windowed_sources(
&self, from: Timestamp, to: Timestamp, env: &project::Env,
&self, from: Timestamp, to: Timestamp, env: &project::Env, fields: &[aura_ingest::M1Field],
) -> Vec<Box<dyn aura_engine::Source>> {
match self {
DataSource::Synthetic => vec![Box::new(walkforward_window_source(from, to))],
DataSource::Real { server, symbol, .. } => vec![Box::new(
aura_ingest::M1FieldSource::open_window(server, symbol, Some(from), Some(to), aura_ingest::M1Field::Close)
.unwrap_or_else(|| no_real_data(symbol, env)),
)],
DataSource::Real { server, symbol, .. } => {
aura_ingest::open_columns_window(server, symbol, Some(from), Some(to), fields)
.unwrap_or_else(|| no_real_data(symbol, env))
}
}
}
@@ -1118,6 +1126,10 @@ fn reproduce_family_in(
eprintln!("aura: {m}");
std::process::exit(1);
});
if matches!(data, DataSource::Synthetic) && !binding.close_only() {
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 point = point_from_params(&space, &stored.manifest.params);
// A MonteCarlo member carries no tuning params (the params-join is empty), so its
@@ -1150,7 +1162,7 @@ fn reproduce_family_in(
// stored window bounds; the winner params come from the shared
// manifest->cells recovery below (as Sweep members do).
let (from, to) = stored.manifest.window;
let s = data.windowed_sources(from, to, env);
let s = data.windowed_sources(from, to, env, &binding.columns());
let w = window_of(&s).expect("non-empty OOS window");
(s, w)
}
@@ -1163,9 +1175,9 @@ fn reproduce_family_in(
_ => match data {
DataSource::Real { .. } => {
let (from, to) = stored.manifest.window;
(data.windowed_sources(from, to, env), (from, to))
(data.windowed_sources(from, to, env, &binding.columns()), (from, to))
}
DataSource::Synthetic => (data.run_sources(env), data.full_window(env)),
DataSource::Synthetic => (data.run_sources(env, &binding.columns()), data.full_window(env)),
},
};
let rerun = run_blueprint_member(
@@ -1374,13 +1386,15 @@ fn wrap_r(
}
/// Resolve a `RunData` selector to the `(sources, window, pip_size)` triple the
/// r-sma run paths feed to the harness: the built-in synthetic R stream,
/// or a lazily-streamed real M1 close source (with its sidecar pip + probed window).
/// run paths feed to the harness: the built-in synthetic R stream (close-only —
/// the caller guards the binding shape), or the lazily-streamed real sources of
/// the binding's resolved columns (with the sidecar pip + probed window).
/// Single definition used by `run_signal_r`.
#[allow(clippy::type_complexity)]
fn resolve_run_data(
data: &RunData,
env: &project::Env,
binding: &binding::ResolvedBinding,
) -> (
Vec<Box<dyn aura_engine::Source>>,
(Timestamp, Timestamp),
@@ -1394,8 +1408,7 @@ fn resolve_run_data(
(sources, window, SYNTHETIC_PIP_SIZE)
}
RunData::Real { symbol, from, to } => {
let (source, window, pip_size) = open_real_source(symbol, *from, *to, env);
(vec![source], window, pip_size)
open_real_source(symbol, *from, *to, env, &binding.columns())
}
}
}
@@ -1416,6 +1429,10 @@ fn run_signal_r(
eprintln!("aura: {m}");
std::process::exit(1);
});
if matches!(data, RunData::Synthetic) && !binding.close_only() {
eprintln!("aura: {}", binding::synthetic_refusal(signal.name(), &binding));
std::process::exit(1);
}
let names: Vec<String> = signal
.param_space()
.iter()
@@ -1427,7 +1444,7 @@ fn run_signal_r(
// The req tap (r_equity recorder) is wired but not persisted on this path; keep the
// 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);
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 flat = wrapped
.compile_with_params(params)
@@ -1569,6 +1586,9 @@ fn blueprint_sweep_family(
// Strict binding resolution (name defaults — the verb path carries no
// campaign overrides): the family's open plan and wrap plan in one value.
let binding = binding::resolve_binding(probe_signal.name(), probe_signal.input_roles(), &BTreeMap::new())?;
if matches!(data, DataSource::Synthetic) && !binding.close_only() {
return Err(binding::synthetic_refusal(probe_signal.name(), &binding));
}
let pip = data.pip_size();
let window = data.full_window(env);
// a single throwaway floated build, only to resolve param_space (borrow) then seed
@@ -1601,7 +1621,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), 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.
@@ -1632,7 +1652,7 @@ fn blueprint_sweep_over(
binder = binder.axis(n, vals.clone());
}
binder.sweep_with_lattice(|point| {
let sources = data.windowed_sources(from, to, env);
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)
})
@@ -1650,7 +1670,7 @@ fn run_oos_blueprint(
let reload = blueprint_from_json(doc, &|t| env.resolve(t))
.expect("doc parse-validated at the dispatch boundary; reload is infallible");
let pip = data.pip_size();
let sources = data.windowed_sources(from, to, env);
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);
(Vec::new(), report)
@@ -1685,6 +1705,10 @@ fn blueprint_walkforward_family(
eprintln!("aura: {m}");
std::process::exit(1);
});
if matches!(data, DataSource::Synthetic) && !binding.close_only() {
eprintln!("aura: {}", binding::synthetic_refusal(probe_signal.name(), &binding));
std::process::exit(1);
}
// Validate the `--axis` grid ONCE at the dispatch boundary, mirroring `aura sweep`
// (which resolves its axes a single time before any member runs). `walk_forward` fans
// the per-window closure out across the windows in parallel, so a `BindError` raised
@@ -1746,6 +1770,11 @@ fn blueprint_mc_family(
// Strict binding resolution (name defaults — mc's synthetic family binds
// no campaign overrides); the exit-free Err contract of this builder.
let binding = binding::resolve_binding(probe_signal.name(), probe_signal.input_roles(), &BTreeMap::new())?;
if !binding.close_only() {
// MC draws ALWAYS run the seeded synthetic close walk (real-data mc
// routes through the campaign sugar and never reaches this builder).
return Err(binding::synthetic_refusal(probe_signal.name(), &binding));
}
let pip = data.pip_size();
// probe the wrapped param_space (the same probe the sweep resolves against);
// MC needs it empty. `blueprint_axis_probe` is the single source of that wrap.
@@ -3871,13 +3900,52 @@ mod tests {
let env = project::Env::std();
let d = DataSource::Synthetic;
assert_eq!(d.pip_size(), SYNTHETIC_PIP_SIZE);
assert!(!d.run_sources(&env).is_empty());
assert!(!d.run_sources(&env, &[aura_ingest::M1Field::Close]).is_empty());
assert_eq!(d.wf_window_sizes(), (24, 12, 12));
// full_window equals window_of over the showcase stream (byte-unchanged source)
let s: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(VecSource::new(showcase_prices()))];
assert_eq!(d.full_window(&env), window_of(&s).unwrap());
}
/// A multi-column blueprint over synthetic data (a single close walk)
/// refuses honestly through the builders' exit-free Err contract, naming
/// the beyond-close columns and the --real remedy — never a panic from a
/// source-count mismatch. High/low-consuming, closed (mc requires it),
/// with the mandatory `bias` output.
const OHLC_REFUSAL_BLUEPRINT: &str = r#"{
"format_version": 1,
"blueprint": {
"name": "hl_range",
"nodes": [ {"primitive":{"type":"Sub"}} ],
"edges": [],
"input_roles": [
{"name":"high","targets":[{"node":0,"slot":0}],"source":"F64"},
{"name":"low","targets":[{"node":0,"slot":1}],"source":"F64"}
],
"output": [{"node":0,"field":0,"name":"bias"}]
}
}"#;
#[test]
fn synthetic_data_refuses_a_multi_column_blueprint() {
let env = project::Env::std();
let err = blueprint_mc_family(OHLC_REFUSAL_BLUEPRINT, 2, &DataSource::Synthetic, &env)
.expect_err("a high/low blueprint cannot run over the synthetic close walk");
assert_eq!(
err,
"strategy \"hl_range\" consumes columns beyond close (high, low) — synthetic \
data generates a close series only; run with --real <SYMBOL>"
);
let err = blueprint_sweep_family(
OHLC_REFUSAL_BLUEPRINT,
&[("x".to_string(), vec![Scalar::i64(1)])],
&DataSource::Synthetic,
&env,
)
.expect_err("the synthetic sweep path refuses the same shape");
assert!(err.contains("consumes columns beyond close"), "got: {err}");
}
#[test]
fn wf_real_roller_sizes_are_90_30_30_days_in_ns() {
// Independent expected value: a day reconstructed from its time units
@@ -4134,7 +4202,7 @@ mod tests {
reload(),
&[],
&space,
data.run_sources(&env),
data.run_sources(&env, &binding.columns()),
window,
0,
pip,