From 2bc033b8cc442408cee2e6a9f4949e7ced781fbf Mon Sep 17 00:00:00 2001 From: boris Date: Sat, 3 Oct 2026 12:29:11 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E9=80=90=E5=86=B3=E7=AD=96?= =?UTF-8?q?=E5=8A=A8=E6=80=81=E6=A8=A1=E5=9E=8B=E6=8A=95=E5=BD=B1=E5=B9=B6?= =?UTF-8?q?=E9=9A=94=E7=A6=BB=E5=85=B1=E4=BA=AB=E6=95=B0=E6=8D=AE=E4=B8=8E?= =?UTF-8?q?=E9=A6=96=E6=97=A5=E9=A2=84=E7=83=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/fidc-core/src/data.rs | 81 +++++- crates/fidc-core/src/decision_projection.rs | 272 +++++++++++++++++++ crates/fidc-core/src/engine.rs | 282 +++++++++++++++++++- crates/fidc-core/src/lib.rs | 6 + 4 files changed, 622 insertions(+), 19 deletions(-) create mode 100644 crates/fidc-core/src/decision_projection.rs diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index a21b7dc..11a220a 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -1425,6 +1425,7 @@ pub struct DataSet { factor_by_date: Arc>>, factor_symbol_ids_by_date: Arc>>, factor_row_positions_by_date: Arc>, + decision_projection: Option, factor_market_cap_order_by_date: Arc>>, factor_text_by_date: Arc>>, factor_text_index: Arc>, @@ -1451,6 +1452,13 @@ pub struct DataSet { completed_minute_bars: Arc>>, } +#[derive(Debug, Clone)] +struct DecisionFactorProjection { + date: NaiveDate, + rows: Vec, + symbol_ids: Vec, +} + struct DailySymbolRows<'a, T> { rows: &'a [T], symbol_ids: &'a [u32], @@ -1928,6 +1936,7 @@ impl DataSet { factor_by_date: Arc::new(factor_by_date), factor_symbol_ids_by_date: Arc::new(factor_symbol_ids_by_date), factor_row_positions_by_date: Arc::new(factor_row_positions_by_date), + decision_projection: None, factor_market_cap_order_by_date: Arc::new(factor_market_cap_order_by_date), factor_text_by_date: Arc::new(factor_text_by_date), factor_text_index: Arc::new(factor_text_index), @@ -2058,6 +2067,7 @@ impl DataSet { } } + let projected_factors = self.decision_projection.as_ref().filter(|item| item.date == date); DailySnapshotView { market: rows_on( date, @@ -2065,12 +2075,12 @@ impl DataSet { &self.market_symbol_ids_by_date, &self.market_row_positions_by_date, ), - factors: rows_on( - date, - &self.factor_by_date, - &self.factor_symbol_ids_by_date, + factors: projected_factors.map(|item| DailySymbolRows { + rows: &item.rows, symbol_ids: &item.symbol_ids, row_positions: None, + }).unwrap_or_else(|| rows_on( + date, &self.factor_by_date, &self.factor_symbol_ids_by_date, &self.factor_row_positions_by_date, - ), + )), candidates: rows_on( date, &self.candidate_by_date, @@ -2170,6 +2180,11 @@ impl DataSet { date: NaiveDate, symbol_id: u32, ) -> Option<&DailyFactorSnapshot> { + if let Some(item) = self.decision_projection.as_ref().filter(|item| item.date == date) + && let Some(row) = find_by_symbol_id(&item.rows, &item.symbol_ids, symbol_id) + { + return Some(row); + } let rows = self.factor_by_date.get(&date)?; if let Some(index) = dense_row_position(&self.factor_row_positions_by_date, date, symbol_id) { @@ -2178,6 +2193,45 @@ impl DataSet { find_by_symbol_id(rows, self.factor_symbol_ids_by_date.get(&date)?, symbol_id) } + pub(crate) fn clear_decision_projection(&mut self) { + self.decision_projection = None; + } + + pub(crate) fn require_unmaterialized_projection_fields( + &self, fields: &BTreeSet, + ) -> Result<(), crate::engine::BacktestError> { + if self.factor_by_date.values().flatten().any(|row| + fields.iter().any(|field| factor_numeric_value(row, field).is_some())) + { + return Err(crate::engine::BacktestError::Execution( + "decision projection fields already exist in static inputs; historical score replay is forbidden".into(), + )); + } + Ok(()) + } + + pub(crate) fn install_decision_projection( + &mut self, date: NaiveDate, projections: &[crate::decision_projection::DecisionProjectionRow], + ) -> Result<(), crate::engine::BacktestError> { + let mut rows = Vec::with_capacity(projections.len()); + for projection in projections { + let symbol_id = self.symbol_id(&projection.symbol).ok_or_else(|| + crate::engine::BacktestError::Execution("projection symbol has no frozen base instrument".into()))?; + let mut row = self.factor_by_symbol_id(date, symbol_id).cloned().ok_or_else(|| + crate::engine::BacktestError::Execution("projection symbol has no same-date raw factor observation".into()))?; + for (field, value) in &projection.values { + row.extra_factors.insert(Cow::Owned(field.clone()), *value); + } + rows.push((symbol_id, row)); + } + rows.sort_unstable_by_key(|(symbol_id, _)| *symbol_id); + let (symbol_ids, rows) = rows.into_iter().unzip(); + // Only the current selection view is projected. Economic observations, + // history, and other backtest clones keep the shared immutable base. + self.decision_projection = Some(DecisionFactorProjection { date, rows, symbol_ids }); + Ok(()) + } + pub fn candidate(&self, date: NaiveDate, symbol: &str) -> Option<&CandidateEligibility> { let symbol_id = self.symbol_id(symbol)?; self.candidate_by_symbol_id(date, symbol_id) @@ -3334,13 +3388,13 @@ impl DataSet { } pub fn factor_snapshots_on(&self, date: NaiveDate) -> Vec<&DailyFactorSnapshot> { - self.factor_by_date - .get(&date) - .map(|rows| rows.iter().collect()) - .unwrap_or_default() + self.factor_snapshot_rows_on(date).iter().collect() } pub fn factor_snapshot_rows_on(&self, date: NaiveDate) -> &[DailyFactorSnapshot] { + if let Some(item) = self.decision_projection.as_ref().filter(|item| item.date == date) { + return &item.rows; + } self.factor_by_date .get(&date) .map(Vec::as_slice) @@ -3348,6 +3402,9 @@ impl DataSet { } pub fn factor_symbol_ids_on(&self, date: NaiveDate) -> &[u32] { + if let Some(item) = self.decision_projection.as_ref().filter(|item| item.date == date) { + return &item.symbol_ids; + } self.factor_symbol_ids_by_date .get(&date) .map(Vec::as_slice) @@ -3411,7 +3468,7 @@ impl DataSet { date, benchmark, market: self.market_by_date.get(&date).cloned().unwrap_or_default(), - factors: self.factor_by_date.get(&date).cloned().unwrap_or_default(), + factors: self.factor_snapshot_rows_on(date).to_vec(), candidates: self .candidate_by_date .get(&date) @@ -3537,8 +3594,8 @@ impl DataSet { } let output_field = normalize_field(output_field); let mut rows = Vec::new(); - for (_, snapshots) in self.factor_by_date.range(start..=end) { - let Some(snapshot) = snapshots.iter().find(|row| row.symbol == symbol) else { + for (date, _) in self.factor_by_date.range(start..=end) { + let Some(snapshot) = self.factor(*date, symbol) else { continue; }; for field in fields { diff --git a/crates/fidc-core/src/decision_projection.rs b/crates/fidc-core/src/decision_projection.rs new file mode 100644 index 0000000..33efebf --- /dev/null +++ b/crates/fidc-core/src/decision_projection.rs @@ -0,0 +1,272 @@ +//! Run-local, receipt-bound model projections for a completed decision session. + +use std::collections::{BTreeMap, BTreeSet}; +use std::path::{Component, Path}; + +use chrono::{NaiveDate, NaiveDateTime, NaiveTime}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; + +use crate::engine::BacktestError; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct DecisionProjectionContract { + pub contract_sha256: String, + pub fields: BTreeSet, + pub input_cutoff_time: NaiveTime, +} + +impl DecisionProjectionContract { + pub fn validate(&self) -> Result<(), BacktestError> { + require_sha256(&self.contract_sha256)?; + if self.fields.is_empty() || self.fields.iter().any(|field| !valid_projection_field(field)) { + return Err(error("projection fields must be explicit canonical numeric identifiers, not market or typed factors")); + } + Ok(()) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct DecisionProjectionRequest { + pub contract_sha256: String, + pub decision_date: NaiveDate, + pub execution_date: NaiveDate, + pub as_of: NaiveDateTime, + pub symbols: BTreeSet, + pub fields: BTreeSet, +} + +impl DecisionProjectionRequest { + pub fn sha256(&self) -> Result { + json_sha256(self) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct DecisionProjectionRow { + pub symbol: String, + pub values: BTreeMap, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct DecisionProjectionExclusion { + pub symbol: String, + pub classification: String, + pub reason: String, + pub source_sha256: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct DecisionProjectionReference { + pub path: String, + pub sha256: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct DecisionProjectionBatch { + pub contract_sha256: String, + pub decision_date: NaiveDate, + pub execution_date: NaiveDate, + pub input_available_at: NaiveDateTime, + pub rows: Vec, + pub exclusions: Vec, + pub inference_reference: DecisionProjectionReference, + pub provider_request_sha256: String, + pub provider_response_sha256: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct DecisionProjectionReceipt { + pub schema_version: String, + pub request: DecisionProjectionRequest, + pub request_sha256: String, + pub input_available_at: NaiveDateTime, + pub inference_reference: DecisionProjectionReference, + pub provider_request_sha256: String, + pub provider_response_sha256: String, + pub values_sha256: String, + pub row_count: usize, + pub excluded_count: usize, +} + +impl DecisionProjectionBatch { + pub(crate) fn validate( + &self, + request: &DecisionProjectionRequest, + ) -> Result { + if self.contract_sha256 != request.contract_sha256 + || self.decision_date != request.decision_date + || self.execution_date != request.execution_date + || request.as_of.date() != request.decision_date + || request.decision_date >= request.execution_date + || self.input_available_at > request.as_of + { + return Err(error("projection identity or causal availability differs from the decision request")); + } + require_sha256(&self.inference_reference.sha256)?; + require_sha256(&self.provider_request_sha256)?; + require_sha256(&self.provider_response_sha256)?; + let path = Path::new(&self.inference_reference.path); + if !path.is_absolute() || path.components().any(|part| matches!(part, Component::ParentDir)) { + return Err(error("projection inference receipt requires an absolute immutable artifact reference")); + } + if request.symbols.is_empty() || request.fields.is_empty() { + return Err(error("projection request population and fields must be explicit")); + } + let mut covered = BTreeSet::new(); + for row in &self.rows { + if !request.symbols.contains(&row.symbol) || !covered.insert(row.symbol.clone()) { + return Err(error("projection returned an unexpected or duplicate symbol")); + } + if row.values.keys().collect::>() != request.fields.iter().collect() + || row.values.values().any(|value| !value.is_finite()) + { + return Err(error("projection numeric field coverage is incomplete or non-finite")); + } + } + for exclusion in &self.exclusions { + if !request.symbols.contains(&exclusion.symbol) || !covered.insert(exclusion.symbol.clone()) { + return Err(error("projection exclusion overlaps a value or repeats an unexpected symbol")); + } + if exclusion.classification != "valid_market_absence" || exclusion.reason.trim().is_empty() { + return Err(error("projection missing data is unclassified; no order skip is permitted")); + } + require_sha256(&exclusion.source_sha256)?; + } + if covered != request.symbols { + return Err(error("projection did not explain every requested symbol")); + } + // Digest the actual normalized values consumed by this engine, not a + // provider's summary flag. Stable ordering preserves same-input replay. + let rows = self.rows.iter().map(|row| (&row.symbol, &row.values)).collect::>(); + let exclusions = self.exclusions.iter().map(|item| (&item.symbol, item)).collect::>(); + Ok(DecisionProjectionReceipt { + schema_version: "fidc.decision-projection-receipt/v1".into(), + request: request.clone(), + request_sha256: request.sha256()?, + input_available_at: self.input_available_at, + inference_reference: self.inference_reference.clone(), + provider_request_sha256: self.provider_request_sha256.clone(), + provider_response_sha256: self.provider_response_sha256.clone(), + values_sha256: json_sha256(&(rows, exclusions))?, + row_count: self.rows.len(), + excluded_count: self.exclusions.len(), + }) + } +} + +fn valid_projection_field(field: &str) -> bool { + let mut bytes = field.bytes(); + let identifier = bytes.next().is_some_and(|c| c.is_ascii_lowercase() || c == b'_') + && bytes.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == b'_'); + identifier && !matches!(field, + "market_cap" | "market_cap_bn" | "free_float_cap" | "free_float_cap_bn" + | "pe" | "pe_ttm" | "turnover_ratio" | "effective_turnover_ratio" + | "adjustment_factor_backward1" | "open" | "day_open" | "close" + | "high" | "low" | "last" | "last_price" | "volume" | "amount" + | "prev_close" | "bid1" | "ask1" | "upper_limit" | "lower_limit") +} + +fn require_sha256(value: &str) -> Result<(), BacktestError> { + if value.len() != 64 || !value.bytes().all(|c| c.is_ascii_digit() || (b'a'..=b'f').contains(&c)) { + return Err(error("projection requires canonical SHA256 identities")); + } + Ok(()) +} + +fn json_sha256(value: &impl Serialize) -> Result { + let bytes = serde_json::to_vec(value).map_err(|_| error("projection receipt could not be serialized"))?; + Ok(format!("{:x}", Sha256::digest(bytes))) +} + +fn error(message: &str) -> BacktestError { + BacktestError::Execution(message.into()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn request() -> DecisionProjectionRequest { + let decision_date = NaiveDate::from_ymd_opt(2025, 1, 2).unwrap(); + DecisionProjectionRequest { + contract_sha256: "1".repeat(64), decision_date, + execution_date: NaiveDate::from_ymd_opt(2025, 1, 3).unwrap(), + as_of: decision_date.and_hms_opt(16, 0, 0).unwrap(), + symbols: BTreeSet::from(["000001.SZ".into(), "000002.SZ".into()]), + fields: BTreeSet::from(["model_score_a".into()]), + } + } + + fn batch() -> DecisionProjectionBatch { + let request = request(); + DecisionProjectionBatch { + contract_sha256: request.contract_sha256, + decision_date: request.decision_date, execution_date: request.execution_date, + input_available_at: request.as_of, + rows: vec![DecisionProjectionRow { symbol: "000001.SZ".into(), values: BTreeMap::from([("model_score_a".into(), 0.5)]) }], + exclusions: vec![DecisionProjectionExclusion { symbol: "000002.SZ".into(), classification: "valid_market_absence".into(), reason: "dated suspension evidence".into(), source_sha256: "2".repeat(64) }], + inference_reference: DecisionProjectionReference { path: "/private/inference.json".into(), sha256: "3".repeat(64) }, + provider_request_sha256: "4".repeat(64), provider_response_sha256: "5".repeat(64), + } + } + + #[test] + fn exact_projection_coverage_and_classified_absence_are_receipted() { + let receipt = batch().validate(&request()).unwrap(); + assert_eq!((receipt.row_count, receipt.excluded_count), (1, 1)); + assert_eq!(receipt.request_sha256, request().sha256().unwrap()); + let mut changed = batch(); changed.rows[0].values.insert("model_score_a".into(), 0.6); + assert_ne!(changed.validate(&request()).unwrap().values_sha256, receipt.values_sha256); + } + + #[test] + fn projection_contract_does_not_shadow_core_market_or_typed_fields() { + for field in ["", "Score", "model-score", "close", "market_cap_bn", "adjustment_factor_backward1"] { + let contract = DecisionProjectionContract { contract_sha256: "1".repeat(64), fields: BTreeSet::from([field.into()]), input_cutoff_time: NaiveTime::from_hms_opt(16, 0, 0).unwrap() }; + assert!(contract.validate().is_err(), "{field}"); + } + } + + #[test] + fn projection_rejects_incomplete_duplicate_extra_and_overlapping_population() { + let mut examples = vec![]; + let mut item = batch(); item.exclusions.clear(); examples.push(item); + let mut item = batch(); item.rows.push(item.rows[0].clone()); examples.push(item); + let mut item = batch(); item.rows[0].symbol = "999999.SZ".into(); examples.push(item); + let mut item = batch(); item.exclusions[0].symbol = item.rows[0].symbol.clone(); examples.push(item); + for item in examples { assert!(item.validate(&request()).is_err()); } + } + + #[test] + fn projection_rejects_nonfinite_wrong_fields_and_unproven_absence() { + let mut examples = vec![]; + for value in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] { + let mut item = batch(); item.rows[0].values.insert("model_score_a".into(), value); examples.push(item); + } + let mut item = batch(); item.rows[0].values.clear(); examples.push(item); + let mut item = batch(); item.rows[0].values.insert("extra".into(), 1.0); examples.push(item); + let mut item = batch(); item.exclusions[0].classification = "unknown".into(); examples.push(item); + let mut item = batch(); item.exclusions[0].reason.clear(); examples.push(item); + let mut item = batch(); item.exclusions[0].source_sha256.clear(); examples.push(item); + for item in examples { assert!(item.validate(&request()).is_err()); } + } + + #[test] + fn projection_rejects_future_drift_or_unbound_receipts() { + let mut examples = vec![]; + let mut item = batch(); item.input_available_at += chrono::Duration::seconds(1); examples.push(item); + let mut item = batch(); item.decision_date = item.execution_date; examples.push(item); + let mut item = batch(); item.contract_sha256 = "0".repeat(64); examples.push(item); + let mut item = batch(); item.inference_reference.path = "/private/../inference.json".into(); examples.push(item); + let mut item = batch(); item.provider_response_sha256 = "A".repeat(64); examples.push(item); + for item in examples { assert!(item.validate(&request()).is_err()); } + } +} diff --git a/crates/fidc-core/src/engine.rs b/crates/fidc-core/src/engine.rs index d5c1943..ca24017 100644 --- a/crates/fidc-core/src/engine.rs +++ b/crates/fidc-core/src/engine.rs @@ -7,6 +7,10 @@ use thiserror::Error; use crate::broker::{BrokerExecutionReport, BrokerSimulator, MatchingType}; use crate::cost::CostModel; use crate::data::{BenchmarkSnapshot, DataSet, DataSetError, IntradayExecutionQuote, PriceField}; +use crate::decision_projection::{ + DecisionProjectionBatch, DecisionProjectionContract, DecisionProjectionReceipt, + DecisionProjectionRequest, +}; use crate::event_bus::{BacktestProcessMod, BacktestProcessModLoader, ProcessEventBus}; use crate::events::{ AccountEvent, FillEvent, OrderEvent, OrderSide, OrderStatus, PositionEvent, ProcessEvent, @@ -137,6 +141,7 @@ pub struct BacktestResult { pub daily_holdings: Vec, pub metrics: BacktestMetrics, pub terminal_audit: BacktestTerminalAudit, + pub decision_projection_receipts: Vec, } #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] @@ -234,6 +239,12 @@ type ExecutionQuoteLoader = Box< dyn FnMut(ExecutionQuoteRequest) -> Result, BacktestError> + Send, >; +struct DecisionProjectionLoader { + contract: DecisionProjectionContract, + load: Box Result + Send>, + requested_dates: BTreeSet, +} + #[derive(Debug, Clone, Serialize)] pub struct AnalyzerTradeRow { #[serde(with = "date_format")] @@ -303,6 +314,8 @@ pub struct AnalyzerReport { pub benchmark_series: Vec, pub metrics: BacktestMetrics, pub terminal_audit: BacktestTerminalAudit, + #[serde(skip_serializing_if = "Vec::is_empty")] + pub decision_projection_receipts: Vec, } impl BacktestResult { @@ -347,6 +360,7 @@ impl BacktestResult { benchmark_series: self.benchmark_series.clone(), metrics: self.metrics.clone(), terminal_audit: self.terminal_audit.clone(), + decision_projection_receipts: self.decision_projection_receipts.clone(), } } @@ -486,6 +500,7 @@ pub struct BacktestEngine { futures_cost_model: FuturesTransactionCostModel, futures_validation_config: FuturesValidationConfig, execution_quote_loader: Option, + decision_projection_loader: Option, execution_quote_request_cache: BTreeSet<(NaiveDate, String, Option, Option)>, execution_absence_notes: BTreeMap>, @@ -598,11 +613,12 @@ pub fn backtest_execution_dates_with_rules( impl BacktestEngine { pub fn new( - data: DataSet, + mut data: DataSet, strategy: S, broker: BrokerSimulator, config: BacktestConfig, ) -> Self { + data.clear_decision_projection(); Self { data, strategy, @@ -623,6 +639,7 @@ impl BacktestEngine { futures_cost_model: FuturesTransactionCostModel::default(), futures_validation_config: FuturesValidationConfig::default(), execution_quote_loader: None, + decision_projection_loader: None, execution_quote_request_cache: BTreeSet::new(), execution_absence_notes: BTreeMap::new(), execution_lifecycle_reported: BTreeSet::new(), @@ -664,7 +681,8 @@ impl BacktestEngine { self } - pub fn into_data(self) -> DataSet { + pub fn into_data(mut self) -> DataSet { + self.data.clear_decision_projection(); self.data } @@ -678,6 +696,89 @@ impl BacktestEngine { self } + pub fn with_decision_projection_loader( + mut self, contract: DecisionProjectionContract, loader: F, + ) -> Result + where + F: FnMut(DecisionProjectionRequest) -> Result + Send + 'static, + { + contract.validate()?; + if self.config.decision_lag_trading_days == 0 || self.decision_projection_loader.is_some() { + return Err(BacktestError::Execution( + "decision projection requires a single loader and a completed prior trading session".into(), + )); + } + self.data.require_unmaterialized_projection_fields(&contract.fields)?; + self.decision_projection_loader = Some(DecisionProjectionLoader { + contract, load: Box::new(loader), requested_dates: BTreeSet::new(), + }); + Ok(self) + } + + fn prepare_decision_projection( + &mut self, execution_date: NaiveDate, decision_date: NaiveDate, + ) -> Result, BacktestError> { + let Some(loader) = &mut self.decision_projection_loader else { return Ok(None); }; + if !loader.requested_dates.insert(execution_date) { + return Err(BacktestError::Execution("projection request was already dispatched by this engine; verify its durable receipt before recovery".into())); + } + let symbols = self.data.candidate_snapshot_rows_on(decision_date).iter() + .map(|row| row.symbol.to_string()).collect::>(); + if symbols.is_empty() || symbols.iter().any(|symbol| self.data.factor(decision_date, symbol).is_none()) { + return Err(BacktestError::Execution("decision projection requires complete same-date candidate and raw factor observations".into())); + } + let request = DecisionProjectionRequest { + contract_sha256: loader.contract.contract_sha256.clone(), + decision_date, execution_date, + as_of: decision_date.and_time(loader.contract.input_cutoff_time), + symbols, fields: loader.contract.fields.clone(), + }; + let batch = (loader.load)(request.clone())?; + let receipt = batch.validate(&request)?; + self.data.install_decision_projection(decision_date, &batch.rows)?; + Ok(Some(receipt)) + } + + fn validate_decision_projection_calendar(&self) -> Result<(), BacktestError> { + let Some(loader) = &self.decision_projection_loader else { return Ok(()); }; + if !loader.requested_dates.is_empty() { + return Err(BacktestError::Execution("a projection-enabled engine cannot replay its dispatched run in place".into())); + } + let dates = self.data.calendar().days(); + for (index, execution_date) in dates.iter().copied().enumerate() { + if self.config.start_date.is_some_and(|start| execution_date < start) + || self.config.end_date.is_some_and(|end| execution_date > end) { continue; } + let Some(decision_index) = index.checked_sub(self.config.decision_lag_trading_days) else { + if self.config.start_date.is_some() { + return Err(BacktestError::Execution("projection run is missing the completed signal session before its requested start".into())); + } + continue; + }; + let decision_date = dates[decision_index]; + if self.data.market_snapshot_rows_on(execution_date).is_empty() + || self.data.factor_snapshot_rows_on(decision_date).is_empty() + || self.data.candidate_snapshot_rows_on(decision_date).is_empty() + { + return Err(BacktestError::Execution(format!( + "projection calendar has unresolved market/candidate/factor data for decision {decision_date} / execution {execution_date}; no session may be silently dropped" + ))); + } + } + Ok(()) + } + + fn decision_projection_execution_schedule(&self) -> Vec<(NaiveDate, Option<(usize, NaiveDate)>)> { + let dates = self.data.calendar().days(); + dates.iter().copied().enumerate() + .filter(|(_, date)| self.config.start_date.is_none_or(|start| *date >= start)) + .filter(|(_, date)| self.config.end_date.is_none_or(|end| *date <= end)) + .enumerate().map(|(execution_index, (calendar_index, execution_date))| { + let signal = calendar_index.checked_sub(self.config.decision_lag_trading_days) + .map(|signal_index| (execution_index, dates[signal_index])); + (execution_date, signal) + }).collect() + } + pub fn with_dividend_reinvestment(mut self, enabled: bool) -> Self { self.dividend_reinvestment = enabled; self @@ -3026,6 +3127,7 @@ where where F: FnMut(&BacktestDayProgress) -> Result<(), BacktestError>, { + self.validate_decision_projection_calendar()?; let mut portfolio = PortfolioState::new(self.config.initial_cash); let mut manual_cursor = self.manual_execution_source.as_ref().map(|source| crate::manual_execution::ManualReplayCursor::from_shared(source.clone())).transpose().map_err(BacktestError::Execution)?; @@ -3033,13 +3135,15 @@ where self.subscriptions = self.strategy.initial_subscriptions(); let scheduler_calendar = self.data.calendar().clone(); let scheduler = Scheduler::new(&scheduler_calendar); - let execution_schedule = backtest_execution_schedule_with_rules( + let execution_schedule = if self.decision_projection_loader.is_some() { + self.decision_projection_execution_schedule() + } else { backtest_execution_schedule_with_rules( &self.data, self.config.start_date, self.config.end_date, self.config.decision_lag_trading_days, &self.strategy.schedule_rules(), - ); + ) }; let execution_dates = execution_schedule .iter() .map(|(execution_date, _)| *execution_date) @@ -3112,12 +3216,20 @@ where daily_holdings: Vec::new(), metrics: BacktestMetrics::default(), terminal_audit: BacktestTerminalAudit::default(), + decision_projection_receipts: Vec::new(), }; let mut stock_equity_by_date = BTreeMap::::new(); let mut previous_external_cash_flow_total = portfolio.external_cash_flow_total(); let mut cumulative_corporate_allocation_count = 0; for (execution_idx, execution_date) in execution_dates.iter().copied().enumerate() { + let decision_slot = execution_schedule.get(execution_idx).and_then(|(_, slot)| *slot); + self.data.clear_decision_projection(); + if let Some((_, decision_date)) = decision_slot + && let Some(receipt) = self.prepare_decision_projection(execution_date, decision_date)? + { + result.decision_projection_receipts.push(receipt); + } let day_order_start = result.order_events.len(); let day_fill_start = result.fills.len(); let day_manual_start = result.manual_executions.len(); @@ -3187,9 +3299,6 @@ where execution_date, execution_date, ); - let decision_slot = execution_schedule - .get(execution_idx) - .and_then(|(_, decision_slot)| *decision_slot); let Some((decision_index, decision_date)) = decision_slot else { let mut process_events = Vec::new(); let mut report = self.execute_deferred_session_without_signal( @@ -7192,6 +7301,165 @@ mod tests { } } + fn projection_contract() -> crate::DecisionProjectionContract { + crate::DecisionProjectionContract { + contract_sha256: "a".repeat(64), fields: BTreeSet::from(["model_score_a".into()]), + input_cutoff_time: NaiveTime::from_hms_opt(16, 0, 0).unwrap(), + } + } + + fn projection_batch(request: &crate::DecisionProjectionRequest, score: f64) -> crate::DecisionProjectionBatch { + crate::DecisionProjectionBatch { + contract_sha256: request.contract_sha256.clone(), decision_date: request.decision_date, + execution_date: request.execution_date, input_available_at: request.as_of, + rows: request.symbols.iter().map(|symbol| crate::DecisionProjectionRow { + symbol: symbol.clone(), values: BTreeMap::from([("model_score_a".into(), score)]), + }).collect(), exclusions: vec![], + inference_reference: crate::DecisionProjectionReference { + path: format!("/private/inference-{}.json", request.execution_date), sha256: "b".repeat(64), + }, provider_request_sha256: "c".repeat(64), provider_response_sha256: "d".repeat(64), + } + } + + struct ProjectionProbe { seen: Arc>> } + + impl Strategy for ProjectionProbe { + fn name(&self) -> &str { "decision-projection-probe" } + + fn before_trading(&mut self, ctx: &StrategyContext<'_>) -> Result<(), crate::BacktestError> { + assert!(ctx.data.factor_numeric_value(ctx.decision_date, SYMBOL, "model_score_a").is_some()); + assert!(ctx.data.factor_numeric_value(ctx.execution_date, SYMBOL, "model_score_a").is_none()); + Ok(()) + } + + fn on_day(&mut self, ctx: &StrategyContext<'_>) -> Result { + let score = ctx.data.factor_numeric_value(ctx.decision_date, SYMBOL, "model_score_a") + .ok_or_else(|| crate::BacktestError::Execution("missing current projection".into()))?; + self.seen.lock().unwrap().push((ctx.decision_date, score)); + let symbol_id = ctx.data.symbol_id(SYMBOL).unwrap(); + assert_eq!(ctx.data.daily_snapshot_view(ctx.decision_date).factor(symbol_id) + .unwrap().extra_factors.get("model_score_a").copied(), Some(score)); + assert_eq!(ctx.data.factor_snapshot_rows_on(ctx.decision_date)[0].extra_factors.get("model_score_a").copied(), Some(score)); + assert_eq!(ctx.data.get_factor(SYMBOL, ctx.decision_date, ctx.decision_date, "model_score_a")[0].value, score); + Ok(StrategyDecision { order_intents: if score > 0.0 { vec![OrderIntent::Shares { + symbol: SYMBOL.into(), quantity: 100, reason: "current causal model score".into(), + }] } else { vec![] }, ..Default::default() }) + } + } + + fn projection_probe_engine(data: DataSet, seen: Arc>>) + -> BacktestEngine + { + let broker = BrokerSimulator::new_with_execution_price(ChinaAShareCostModel::default(), ChinaEquityRuleHooks, PriceField::Open) + .with_matching_type(MatchingType::NextBarOpen).with_volume_limit(false).with_liquidity_limit(false); + BacktestEngine::new(data, ProjectionProbe { seen }, broker, BacktestConfig { + initial_cash: 10000.0, benchmark_code: "000852.SH".into(), start_date: None, end_date: None, + decision_lag_trading_days: 1, execution_price_field: PriceField::Open, + }) + } + + #[test] + fn decision_projection_is_called_before_callbacks_and_is_not_shared_or_carried_forward() { + let dates = [d(2025, 1, 2), d(2025, 1, 3), d(2025, 1, 6)]; + let data = dataset_from_market_and_candidates(dates.iter().map(|date| market(*date, 10.0, 10.0)).collect(), dates.iter().map(|date| candidate(*date)).collect()); + let base = data.clone(); + let seen = Arc::new(Mutex::new(vec![])); + let requested = Arc::new(Mutex::new(vec![])); + let recorded = requested.clone(); + let mut engine = projection_probe_engine(data, seen.clone()).with_decision_projection_loader(projection_contract(), move |request| { + recorded.lock().unwrap().push(request.clone()); + Ok(projection_batch(&request, if request.decision_date == dates[0] { 0.5 } else { -0.5 })) + }).unwrap(); + let result = engine.run().unwrap(); + assert_eq!(*seen.lock().unwrap(), vec![(dates[0], 0.5), (dates[1], -0.5)]); + assert_eq!(requested.lock().unwrap().len(), 2); + assert_eq!(result.decision_projection_receipts.len(), 2); + assert_eq!(result.fills.len(), 1); + assert_eq!(result.fills[0].date, dates[1]); + for date in dates { assert_eq!(base.factor_numeric_value(date, SYMBOL, "model_score_a"), None); } + let run_data = engine.into_data(); + assert_eq!(run_data.factor_numeric_value(dates[0], SYMBOL, "model_score_a"), None); + assert_eq!(run_data.factor_numeric_value(dates[1], SYMBOL, "model_score_a"), None); + let clean = projection_probe_engine(run_data.clone(), Arc::new(Mutex::new(vec![]))).into_data(); + assert_eq!(clean.factor_numeric_value(dates[1], SYMBOL, "model_score_a"), None); + } + + #[test] + fn decision_projection_failure_never_reaches_strategy_and_does_not_blindly_retry() { + let seen = Arc::new(Mutex::new(vec![])); + let attempts = Arc::new(Mutex::new(0)); + let counted = attempts.clone(); + let mut engine = projection_probe_engine(dataset(), seen.clone()).with_decision_projection_loader(projection_contract(), move |_| { + *counted.lock().unwrap() += 1; + Err(crate::BacktestError::Execution("provider outcome unknown; inspect its receipt".into())) + }).unwrap(); + assert!(engine.run().unwrap_err().to_string().contains("provider outcome unknown")); + assert!(seen.lock().unwrap().is_empty()); + assert!(engine.run().unwrap_err().to_string().contains("cannot replay")); + assert_eq!(*attempts.lock().unwrap(), 1); + } + + #[test] + fn decision_projection_requested_first_session_uses_visible_preperiod_inputs() { + let first = d(2025, 1, 3); + let seen = Arc::new(Mutex::new(vec![])); + let mut engine = projection_probe_engine(dataset(), seen.clone()); + engine.config.start_date = Some(first); + engine.config.end_date = Some(first); + let mut engine = engine.with_decision_projection_loader(projection_contract(), move |request| { + assert_eq!(request.execution_date, first); + assert_eq!(request.decision_date, d(2025, 1, 2)); + Ok(projection_batch(&request, 0.5)) + }).unwrap(); + let result = engine.run().unwrap(); + assert_eq!(result.decision_projection_receipts.len(), 1); + assert_eq!(result.fills[0].date, first); + assert_eq!(*seen.lock().unwrap(), vec![(d(2025, 1, 2), 0.5)]); + } + + #[test] + fn decision_projection_does_not_silently_shift_a_start_without_prior_inputs() { + let mut engine = projection_probe_engine(dataset(), Arc::new(Mutex::new(vec![]))); + engine.config.start_date = Some(d(2025, 1, 2)); + let mut engine = engine.with_decision_projection_loader(projection_contract(), |_| panic!("missing warmup must be classified before dispatch")).unwrap(); + assert!(engine.run().unwrap_err().to_string().contains("before its requested start")); + } + + #[test] + fn decision_projection_rejects_future_response_before_any_strategy_callback() { + let seen = Arc::new(Mutex::new(vec![])); + let mut engine = projection_probe_engine(dataset(), seen.clone()).with_decision_projection_loader(projection_contract(), move |request| { + let mut batch = projection_batch(&request, 1.0); + batch.input_available_at += chrono::Duration::seconds(1); + Ok(batch) + }).unwrap(); + assert!(engine.run().unwrap_err().to_string().contains("causal availability")); + assert!(seen.lock().unwrap().is_empty()); + } + + #[test] + fn decision_projection_rejects_missing_calendar_inputs_instead_of_dropping_a_session() { + let dates = [d(2025, 1, 2), d(2025, 1, 3), d(2025, 1, 6)]; + let data = DataSet::from_components(vec![default_instrument()], dates.iter().map(|date| market(*date, 10.0, 10.0)).collect(), + vec![factor(dates[0])], dates.iter().map(|date| candidate(*date)).collect(), dates.iter().map(|date| benchmark(*date)).collect()).unwrap(); + let mut engine = projection_probe_engine(data, Arc::new(Mutex::new(vec![]))).with_decision_projection_loader(projection_contract(), move |_| panic!("unresolved data must fail before dispatch")).unwrap(); + assert!(engine.run().unwrap_err().to_string().contains("no session may be silently dropped")); + } + + #[test] + fn decision_projection_refuses_static_scores_and_same_session_inputs() { + let first = d(2025, 1, 2); let second = d(2025, 1, 3); + let mut factors = vec![factor(first), factor(second)]; + factors[1].extra_factors.insert(std::borrow::Cow::Borrowed("model_score_a"), 0.5); + let data = DataSet::from_components(vec![default_instrument()], vec![market(first, 10.0, 10.0), market(second, 10.0, 10.0)], factors, + vec![candidate(first), candidate(second)], vec![benchmark(first), benchmark(second)]).unwrap(); + let result = projection_probe_engine(data, Arc::new(Mutex::new(vec![]))).with_decision_projection_loader(projection_contract(), |_| unreachable!()); + assert!(matches!(result, Err(error) if error.to_string().contains("historical score replay"))); + let mut engine = projection_probe_engine(dataset(), Arc::new(Mutex::new(vec![]))); + engine.config.decision_lag_trading_days = 0; + assert!(engine.with_decision_projection_loader(projection_contract(), |_| unreachable!()).is_err()); + } + fn dataset_with( first_market: DailyMarketSnapshot, second_market: DailyMarketSnapshot, diff --git a/crates/fidc-core/src/lib.rs b/crates/fidc-core/src/lib.rs index c2830e7..72fbf2d 100644 --- a/crates/fidc-core/src/lib.rs +++ b/crates/fidc-core/src/lib.rs @@ -3,6 +3,7 @@ pub mod calendar; pub mod cost; mod corporate_book; pub mod data; +pub mod decision_projection; mod numeric_factors; pub mod daily_patterns; pub mod pattern_context; @@ -60,6 +61,11 @@ pub use data::{ IntradayMarketSnapshotOverlay, IntradayOrderBookDepthLevel, NumericFactorMap, PriceBar, PriceField, SecuritiesMarginRecord, SplitRecord, YieldCurvePoint, }; +pub use decision_projection::{ + DecisionProjectionBatch, DecisionProjectionContract, DecisionProjectionExclusion, + DecisionProjectionReceipt, DecisionProjectionReference, DecisionProjectionRequest, + DecisionProjectionRow, +}; pub use engine::{ AnalyzerMonthlyReturnRow, AnalyzerPositionRow, AnalyzerReport, AnalyzerRiskSummary, AnalyzerTradeRow, BacktestConfig, BacktestDayProgress, BacktestEngine, BacktestError,