From 4d72a2ca0713f8212f8a7696a31c0c5af18b66a0 Mon Sep 17 00:00:00 2001 From: vntrevx <20063774+vntrevx@users.noreply.github.com> Date: Tue, 11 Aug 2026 21:21:24 +0900 Subject: [PATCH 1/3] chore(roadmap): start M21-04 --- planning/roadmap-state.json | 25 ++++++++++++++++++++----- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/planning/roadmap-state.json b/planning/roadmap-state.json index d7c50f5..abe3644 100644 --- a/planning/roadmap-state.json +++ b/planning/roadmap-state.json @@ -1,10 +1,10 @@ { "schema_version": "1.0.0", "roadmap_id": "nfi-backtest-engine-post-v1.1.0", - "revision": 140, - "updated_at": "2026-08-11T20:43:47+09:00", + "revision": 141, + "updated_at": "2026-08-11T21:20:59+09:00", "acceptance_commands": "planning/acceptance-commands.json", - "active_task_id": null, + "active_task_id": "M21-04", "execution_policy": { "max_in_progress": 1, "selection": "lowest order pending task whose dependencies are completed", @@ -4917,7 +4917,7 @@ "order": 2104, "milestone": "M21", "title": "Connect the in-memory Rust vector and simulation pipeline", - "status": "pending", + "status": "in_progress", "depends_on": [ "M21-03" ], @@ -4936,7 +4936,7 @@ "end-to-end Native vector fixture", "transport parity and performance report" ], - "started_at": null, + "started_at": "2026-08-11T21:20:59+09:00", "completed_at": null, "commit_sha": null, "evidence": [], @@ -8592,6 +8592,21 @@ "next_eligible_task": "M21-04", "next_task_manual_gate": false } + }, + { + "sequence": 188, + "timestamp": "2026-08-11T21:20:59+09:00", + "task_id": "M21-04", + "event": "task_started", + "details": { + "objective": "connect generic Rust vector output to the simulator through an exact in-memory transport while preserving sealed Feather replay", + "sealed_feather_replay_must_remain": true, + "pair_dag_parallelism_allowed": true, + "parallel_wallet_mutation_allowed": false, + "transport_trade_surface_and_full_state_exact_required": true, + "transport_performance_evidence_required": true, + "runtime_hardcoding_allowed": false + } } ] } From d9529131eaa629cf598a97569931eed844e7066a Mon Sep 17 00:00:00 2001 From: vntrevx <20063774+vntrevx@users.noreply.github.com> Date: Tue, 11 Aug 2026 21:56:29 +0900 Subject: [PATCH 2/3] feat(vector): connect in-memory Rust pipeline --- CHANGELOG.md | 9 + README.md | 12 +- ...in-memory-transport-m21-04-2026-08-11.json | 74 + docs/architecture.md | 14 +- docs/native-in-memory-vector.md | 55 + docs/native-signal-program.md | 5 +- docs/native-vector-core.md | 4 +- docs/native-vector-shadow.md | 11 +- rust/Cargo.lock | 47 + rust/crates/nfi-vector-io/Cargo.toml | 2 + rust/crates/nfi-vector-io/src/failures.rs | 9 + rust/crates/nfi-vector-io/src/in_memory.rs | 1195 +++++++++++++++++ rust/crates/nfi-vector-io/src/lib.rs | 7 + rust/crates/nfi-vector-io/src/schema.rs | 8 +- rust/crates/nfi-vector-io/src/values.rs | 19 +- 15 files changed, 1449 insertions(+), 22 deletions(-) create mode 100644 benchmarks/evidence/native-in-memory-transport-m21-04-2026-08-11.json create mode 100644 docs/native-in-memory-vector.md create mode 100644 rust/crates/nfi-vector-io/src/in_memory.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 15ad617..ade3101 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,15 @@ All notable changes are recorded here. This project follows Semantic Versioning. +## Unreleased + +- Connected typed Rust vector output directly to the chronological simulator, with + pair-parallel DAG preparation and no parallel wallet or order mutation. +- Preserved SHA-verified Feather as the bounded-memory evidence/replay path and added + zero-tolerance trade-surface and every-candle full-state transport parity tests. +- Kept latest-X7 Full Native qualification and the v1.6.0 release behind the remaining + M22 source-lowering, dual-mode exactness, performance, storage, and three-OS gates. + ## 1.5.0 - 2026-08-10 - Promoted source-compiled generic state-machine programs to the sole current diff --git a/README.md b/README.md index 5f0f95f..4300cee 100644 --- a/README.md +++ b/README.md @@ -207,10 +207,14 @@ opens a Draft fixture PR. Neither kind is auto-approved or auto-merged. For supported X7 stateful callbacks, `run.json` records the `x7-generic-stateful` Native lane and every source-compiled primary program. The -specialized X7 vector transport remains only an input contract. Current manager payloads -execute generic programs without a handwritten X7 shadow; a missing or retired execution -mode blocks Native before simulation and leaves the visible official fallback available. -Historical schema readers remain available for sealed evidence replay. +Rust vector core can now pass typed pair output directly into the chronological simulator; +independent pair preparation is parallel, while wallet and order mutation remains one stable +timestamp stream. SHA-verified Feather remains the durable evidence/replay path, and both transports +must produce the same trade surface and every-candle state. Current manager payloads execute generic +programs without a handwritten X7 shadow; a missing or retired execution mode blocks Native before +simulation and leaves the visible official fallback available. Historical schema readers remain +available for sealed evidence replay. See +[Native In-Memory Vector Transport](docs/native-in-memory-vector.md). ## Confirm with official Freqtrade diff --git a/benchmarks/evidence/native-in-memory-transport-m21-04-2026-08-11.json b/benchmarks/evidence/native-in-memory-transport-m21-04-2026-08-11.json new file mode 100644 index 0000000..fa223b7 --- /dev/null +++ b/benchmarks/evidence/native-in-memory-transport-m21-04-2026-08-11.json @@ -0,0 +1,74 @@ +{ + "schema_version": "1.0.0", + "status": "diagnostic-only", + "recorded_at": "2026-08-11", + "claim_boundary": "M21-04 typed Rust vector-to-simulator transport. Exactness uses a four-row full-state fixture; performance uses four synthetic 25000-row pairs on one Linux development host. This is not latest-X7 qualification, a five-year benchmark, or a cross-platform release speed claim.", + "contract": { + "vector_program": "benchmarks/reference/vector-shadow/tag-program.json", + "vector_program_sha256": "eea095daeb5329274e3c825026cbed327ff40ada026efded143f6eba4db16e3e", + "transport_source": "rust/crates/nfi-vector-io/src/in_memory.rs", + "transport_source_sha256": "3cf2e59c50b1d286d21f54907a9c96cd69e54783d92a5e5b9a2b02d68e7894a8", + "signal_gate": "exact numeric == 1 or Boolean true", + "pair_order": "indexed input order", + "wallet_execution": "single global chronological loop", + "feather_replay_retained": true + }, + "exact_observation": { + "workload_rows": 4, + "pair_count": 1, + "trade_surface_exact": true, + "full_state_exact": true, + "comparison": "owned direct input versus SHA-verified file-backed Feather input", + "direct": { + "pair_prepare_ns": 984131, + "estimated_source_column_bytes": 358, + "estimated_simulation_owned_bytes": 1216 + }, + "feather": { + "manifest_ns": 74852, + "vector_hash_ns": 23223, + "feather_decode_ns": 219656, + "file_backed_bytes": 356 + } + }, + "performance_observation": { + "workload_rows": 100000, + "rows_per_pair": 25000, + "pair_count": 4, + "feature_column_count": 4, + "direct": { + "vector_execute_ns": 901416, + "pair_prepare_ns": 11833000, + "transport_ns": 12734416, + "estimated_source_column_bytes": 8137532, + "estimated_simulation_owned_bytes": 29600052, + "configured_worker_limit": 16 + }, + "feather": { + "manifest_ns": 95616, + "vector_hash_ns": 7248024, + "feather_decode_ns": 18585906, + "transport_ns": 25929546, + "file_backed_bytes": 8900000 + }, + "observed_transport_speedup": 2.036178651616219, + "trade_surface_exact": true + }, + "environment": { + "kernel": "Linux 7.0.0-28-generic", + "architecture": "x86_64", + "logical_cpu_limit": 16, + "memory_bytes": 50195083264, + "cargo_profile": "release" + }, + "commands": [ + "cargo test --release -p nfi-vector-io --locked in_memory::tests::direct_transport_is_full_state_exact_to_sealed_feather_replay -- --nocapture", + "cargo test --release -p nfi-vector-io --locked in_memory::tests::large_pair_transport_profile_is_reported_without_a_speed_assertion -- --ignored --nocapture" + ], + "limitations": [ + "Each timing was measured once and includes one local Rayon pool configuration.", + "The synthetic workload has no open trades and measures transport, not callback or simulation speed.", + "The direct path retains owned simulator rows; Feather remains the bounded-heap replay option.", + "M22-01 must still compile and independently qualify the latest NFI Spot and Futures source." + ] +} diff --git a/docs/architecture.md b/docs/architecture.md index 52d6db9..effc5d1 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -31,7 +31,11 @@ strategy.py + config + candles AST preflight + frozen input/data identity | v -SHA-bound Feather vectors + projected callback columns +compiled Vector IR + typed pair columns + | + +------> parallel Rust pair DAG ------> direct owned simulator input + | + +------> SHA-bound Feather ----------> verified file-backed replay | v Rust global chronological portfolio loop @@ -44,9 +48,10 @@ Rust global chronological portfolio loop common canonical state trace -----------------> full exact check ``` -Python is not called once per candle. Python owns preparation, columnar transforms, -reports, and proof artifacts. Rust owns mutable trade/order/wallet state and the hot -chronological loop. +Python is not called once per candle. Python owns source compilation, reports, and proof artifacts. +Supported compiled vector work and typed transport can remain inside Rust. Rust owns mutable +trade/order/wallet state and the hot chronological loop. Latest-X7 entry into this complete path is +still separately qualified by the roadmap; unsupported source constructs fail closed. The compact vector manifest stores a relative Feather path, file SHA-256, row count, feature names, pair limits, precision, and historical price-step changes. Rust @@ -79,6 +84,7 @@ competition, shared stake, rebuy timing, rejected signals, and equal-timestamp o Parallel work is limited to operations without shared mutable portfolio state: - indicators by pair; +- complete compiled vector DAGs and typed simulator-input conversion by pair; - independent strategies, timeranges, fixtures, and candidate jobs; - offline scoring and report generation. diff --git a/docs/native-in-memory-vector.md b/docs/native-in-memory-vector.md new file mode 100644 index 0000000..f4512a6 --- /dev/null +++ b/docs/native-in-memory-vector.md @@ -0,0 +1,55 @@ +# Native In-Memory Vector Transport + +M21-04 connects typed Rust vector output to the existing chronological simulator without writing +an intermediate Feather file. This is a transport change only: it does not change signal timing, +pair priority, wallet mutation, order IDs, or callback semantics. + +## Runtime contract + +`nfi-vector-io` accepts one `MutationFrame` per pair through `InMemoryVectorPair`. The frame carries +the same `date`, OHLCV, callback-feature, and `nfi_exec_*` columns as the sealed Feather manifest. +The adapter preserves: + +- millisecond timestamps and exact Float64 bits, including signed zero and canonical NaN features; +- Freqtrade's exact numeric-`1`/Boolean-`true` signal gate; +- raw non-empty entry and exit tags, including trailing whitespace; +- previous-close, funding-event, precision, stake-limit, and execution-start metadata; +- caller pair order in the resulting `SimulationInput`. + +Missing columns, mixed types, Boolean feature nulls, duplicate pairs, and invalid execution ranges +fail before simulation. Signal numbers, tags, pairs, strategy names, and source hashes remain data, +never runtime branches. + +## Safe parallel boundary + +`execute_in_memory_pair_dag_profiled` runs independent pair vector tasks with Rayon and collects +them by original index. Pair conversion is also independent and parallel. Both stages finish before +the returned input enters the simulator. The simulator continues to mutate one shared wallet in one +global timestamp stream; equal-timestamp pair order is unchanged. + +Rayon uses its configured worker limit, including `RAYON_NUM_THREADS`; no machine-specific thread +count is compiled into the engine. `InMemoryVectorProfile` records vector execution, pair conversion, +row/column counts, estimated source-buffer and retained simulator bytes, and the configured worker +limit. + +## Feather replay proof + +The SHA-verified Feather path remains supported. A focused Rust test writes one typed frame to +Feather, loads its file-backed representation, and compares it with the direct owned representation. +Both are run through the simulator with an every-candle observer; `SimulationResult` and every full +state event must be exactly equal. + +```bash +cargo test --manifest-path rust/Cargo.toml -p nfi-vector-io --locked +``` + +The direct path removes vector hashing, IPC decoding, and private spool I/O from a Native pipeline. +The Feather path remains the durable evidence and replay format and is still preferable when bounded +heap use matters more than transport latency. + +## Claim boundary + +This completes the generic Rust vector-to-simulator connection. It does not claim that the latest X7 +source fully compiles: M22-01 still owns remaining source lowering and independent latest-upstream +Spot/Futures full-state qualification. Until that passes, unsupported X7 constructs remain +fail-closed and the visible official Freqtrade fallback remains available. diff --git a/docs/native-signal-program.md b/docs/native-signal-program.md index 054a3b4..fc848aa 100644 --- a/docs/native-signal-program.md +++ b/docs/native-signal-program.md @@ -45,5 +45,6 @@ source location. Signal numbers, strategy names, pairs, timeranges, and source h never runtime branches; the source hash is evidence identity only. The independent Rust execution and exact raw-signal comparison are documented in -[`native-vector-shadow.md`](native-vector-shadow.md). Production in-memory simulator transport -remains the separate M21-04 boundary. +[`native-vector-shadow.md`](native-vector-shadow.md). The completed +[`native-in-memory-vector.md`](native-in-memory-vector.md) transport carries its shifted output into +the simulator without changing the Signal contract. diff --git a/docs/native-vector-core.md b/docs/native-vector-core.md index 9f4af68..878a187 100644 --- a/docs/native-vector-core.md +++ b/docs/native-vector-core.md @@ -127,8 +127,8 @@ M20-05 establishes exact multi-timeframe primitives and compiler contracts, but their in-memory `VectorEngine` connection is deliberately owned by M21. M21-01 and M21-02 establish Native signal assignment and exact tag-generation contracts. M21-03 independently proves Python/Rust Indicator, Signal, Tag, and -execution-index equality. M21-04 owns the in-memory simulator connection. M22 owns -latest-upstream Spot/Futures +execution-index equality. M21-04 connects typed pair output to the simulator, preserves +Feather replay, and keeps wallet mutation single-threaded. M22 owns latest-upstream Spot/Futures full-state qualification, removal of Python strategy execution from the Native lane, and release certification. Until those proofs pass, unsupported source constructs remain fail-closed and the official Freqtrade fallback remains the diff --git a/docs/native-vector-shadow.md b/docs/native-vector-shadow.md index 63f9738..e757fcb 100644 --- a/docs/native-vector-shadow.md +++ b/docs/native-vector-shadow.md @@ -35,8 +35,9 @@ unsupported helper behavior, and any exact-value difference. ## Claim boundary -This proves independent Rust vector semantics for the committed reachable contract. M21-04 still -owns the production in-memory connection from Rust vector output to the simulator and retained -Feather replay. The latest X7 source also remains fail-closed at its separately recorded dynamic -configuration boundary until the remaining source compiler lowering is complete. Full Native -latest-NFI qualification therefore remains M22-01, not an M21-03 claim. +This proves independent Rust vector semantics for the committed reachable contract. The M21-04 +[in-memory transport](native-in-memory-vector.md) now connects that typed output to the simulator +and proves exact equality with retained Feather replay. The latest X7 source still remains +fail-closed at its separately recorded dynamic configuration boundary until the remaining source +compiler lowering is complete. Full Native latest-NFI qualification therefore remains M22-01, not +an M21-03 or transport-only claim. diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 1fe654a..b42c610 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -144,6 +144,31 @@ dependencies = [ "libc", ] +[[package]] +name = "crossbeam-deque" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5181e0de7b61eb03a81e347d6dd8797bae9da5146707b51077e2d71a54ec0ceb" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61803da095bee82a81bb1a452ecc25d3b2f1416d1897eb86430c6159ef717c17" + [[package]] name = "crypto-common" version = "0.1.7" @@ -404,6 +429,8 @@ version = "1.5.0" dependencies = [ "arrow2", "nfi-sim-core", + "nfi-vector-core", + "rayon", "serde", "serde_json", "sha2", @@ -564,6 +591,26 @@ version = "5.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" +[[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + [[package]] name = "rust_decimal" version = "1.42.1" diff --git a/rust/crates/nfi-vector-io/Cargo.toml b/rust/crates/nfi-vector-io/Cargo.toml index 7893415..71cceac 100644 --- a/rust/crates/nfi-vector-io/Cargo.toml +++ b/rust/crates/nfi-vector-io/Cargo.toml @@ -14,6 +14,8 @@ path = "src/lib.rs" [dependencies] arrow2 = { version = "0.18", features = ["io_ipc", "io_ipc_compression"] } nfi-sim-core = { path = "../nfi-sim-core" } +nfi-vector-core = { path = "../nfi-vector-core" } +rayon = "1" serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.10" diff --git a/rust/crates/nfi-vector-io/src/failures.rs b/rust/crates/nfi-vector-io/src/failures.rs index 9217693..d263567 100644 --- a/rust/crates/nfi-vector-io/src/failures.rs +++ b/rust/crates/nfi-vector-io/src/failures.rs @@ -7,6 +7,8 @@ use thiserror::Error; #[derive(Debug, Error)] pub enum VectorInputError { + #[error("Native vector execution failed: {0}")] + VectorCore(#[from] nfi_vector_core::VectorCoreError), #[error("cannot read vector manifest {path}: {source}")] ReadManifest { path: PathBuf, @@ -97,4 +99,11 @@ pub enum VectorInputError { index: usize, rows: usize, }, + #[error("pair {pair:?} in-memory column {column:?} has type {actual}; expected {expected}")] + InMemoryColumnType { + pair: String, + column: String, + actual: &'static str, + expected: &'static str, + }, } diff --git a/rust/crates/nfi-vector-io/src/in_memory.rs b/rust/crates/nfi-vector-io/src/in_memory.rs new file mode 100644 index 0000000..06e2f72 --- /dev/null +++ b/rust/crates/nfi-vector-io/src/in_memory.rs @@ -0,0 +1,1195 @@ +//! Exact in-memory handoff from generic Rust vector output to the simulator. + +use std::collections::{BTreeMap, BTreeSet}; +use std::mem::size_of; +use std::time::{Duration, Instant}; + +use nfi_sim_core::{ + Candle, EntrySignal, ExitSignal, FeatureColumn, PairSeries, PortfolioConfig, PriceStepChange, + SimulationInput, SIMULATOR_SCHEMA_VERSION, +}; +use nfi_vector_core::column::{OwnedColumn, ValueType}; +use nfi_vector_core::mutation::MutationFrame; +use rayon::prelude::*; +use serde::Serialize; + +use crate::VectorInputError; + +const EMPTY_TAG_TRANSPORT_SENTINEL: &str = "__nfi_bte_empty_tag_column__"; + +/// One pair whose complete typed vector frame already lives in Rust memory. +/// +/// The frame uses the same `nfi_exec_*` columns as the sealed Feather contract. +/// Pair preparation may run in parallel, while the returned `SimulationInput` +/// preserves this vector's order for the single chronological wallet loop. +#[derive(Debug)] +pub struct InMemoryVectorPair { + pub pair: String, + pub execution_start_index: usize, + pub amount_step: Option, + pub price_step: Option, + pub price_steps: Vec, + pub minimum_stake: Option, + pub minimum_amount: Option, + pub minimum_cost: Option, + pub feature_columns: Vec, + pub options: VectorPairOptions, + pub frame: MutationFrame, +} + +/// Compact pair capabilities shared by the direct and Feather semantics. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct VectorPairOptions(u8); + +impl Default for VectorPairOptions { + fn default() -> Self { + Self(Self::USE_EXIT_SIGNAL) + } +} + +impl VectorPairOptions { + const CAN_SHORT: u8 = 1; + const INCLUDE_FUNDING: u8 = 1 << 1; + const USE_EXIT_SIGNAL: u8 = 1 << 2; + const INCLUDE_PREVIOUS_CLOSE: u8 = 1 << 3; + + #[must_use] + pub const fn with_can_short(mut self, enabled: bool) -> Self { + self.set(Self::CAN_SHORT, enabled); + self + } + + #[must_use] + pub const fn with_funding(mut self, enabled: bool) -> Self { + self.set(Self::INCLUDE_FUNDING, enabled); + self + } + + #[must_use] + pub const fn with_exit_signal(mut self, enabled: bool) -> Self { + self.set(Self::USE_EXIT_SIGNAL, enabled); + self + } + + #[must_use] + pub const fn with_previous_close(mut self, enabled: bool) -> Self { + self.set(Self::INCLUDE_PREVIOUS_CLOSE, enabled); + self + } + + const fn set(&mut self, flag: u8, enabled: bool) { + if enabled { + self.0 |= flag; + } else { + self.0 &= !flag; + } + } + + const fn can_short(self) -> bool { + self.0 & Self::CAN_SHORT != 0 + } + + const fn include_funding(self) -> bool { + self.0 & Self::INCLUDE_FUNDING != 0 + } + + const fn use_exit_signal(self) -> bool { + self.0 & Self::USE_EXIT_SIGNAL != 0 + } + + const fn include_previous_close(self) -> bool { + self.0 & Self::INCLUDE_PREVIOUS_CLOSE != 0 + } +} + +/// Wall-clock and retained-memory evidence for the direct transport. +#[derive(Debug, Clone, Serialize, PartialEq, Eq)] +pub struct InMemoryVectorProfile { + pub schema_version: &'static str, + pub vector_execute_ns: u64, + pub pair_prepare_ns: u64, + pub pair_count: usize, + pub row_count: usize, + pub feature_column_count: usize, + pub estimated_source_column_bytes: usize, + pub estimated_simulation_owned_bytes: usize, + pub pair_prepare_worker_limit: usize, +} + +struct PreparedPair { + pair: String, + execution_start_index: usize, + amount_step: Option, + price_step: Option, + price_steps: Vec, + minimum_stake: Option, + minimum_amount: Option, + minimum_cost: Option, + feature_columns: BTreeMap, + candles: Vec, + rows: usize, + features: usize, + source_bytes: usize, + owned_bytes: usize, +} + +enum PreparedFeature { + Numbers(Vec), + Booleans(Vec), +} + +impl PreparedPair { + fn into_pair_series(self) -> PairSeries { + PairSeries { + pair: self.pair, + execution_start_index: self.execution_start_index, + amount_step: self.amount_step, + price_step: self.price_step, + price_steps: self.price_steps, + minimum_stake: self.minimum_stake, + minimum_amount: self.minimum_amount, + minimum_cost: self.minimum_cost, + feature_columns: self + .feature_columns + .into_iter() + .map(|(name, values)| { + let column = match values { + PreparedFeature::Numbers(values) => FeatureColumn::numbers(values), + PreparedFeature::Booleans(values) => FeatureColumn::booleans(values), + }; + (name, column) + }) + .collect(), + candles: self.candles.into(), + } + } +} + +/// Convert typed vector outputs directly into the simulator's owned input. +/// +/// # Errors +/// +/// Returns a missing-column, type, null, duplicate-pair, or range error before +/// the chronological simulator is entered. +pub fn assemble_in_memory_vectors( + config: PortfolioConfig, + pairs: Vec, +) -> Result { + assemble_in_memory_vectors_profiled(config, pairs).map(|(input, _)| input) +} + +/// Convert typed vector outputs and report pair-parallel preparation costs. +/// +/// # Errors +/// +/// Returns the same fail-closed transport errors as +/// [`assemble_in_memory_vectors`]. +pub fn assemble_in_memory_vectors_profiled( + config: PortfolioConfig, + pairs: Vec, +) -> Result<(SimulationInput, InMemoryVectorProfile), VectorInputError> { + if pairs.is_empty() { + return Err(VectorInputError::EmptyPairs); + } + let mut names = BTreeSet::new(); + for pair in &pairs { + if pair.pair.is_empty() || !names.insert(pair.pair.as_str()) { + return Err(VectorInputError::InvalidPair(pair.pair.clone())); + } + validate_feature_names(pair)?; + } + + let started = Instant::now(); + // Indexed parallel iteration keeps the caller's deterministic pair order. + // It ends before `SimulationInput` reaches the single-threaded wallet loop. + let prepared = pairs + .into_par_iter() + .map(prepare_pair) + .collect::>() + .into_iter() + .collect::, _>>()?; + let profile = InMemoryVectorProfile { + schema_version: "1.0.0", + vector_execute_ns: 0, + pair_prepare_ns: duration_ns(started.elapsed()), + pair_count: prepared.len(), + row_count: prepared + .iter() + .map(|pair| pair.rows) + .fold(0, usize::saturating_add), + feature_column_count: prepared + .iter() + .map(|pair| pair.features) + .fold(0, usize::saturating_add), + estimated_source_column_bytes: prepared + .iter() + .map(|pair| pair.source_bytes) + .fold(0, usize::saturating_add), + estimated_simulation_owned_bytes: prepared + .iter() + .map(|pair| pair.owned_bytes) + .fold(0, usize::saturating_add), + pair_prepare_worker_limit: rayon::current_num_threads(), + }; + Ok(( + SimulationInput { + schema_version: SIMULATOR_SCHEMA_VERSION.to_owned(), + config, + pairs: prepared + .into_iter() + .map(PreparedPair::into_pair_series) + .collect(), + }, + profile, + )) +} + +/// Execute independent pair vector DAGs in parallel, then build one ordered +/// simulator input for the chronological wallet loop. +/// +/// The executor receives one caller-defined task and must return the complete +/// typed frame for that pair. Indexed Rayon collection preserves task order; +/// no simulation or wallet mutation occurs inside the executor. +/// +/// # Errors +/// +/// Returns the first vector or transport error without exposing a partial +/// `SimulationInput`. +pub fn execute_in_memory_pair_dag_profiled( + config: PortfolioConfig, + tasks: Vec, + execute: Execute, +) -> Result<(SimulationInput, InMemoryVectorProfile), VectorInputError> +where + Task: Send, + Execute: Fn(Task) -> Result + Send + Sync, +{ + let started = Instant::now(); + let pairs = tasks + .into_par_iter() + .map(execute) + .collect::>() + .into_iter() + .collect::, _>>()?; + let vector_execute_ns = duration_ns(started.elapsed()); + let (input, mut profile) = assemble_in_memory_vectors_profiled(config, pairs)?; + profile.vector_execute_ns = vector_execute_ns; + Ok((input, profile)) +} + +fn prepare_pair(pair: InMemoryVectorPair) -> Result { + let rows = pair.frame.len(); + if pair.execution_start_index >= rows { + return Err(VectorInputError::ExecutionStart { + pair: pair.pair, + index: pair.execution_start_index, + rows, + }); + } + let source_bytes = pair + .frame + .columns() + .values() + .map(OwnedColumn::estimated_bytes) + .fold(0_usize, usize::saturating_add); + let candles = build_candles(&pair)?; + let feature_columns = pair + .feature_columns + .iter() + .map(|name| build_feature(&pair, name)) + .collect::, _>>()?; + let owned_bytes = candles + .len() + .saturating_mul(size_of::()) + .saturating_add(candle_string_bytes(&candles)) + .saturating_add(pair.pair.len()) + .saturating_add( + pair.feature_columns + .iter() + .map(String::len) + .fold(0_usize, usize::saturating_add), + ) + .saturating_add( + feature_columns + .values() + .map(feature_bytes) + .fold(0_usize, usize::saturating_add), + ); + Ok(PreparedPair { + pair: pair.pair, + execution_start_index: pair.execution_start_index, + amount_step: pair.amount_step, + price_step: pair.price_step, + price_steps: pair.price_steps, + minimum_stake: pair.minimum_stake, + minimum_amount: pair.minimum_amount, + minimum_cost: pair.minimum_cost, + feature_columns, + candles, + rows, + features: pair.feature_columns.len(), + source_bytes, + owned_bytes, + }) +} + +fn build_candles(pair: &InMemoryVectorPair) -> Result, VectorInputError> { + let mut candles = Vec::with_capacity(pair.frame.len()); + let mut previous_close = None; + for row in 0..pair.frame.len() { + let close = required_number(pair, "close", row)?; + let entry_tag = optional_text(pair, "nfi_exec_enter_tag", row)?; + let exit_tag = optional_text(pair, "nfi_exec_exit_tag", row)?; + let enter_long = enabled(pair, "nfi_exec_enter_long", row)?; + let enter_short = if pair.options.can_short() { + enabled(pair, "nfi_exec_enter_short", row)? + } else { + false + }; + let exit_long = if pair.options.use_exit_signal() { + enabled(pair, "nfi_exec_exit_long", row)? + } else { + false + }; + let exit_short = if pair.options.can_short() && pair.options.use_exit_signal() { + enabled(pair, "nfi_exec_exit_short", row)? + } else { + false + }; + let funding_rate = if pair.options.include_funding() { + optional_number(pair, "nfi_exec_funding_rate", row)? + } else { + None + }; + let funding_mark_price = if pair.options.include_funding() { + optional_number(pair, "nfi_exec_funding_mark_price", row)? + } else { + None + }; + let exit_reason = || exit_tag.clone().unwrap_or_else(|| "exit_signal".to_owned()); + candles.push(Candle { + timestamp_ms: required_timestamp(pair, "date", row)?, + open: required_number(pair, "open", row)?, + high: required_number(pair, "high", row)?, + low: required_number(pair, "low", row)?, + close, + volume: required_number(pair, "volume", row)?, + previous_close: pair + .options + .include_previous_close() + .then_some(previous_close) + .flatten(), + enter_long: enter_long.then(|| EntrySignal { + tag: entry_tag.clone(), + leverage: None, + liquidation_price: None, + }), + enter_short: enter_short.then_some(EntrySignal { + tag: entry_tag, + leverage: None, + liquidation_price: None, + }), + exit_long: exit_long.then(|| ExitSignal { + reason: exit_reason(), + }), + exit_short: exit_short.then(|| ExitSignal { + reason: exit_reason(), + }), + funding_rate, + funding_mark_price, + adjustment: None, + }); + previous_close = Some(close); + } + Ok(candles) +} + +fn validate_feature_names(pair: &InMemoryVectorPair) -> Result<(), VectorInputError> { + let mut names = BTreeSet::new(); + for column in &pair.feature_columns { + if column.is_empty() || !names.insert(column.as_str()) { + return Err(VectorInputError::InvalidFeatureName { + pair: pair.pair.clone(), + column: column.clone(), + }); + } + } + Ok(()) +} + +fn column<'a>( + pair: &'a InMemoryVectorPair, + name: &str, +) -> Result<&'a OwnedColumn, VectorInputError> { + pair.frame + .column(name) + .ok_or_else(|| VectorInputError::MissingColumn { + pair: pair.pair.clone(), + column: name.to_owned(), + }) +} + +fn required_timestamp( + pair: &InMemoryVectorPair, + name: &str, + row: usize, +) -> Result { + let column = column(pair, name)?; + if column.as_view().value_type() != ValueType::TimestampMs { + return Err(type_error(pair, name, column, "Timestamp(Millisecond)")); + } + column + .as_view() + .timestamp_ms_at(row) + .ok_or_else(|| null_error(pair, name, row)) +} + +#[allow(clippy::cast_precision_loss)] +// Feather converts Arrow Int64 through Python-compatible float semantics too. +fn required_number( + pair: &InMemoryVectorPair, + name: &str, + row: usize, +) -> Result { + let column = column(pair, name)?; + match column.as_view().value_type() { + ValueType::F64 => column + .as_view() + .f64_at(row) + .ok_or_else(|| null_error(pair, name, row)), + ValueType::I64 => column + .as_view() + .i64_at(row) + .map(|value| value as f64) + .ok_or_else(|| null_error(pair, name, row)), + _ => Err(type_error(pair, name, column, "numeric")), + } +} + +fn optional_number( + pair: &InMemoryVectorPair, + name: &str, + row: usize, +) -> Result, VectorInputError> { + let column = column(pair, name)?; + let value = match column.as_view().value_type() { + ValueType::F64 => column.as_view().f64_at(row), + ValueType::I64 => column.as_view().i64_at(row).map(integer_as_number), + _ => return Err(type_error(pair, name, column, "numeric")), + }; + Ok(value.filter(|value| !value.is_nan())) +} + +fn enabled(pair: &InMemoryVectorPair, name: &str, row: usize) -> Result { + let column = column(pair, name)?; + Ok(match column.as_view().value_type() { + ValueType::F64 => column.as_view().f64_at(row) == Some(1.0), + ValueType::I64 => column.as_view().i64_at(row) == Some(1), + ValueType::Bool => column.as_view().bool_at(row) == Some(true), + _ => return Err(type_error(pair, name, column, "numeric or Boolean")), + }) +} + +fn optional_text( + pair: &InMemoryVectorPair, + name: &str, + row: usize, +) -> Result, VectorInputError> { + let Some(column) = pair.frame.column(name) else { + return Ok(None); + }; + if column.as_view().value_type() != ValueType::Text { + return Err(type_error(pair, name, column, "UTF-8 string")); + } + Ok(column.as_view().text_at(row).and_then(|value| { + (!value.is_empty() && value != EMPTY_TAG_TRANSPORT_SENTINEL).then(|| value.to_owned()) + })) +} + +fn build_feature( + pair: &InMemoryVectorPair, + name: &str, +) -> Result<(String, PreparedFeature), VectorInputError> { + let column = column(pair, name)?; + let values = match column.as_view().value_type() { + ValueType::F64 => PreparedFeature::Numbers( + (0..column.len()) + .map(|row| column.as_view().f64_at(row).unwrap_or(f64::NAN)) + .collect(), + ), + ValueType::I64 => PreparedFeature::Numbers( + (0..column.len()) + .map(|row| { + column + .as_view() + .i64_at(row) + .map_or(f64::NAN, integer_as_number) + }) + .collect(), + ), + ValueType::Bool => PreparedFeature::Booleans( + (0..column.len()) + .map(|row| { + column + .as_view() + .bool_at(row) + .ok_or_else(|| null_error(pair, name, row)) + }) + .collect::, _>>()?, + ), + _ => return Err(type_error(pair, name, column, "numeric or Boolean")), + }; + Ok((name.to_owned(), values)) +} + +fn feature_bytes(column: &PreparedFeature) -> usize { + match column { + PreparedFeature::Numbers(values) => values.len().saturating_mul(size_of::()), + PreparedFeature::Booleans(values) => values.len().saturating_mul(size_of::()), + } +} + +fn candle_string_bytes(candles: &[Candle]) -> usize { + candles + .iter() + .map(|candle| { + candle + .enter_long + .as_ref() + .and_then(|signal| signal.tag.as_ref()) + .map_or(0, String::len) + .saturating_add( + candle + .enter_short + .as_ref() + .and_then(|signal| signal.tag.as_ref()) + .map_or(0, String::len), + ) + .saturating_add( + candle + .exit_long + .as_ref() + .map_or(0, |signal| signal.reason.len()), + ) + .saturating_add( + candle + .exit_short + .as_ref() + .map_or(0, |signal| signal.reason.len()), + ) + }) + .fold(0_usize, usize::saturating_add) +} + +fn type_error( + pair: &InMemoryVectorPair, + name: &str, + column: &OwnedColumn, + expected: &'static str, +) -> VectorInputError { + VectorInputError::InMemoryColumnType { + pair: pair.pair.clone(), + column: name.to_owned(), + actual: column.as_view().value_type().label(), + expected, + } +} + +fn null_error(pair: &InMemoryVectorPair, name: &str, row: usize) -> VectorInputError { + VectorInputError::NullValue { + pair: pair.pair.clone(), + column: name.to_owned(), + row, + } +} + +fn duration_ns(duration: Duration) -> u64 { + u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX) +} + +#[allow(clippy::cast_precision_loss)] +// Matches the existing Feather/Python integer-to-number boundary exactly. +fn integer_as_number(value: i64) -> f64 { + value as f64 +} + +#[cfg(test)] +mod tests { + use std::fs::File; + + use arrow2::array::Array; + use arrow2::chunk::Chunk; + use arrow2::datatypes::{DataType, Field, Schema, TimeUnit}; + use arrow2::io::ipc::write::{FileWriter, WriteOptions}; + use nfi_sim_core::{simulate, simulate_with_observer, SimulationEvent}; + use nfi_vector_core::mutation::{ + materialize_execution_signals, MutationEngine, MutationProgram, + }; + use serde_json::{json, Value}; + use sha2::{Digest, Sha256}; + + use super::*; + use crate::load_vector_manifest_profiled; + + const TAG_PROGRAM: &str = + include_str!("../../../../benchmarks/reference/vector-shadow/tag-program.json"); + + #[test] + fn direct_transport_is_full_state_exact_to_sealed_feather_replay() { + let temporary = tempfile::tempdir().expect("temporary fixture directory"); + let vector = temporary.path().join("pair.feather"); + let frame = fixture_frame(); + write_feather(&vector, &frame); + let encoded = std::fs::read(&vector).expect("fixture vector bytes"); + let vector_sha = format!("{:x}", Sha256::digest(encoded)); + let config = config_document(); + let manifest = json!({ + "schema_version": crate::VECTOR_MANIFEST_SCHEMA_VERSION, + "config": config, + "pairs": [{ + "pair": "AAA/USDT", + "execution_start_index": 1, + "amount_step": null, + "price_step": null, + "price_steps": [], + "minimum_stake": null, + "minimum_amount": null, + "minimum_cost": null, + "vector": { + "path": "pair.feather", + "sha256": vector_sha, + "rows": frame.len(), + "format": "feather-ipc" + }, + "feature_columns": ["score"], + "can_short": false, + "include_funding": false, + "use_exit_signal": true, + "include_previous_close": true + }] + }); + let manifest_path = temporary.path().join("manifest.json"); + std::fs::write( + &manifest_path, + serde_json::to_vec(&manifest).expect("manifest serializes"), + ) + .expect("manifest writes"); + + let (feather, feather_profile) = + load_vector_manifest_profiled(&manifest_path).expect("sealed replay loads"); + let (direct, direct_profile) = assemble_in_memory_vectors_profiled( + serde_json::from_value(config).expect("portfolio config"), + vec![fixture_pair(frame)], + ) + .expect("direct transport loads"); + let (feather_result, feather_events) = simulate_with_events(&feather); + let (direct_result, direct_events) = simulate_with_events(&direct); + + assert_eq!(direct_result, feather_result); + assert_eq!(direct_events, feather_events); + assert!(matches!( + direct.pairs[0].candles, + nfi_sim_core::CandleSeries::Owned(_) + )); + assert!(matches!( + feather.pairs[0].candles, + nfi_sim_core::CandleSeries::FileBacked(_) + )); + println!( + "M21_TRANSPORT_PROFILE={}", + serde_json::to_string(&json!({ + "schema_version": "1.0.0", + "scope": "four-row exact transport diagnostic; not a release speed claim", + "direct": direct_profile, + "feather": feather_profile, + "trade_surface_exact": true, + "full_state_exact": true + })) + .expect("profile serializes") + ); + } + + #[test] + fn pair_parallel_preparation_preserves_input_order_and_reports_memory() { + let config = serde_json::from_value(config_document()).expect("portfolio config"); + let pairs = ["CCC/USDT", "AAA/USDT", "BBB/USDT"] + .into_iter() + .map(|name| { + let mut pair = fixture_pair(fixture_frame()); + pair.pair = name.to_owned(); + pair + }) + .collect(); + let (input, profile) = + assemble_in_memory_vectors_profiled(config, pairs).expect("parallel pair prepare"); + + assert_eq!( + input + .pairs + .iter() + .map(|pair| pair.pair.as_str()) + .collect::>(), + vec!["CCC/USDT", "AAA/USDT", "BBB/USDT"] + ); + assert_eq!(profile.pair_count, 3); + assert_eq!(profile.row_count, 12); + assert_eq!(profile.feature_column_count, 3); + assert!(profile.estimated_source_column_bytes > 0); + assert!(profile.estimated_simulation_owned_bytes > 0); + assert!(profile.pair_prepare_worker_limit > 0); + } + + #[test] + fn generic_rust_vector_dag_feeds_the_simulator_without_feather() { + let config = serde_json::from_value(config_document()).expect("portfolio config"); + let program = MutationProgram::from_json(TAG_PROGRAM).expect("sealed Tag program"); + let tasks = vec!["AAA/USDT", "BBB/USDT"]; + let (input, profile) = execute_in_memory_pair_dag_profiled(config, tasks, |pair| { + let source = tag_source_frame(); + let mutated = MutationEngine::new(&program)?.execute(source)?; + let execution = materialize_execution_signals(&mutated, 1, 1)?; + let mut columns = execution.frame.columns().clone(); + columns.extend(market_columns(8)); + columns.insert( + "score".to_owned(), + mutated.column("score").expect("source score").clone(), + ); + Ok(InMemoryVectorPair { + pair: pair.to_owned(), + execution_start_index: 1, + amount_step: None, + price_step: None, + price_steps: Vec::new(), + minimum_stake: None, + minimum_amount: None, + minimum_cost: None, + feature_columns: vec!["score".to_owned()], + options: VectorPairOptions::default() + .with_exit_signal(true) + .with_previous_close(true), + frame: MutationFrame::new(columns)?, + }) + }) + .expect("pair DAG executes"); + + assert_eq!(profile.pair_count, 2); + assert!(profile.vector_execute_ns > 0); + assert_eq!(input.pairs[0].pair, "AAA/USDT"); + assert_eq!(input.pairs[1].pair, "BBB/USDT"); + assert!(input.pairs.iter().all(|pair| { + pair.candles + .iter() + .skip(pair.execution_start_index) + .any(|candle| candle.enter_long.is_some()) + })); + let mut events = Vec::new(); + simulate_with_observer(&input, |event| events.push(event.clone())) + .expect("DAG-fed simulation"); + assert!(!events.is_empty()); + } + + #[test] + #[ignore = "explicit release-mode transport diagnostic"] + fn large_pair_transport_profile_is_reported_without_a_speed_assertion() { + const ROWS_PER_PAIR: usize = 25_000; + let temporary = tempfile::tempdir().expect("temporary diagnostic directory"); + let frame = diagnostic_frame(ROWS_PER_PAIR); + let names = ["AAA/USDT", "BBB/USDT", "CCC/USDT", "DDD/USDT"]; + let config = config_document(); + let mut manifest_pairs = Vec::new(); + for (index, name) in names.iter().enumerate() { + let file_name = format!("pair-{index}.feather"); + let path = temporary.path().join(&file_name); + write_feather(&path, &frame); + let vector_sha = format!( + "{:x}", + Sha256::digest(std::fs::read(&path).expect("diagnostic vector bytes")) + ); + manifest_pairs.push(json!({ + "pair": name, + "execution_start_index": 1, + "amount_step": null, + "price_step": null, + "price_steps": [], + "minimum_stake": null, + "minimum_amount": null, + "minimum_cost": null, + "vector": { + "path": file_name, + "sha256": vector_sha, + "rows": ROWS_PER_PAIR, + "format": "feather-ipc" + }, + "feature_columns": ["score"], + "can_short": false, + "include_funding": false, + "use_exit_signal": true, + "include_previous_close": true + })); + } + let manifest_path = temporary.path().join("manifest.json"); + std::fs::write( + &manifest_path, + serde_json::to_vec(&json!({ + "schema_version": crate::VECTOR_MANIFEST_SCHEMA_VERSION, + "config": config, + "pairs": manifest_pairs + })) + .expect("diagnostic manifest serializes"), + ) + .expect("diagnostic manifest writes"); + + let (feather, feather_profile) = + load_vector_manifest_profiled(&manifest_path).expect("diagnostic Feather replay"); + let tasks = names + .into_iter() + .map(|name| (name.to_owned(), frame.clone())) + .collect(); + let (direct, direct_profile) = execute_in_memory_pair_dag_profiled( + serde_json::from_value(config).expect("portfolio config"), + tasks, + |(name, frame)| { + let mut pair = fixture_pair(frame); + pair.pair = name; + Ok(pair) + }, + ) + .expect("diagnostic direct transport"); + assert_eq!(simulate(&direct).unwrap(), simulate(&feather).unwrap()); + + let direct_ns = direct_profile + .vector_execute_ns + .saturating_add(direct_profile.pair_prepare_ns); + let feather_ns = feather_profile + .manifest_ns + .saturating_add(feather_profile.vector_hash_ns) + .saturating_add(feather_profile.feather_decode_ns); + println!( + "M21_LARGE_TRANSPORT_PROFILE={}", + serde_json::to_string(&json!({ + "schema_version": "1.0.0", + "scope": "four-pair 100000-row local transport diagnostic; not release-grade", + "direct": direct_profile, + "direct_transport_ns": direct_ns, + "feather": feather_profile, + "feather_transport_ns": feather_ns, + "observed_transport_speedup": duration_ratio(feather_ns, direct_ns), + "trade_surface_exact": true + })) + .expect("diagnostic profile serializes") + ); + } + + #[test] + fn invalid_type_duplicate_pair_and_numeric_truthiness_fail_closed() { + let config = serde_json::from_value(config_document()).expect("portfolio config"); + let mut invalid = fixture_pair(fixture_frame()); + let mut invalid_columns = invalid.frame.columns().clone(); + invalid_columns.insert( + "date".to_owned(), + OwnedColumn::text(vec![Some("not a timestamp".to_owned()); 4]), + ); + invalid.frame = MutationFrame::new(invalid_columns).expect("consistent invalid frame"); + assert!(matches!( + assemble_in_memory_vectors( + serde_json::from_value(config_document()).expect("portfolio config"), + vec![invalid] + ), + Err(VectorInputError::InMemoryColumnType { .. }) + )); + + let duplicate = vec![fixture_pair(fixture_frame()), fixture_pair(fixture_frame())]; + assert!(matches!( + assemble_in_memory_vectors(config, duplicate), + Err(VectorInputError::InvalidPair(_)) + )); + + let original = fixture_frame(); + let mut columns = original.columns().clone(); + columns.insert( + "nfi_exec_enter_long".to_owned(), + OwnedColumn::i64(vec![Some(0), Some(2), Some(-1), Some(0)]), + ); + let frame = MutationFrame::new(columns).expect("replacement signal"); + let direct = assemble_in_memory_vectors( + serde_json::from_value(config_document()).expect("portfolio config"), + vec![fixture_pair(frame)], + ) + .expect("numeric signal frame"); + assert!(direct.pairs[0] + .candles + .iter() + .all(|candle| candle.enter_long.is_none())); + } + + fn fixture_pair(frame: MutationFrame) -> InMemoryVectorPair { + InMemoryVectorPair { + pair: "AAA/USDT".to_owned(), + execution_start_index: 1, + amount_step: None, + price_step: None, + price_steps: Vec::new(), + minimum_stake: None, + minimum_amount: None, + minimum_cost: None, + feature_columns: vec!["score".to_owned()], + options: VectorPairOptions::default() + .with_exit_signal(true) + .with_previous_close(true), + frame, + } + } + + fn fixture_frame() -> MutationFrame { + MutationFrame::new(BTreeMap::from([ + ( + "date".to_owned(), + OwnedColumn::timestamp_ms(vec![ + Some(60_000), + Some(120_000), + Some(180_000), + Some(240_000), + ]), + ), + ( + "open".to_owned(), + OwnedColumn::f64(vec![Some(100.0), Some(101.0), Some(103.0), Some(104.0)]), + ), + ( + "high".to_owned(), + OwnedColumn::f64(vec![Some(101.0), Some(102.0), Some(104.0), Some(105.0)]), + ), + ( + "low".to_owned(), + OwnedColumn::f64(vec![Some(99.0), Some(100.0), Some(102.0), Some(103.0)]), + ), + ( + "close".to_owned(), + OwnedColumn::f64(vec![Some(100.0), Some(101.0), Some(103.0), Some(104.0)]), + ), + ( + "volume".to_owned(), + OwnedColumn::f64(vec![Some(10.0), Some(11.0), Some(12.0), Some(13.0)]), + ), + ( + "score".to_owned(), + OwnedColumn::f64(vec![None, Some(-0.0), Some(f64::NAN), Some(3.0)]), + ), + ( + "nfi_exec_enter_long".to_owned(), + OwnedColumn::i64(vec![Some(0), Some(1), Some(0), Some(0)]), + ), + ( + "nfi_exec_exit_long".to_owned(), + OwnedColumn::i64(vec![Some(0), Some(0), Some(1), Some(0)]), + ), + ( + "nfi_exec_enter_tag".to_owned(), + OwnedColumn::text(vec![None, Some("101 ".to_owned()), None, None]), + ), + ( + "nfi_exec_exit_tag".to_owned(), + OwnedColumn::text(vec![None, None, Some("profit signal ".to_owned()), None]), + ), + ])) + .expect("fixture frame") + } + + fn tag_source_frame() -> MutationFrame { + MutationFrame::new(BTreeMap::from([ + ( + "score".to_owned(), + OwnedColumn::f64(vec![ + Some(-2.0), + Some(-0.5), + Some(0.0), + Some(0.5), + Some(1.5), + Some(2.0), + Some(2.5), + Some(f64::NAN), + ]), + ), + ( + "exit_mask".to_owned(), + OwnedColumn::boolean(vec![ + Some(false), + Some(true), + Some(false), + Some(true), + Some(false), + Some(true), + None, + Some(false), + ]), + ), + ( + "enter_tag".to_owned(), + OwnedColumn::text(vec![Some("stale-entry".to_owned()); 8]), + ), + ( + "exit_tag".to_owned(), + OwnedColumn::text(vec![Some("stale-exit".to_owned()); 8]), + ), + ])) + .expect("Tag source frame") + } + + fn diagnostic_frame(rows: usize) -> MutationFrame { + let indexes = 0..rows; + MutationFrame::new(BTreeMap::from([ + ( + "date".to_owned(), + OwnedColumn::timestamp_ms( + indexes + .clone() + .map(|row| { + Some(60_000_i64 * (i64::try_from(row).expect("diagnostic row") + 1)) + }) + .collect(), + ), + ), + ("open".to_owned(), OwnedColumn::f64(vec![Some(100.0); rows])), + ("high".to_owned(), OwnedColumn::f64(vec![Some(101.0); rows])), + ("low".to_owned(), OwnedColumn::f64(vec![Some(99.0); rows])), + ( + "close".to_owned(), + OwnedColumn::f64(vec![Some(100.0); rows]), + ), + ( + "volume".to_owned(), + OwnedColumn::f64(vec![Some(10.0); rows]), + ), + ( + "score".to_owned(), + OwnedColumn::f64( + indexes + .map(|row| { + Some(f64::from(u32::try_from(row % 101).expect("diagnostic row"))) + }) + .collect(), + ), + ), + ( + "nfi_exec_enter_long".to_owned(), + OwnedColumn::i64(vec![Some(0); rows]), + ), + ( + "nfi_exec_exit_long".to_owned(), + OwnedColumn::i64(vec![Some(0); rows]), + ), + ( + "nfi_exec_enter_tag".to_owned(), + OwnedColumn::text(vec![None; rows]), + ), + ( + "nfi_exec_exit_tag".to_owned(), + OwnedColumn::text(vec![None; rows]), + ), + ])) + .expect("diagnostic frame") + } + + fn market_columns(rows: usize) -> BTreeMap { + let timestamps = (0..rows) + .map(|row| Some(60_000_i64 * (i64::try_from(row).expect("small row") + 1))) + .collect(); + let prices = (0..rows) + .map(|row| Some(100.0 + f64::from(u32::try_from(row).expect("small row")))) + .collect::>(); + BTreeMap::from([ + ("date".to_owned(), OwnedColumn::timestamp_ms(timestamps)), + ("open".to_owned(), OwnedColumn::f64(prices.clone())), + ( + "high".to_owned(), + OwnedColumn::f64( + prices + .iter() + .map(|value| value.map(|item| item + 1.0)) + .collect(), + ), + ), + ( + "low".to_owned(), + OwnedColumn::f64( + prices + .iter() + .map(|value| value.map(|item| item - 1.0)) + .collect(), + ), + ), + ("close".to_owned(), OwnedColumn::f64(prices)), + ( + "volume".to_owned(), + OwnedColumn::f64(vec![Some(10.0); rows]), + ), + ]) + } + + fn config_document() -> Value { + json!({ + "starting_balance": 1_000.0, + "max_open_trades": 1, + "stake_amount": 100.0, + "fee_rate": 0.0, + "stoploss_ratio": -0.5, + "amount_step": 0.00001, + "price_step": 0.01 + }) + } + + fn write_feather(path: &std::path::Path, frame: &MutationFrame) { + let fields = frame + .columns() + .iter() + .map(|(name, column)| Field::new(name, data_type(column), true)) + .collect::>(); + let arrays = frame + .columns() + .values() + .map(|column| match column { + OwnedColumn::F64(values) => Box::new(values.clone()) as Box, + OwnedColumn::I64(values) => Box::new(values.clone()) as Box, + OwnedColumn::Bool(values) => Box::new(values.clone()) as Box, + OwnedColumn::Text(values) => Box::new(values.clone()) as Box, + OwnedColumn::TimestampMs(values) => Box::new(values.clone()) as Box, + }) + .collect(); + let schema = Schema::from(fields); + let mut writer = FileWriter::try_new( + File::create(path).expect("create fixture vector"), + schema, + None, + WriteOptions { compression: None }, + ) + .expect("create Feather writer"); + writer + .write(&Chunk::new(arrays), None) + .expect("write Feather batch"); + writer.finish().expect("finish Feather vector"); + } + + fn data_type(column: &OwnedColumn) -> DataType { + match column.as_view().value_type() { + ValueType::F64 => DataType::Float64, + ValueType::I64 => DataType::Int64, + ValueType::Bool => DataType::Boolean, + ValueType::Text => DataType::Utf8, + ValueType::TimestampMs => DataType::Timestamp(TimeUnit::Millisecond, None), + } + } + + fn simulate_with_events( + input: &SimulationInput, + ) -> (nfi_sim_core::SimulationResult, Vec) { + let mut events = Vec::new(); + let result = simulate_with_observer(input, |event| events.push(event.clone())) + .expect("simulation succeeds"); + (result, events) + } + + #[allow(clippy::cast_precision_loss)] + fn duration_ratio(numerator: u64, denominator: u64) -> f64 { + numerator as f64 / denominator as f64 + } +} diff --git a/rust/crates/nfi-vector-io/src/lib.rs b/rust/crates/nfi-vector-io/src/lib.rs index ca6c200..147db4a 100644 --- a/rust/crates/nfi-vector-io/src/lib.rs +++ b/rust/crates/nfi-vector-io/src/lib.rs @@ -7,12 +7,19 @@ mod decode; mod failures; +#[allow(clippy::module_name_repetitions)] // Public API distinguishes direct and Feather profiles. +mod in_memory; mod loader; mod row; mod schema; mod values; pub use failures::VectorInputError; +pub use in_memory::{ + assemble_in_memory_vectors, assemble_in_memory_vectors_profiled, + execute_in_memory_pair_dag_profiled, InMemoryVectorPair, InMemoryVectorProfile, + VectorPairOptions, +}; pub use loader::{load_vector_manifest, load_vector_manifest_profiled, VectorLoadProfile}; /// Version of the compact manifest consumed by this crate. diff --git a/rust/crates/nfi-vector-io/src/schema.rs b/rust/crates/nfi-vector-io/src/schema.rs index e2025fa..0516d68 100644 --- a/rust/crates/nfi-vector-io/src/schema.rs +++ b/rust/crates/nfi-vector-io/src/schema.rs @@ -114,11 +114,15 @@ pub(crate) fn required_columns(pair: &VectorPair) -> BTreeSet { "close".to_owned(), "volume".to_owned(), "nfi_exec_enter_long".to_owned(), - "nfi_exec_exit_long".to_owned(), ]); + if pair.use_exit_signal.enabled() { + columns.insert("nfi_exec_exit_long".to_owned()); + } if pair.can_short.enabled() { columns.insert("nfi_exec_enter_short".to_owned()); - columns.insert("nfi_exec_exit_short".to_owned()); + if pair.use_exit_signal.enabled() { + columns.insert("nfi_exec_exit_short".to_owned()); + } } if pair.include_funding.enabled() { columns.insert("nfi_exec_funding_rate".to_owned()); diff --git a/rust/crates/nfi-vector-io/src/values.rs b/rust/crates/nfi-vector-io/src/values.rs index 359b6bb..61e130d 100644 --- a/rust/crates/nfi-vector-io/src/values.rs +++ b/rust/crates/nfi-vector-io/src/values.rs @@ -2,7 +2,7 @@ use std::collections::BTreeMap; -use arrow2::array::{Array, PrimitiveArray, Utf8Array}; +use arrow2::array::{Array, BooleanArray, PrimitiveArray, Utf8Array}; use arrow2::chunk::Chunk; use arrow2::datatypes::{DataType, TimeUnit}; @@ -144,7 +144,8 @@ pub(crate) fn is_numeric_type(data_type: &DataType) -> bool { } #[allow(clippy::float_cmp)] -// Freqtrade intentionally activates a signal only for exact numeric 1. An +// Freqtrade intentionally activates a signal only for exact numeric 1. Python +// Boolean `True` is equal to 1, while arbitrary truthy numbers are not. An // epsilon comparison would incorrectly turn nearby strategy outputs into orders. pub(crate) fn enabled( array: &dyn Array, @@ -156,6 +157,13 @@ pub(crate) fn enabled( if array.is_null(row) { return Ok(false); } + if array.data_type() == &DataType::Boolean { + return Ok(array + .as_any() + .downcast_ref::() + .expect("Boolean physical type matches Arrow data type") + .value(row)); + } let value = required_number(array, row, pair, column, absolute_row)?; Ok(value == 1.0) } @@ -208,7 +216,7 @@ pub(crate) fn optional_text( #[cfg(test)] mod tests { - use arrow2::array::PrimitiveArray; + use arrow2::array::{BooleanArray, PrimitiveArray}; use super::enabled; @@ -230,5 +238,10 @@ mod tests { expected ); } + + let booleans = BooleanArray::from([Some(true), Some(false), None]); + assert!(enabled(&booleans, 0, "ETH/USDT", "enter_long", 0).unwrap()); + assert!(!enabled(&booleans, 1, "ETH/USDT", "enter_long", 1).unwrap()); + assert!(!enabled(&booleans, 2, "ETH/USDT", "enter_long", 2).unwrap()); } } From 6ba3fc2db9da809d61ea7b1b9349c4810f2c2a4a Mon Sep 17 00:00:00 2001 From: vntrevx <20063774+vntrevx@users.noreply.github.com> Date: Tue, 11 Aug 2026 22:12:33 +0900 Subject: [PATCH 3/3] chore(roadmap): complete M21-04 --- planning/roadmap-state.json | 52 ++++++++++++++++++++++++++++++++----- 1 file changed, 45 insertions(+), 7 deletions(-) diff --git a/planning/roadmap-state.json b/planning/roadmap-state.json index abe3644..3c1d2e7 100644 --- a/planning/roadmap-state.json +++ b/planning/roadmap-state.json @@ -1,10 +1,10 @@ { "schema_version": "1.0.0", "roadmap_id": "nfi-backtest-engine-post-v1.1.0", - "revision": 141, - "updated_at": "2026-08-11T21:20:59+09:00", + "revision": 142, + "updated_at": "2026-08-11T22:09:26+09:00", "acceptance_commands": "planning/acceptance-commands.json", - "active_task_id": "M21-04", + "active_task_id": null, "execution_policy": { "max_in_progress": 1, "selection": "lowest order pending task whose dependencies are completed", @@ -4917,7 +4917,7 @@ "order": 2104, "milestone": "M21", "title": "Connect the in-memory Rust vector and simulation pipeline", - "status": "in_progress", + "status": "completed", "depends_on": [ "M21-03" ], @@ -4937,9 +4937,16 @@ "transport parity and performance report" ], "started_at": "2026-08-11T21:20:59+09:00", - "completed_at": null, - "commit_sha": null, - "evidence": [], + "completed_at": "2026-08-11T22:09:26+09:00", + "commit_sha": "d9529131eaa629cf598a97569931eed844e7066a", + "evidence": [ + ".nfi/roadmap-acceptance/M21-04/d9529131eaa629cf598a97569931eed844e7066a/acceptance-report.json", + ".nfi/roadmap-acceptance/M21-04/d9529131eaa629cf598a97569931eed844e7066a/SHA256SUMS.txt", + ".nfi/roadmap-acceptance/M21-04/d9529131eaa629cf598a97569931eed844e7066a/x7-spot/run.json", + ".nfi/roadmap-acceptance/M21-04/d9529131eaa629cf598a97569931eed844e7066a/x7-futures/run.json", + "benchmarks/evidence/native-in-memory-transport-m21-04-2026-08-11.json", + "docs/native-in-memory-vector.md" + ], "blocker": null }, { @@ -8607,6 +8614,37 @@ "transport_performance_evidence_required": true, "runtime_hardcoding_allowed": false } + }, + { + "sequence": 189, + "timestamp": "2026-08-11T22:09:26+09:00", + "task_id": "M21-04", + "event": "task_completed", + "details": { + "implementation_commit": "d9529131eaa629cf598a97569931eed844e7066a", + "in_memory_simulator_connection_complete": true, + "sealed_feather_replay_retained": true, + "pair_dag_parallel": true, + "pair_order_preserved": true, + "parallel_wallet_mutation": false, + "transport_trade_surface_exact": true, + "transport_full_state_exact": true, + "transport_diagnostic_rows": 100000, + "transport_diagnostic_pairs": 4, + "observed_transport_speedup": 2.036178651616219, + "performance_claim": "single-host synthetic transport diagnostic only", + "python_test_count": 804, + "future_compatibility_test_count": 143, + "rust_workspace_default_test_count": 224, + "spot_trade_surface_and_full_state_exact": true, + "futures_trade_surface_and_full_state_exact": true, + "runtime_hardcoding_added": false, + "latest_x7_fully_compiled": false, + "python_strategy_execution_removed_from_native": false, + "full_native_strategy_claim": false, + "next_eligible_task": "M22-01", + "next_task_manual_gate": false + } } ] }