增加逐决策动态模型投影并隔离共享数据与首日预热

This commit is contained in:
boris
2026-10-03 12:29:11 +08:00
parent 7485d4ceb9
commit 2bc033b8cc
4 changed files with 622 additions and 19 deletions
+69 -12
View File
@@ -1425,6 +1425,7 @@ pub struct DataSet {
factor_by_date: Arc<BTreeMap<NaiveDate, Vec<DailyFactorSnapshot>>>,
factor_symbol_ids_by_date: Arc<BTreeMap<NaiveDate, Vec<u32>>>,
factor_row_positions_by_date: Arc<Option<DenseRowPositionIndex>>,
decision_projection: Option<DecisionFactorProjection>,
factor_market_cap_order_by_date: Arc<BTreeMap<NaiveDate, Vec<u32>>>,
factor_text_by_date: Arc<BTreeMap<NaiveDate, Vec<FactorTextValue>>>,
factor_text_index: Arc<HashMap<(NaiveDate, String, String), FactorTextValue>>,
@@ -1451,6 +1452,13 @@ pub struct DataSet {
completed_minute_bars: Arc<BTreeMap<(NaiveDate,String),Vec<crate::session_events::MinuteBar>>>,
}
#[derive(Debug, Clone)]
struct DecisionFactorProjection {
date: NaiveDate,
rows: Vec<DailyFactorSnapshot>,
symbol_ids: Vec<u32>,
}
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<String>,
) -> 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 {
+272
View File
@@ -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<String>,
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<String>,
pub fields: BTreeSet<String>,
}
impl DecisionProjectionRequest {
pub fn sha256(&self) -> Result<String, BacktestError> {
json_sha256(self)
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(deny_unknown_fields)]
pub struct DecisionProjectionRow {
pub symbol: String,
pub values: BTreeMap<String, f64>,
}
#[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<DecisionProjectionRow>,
pub exclusions: Vec<DecisionProjectionExclusion>,
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<DecisionProjectionReceipt, BacktestError> {
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::<BTreeSet<_>>() != 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::<BTreeMap<_, _>>();
let exclusions = self.exclusions.iter().map(|item| (&item.symbol, item)).collect::<BTreeMap<_, _>>();
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<String, BacktestError> {
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()); }
}
}
+275 -7
View File
@@ -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<HoldingSummary>,
pub metrics: BacktestMetrics,
pub terminal_audit: BacktestTerminalAudit,
pub decision_projection_receipts: Vec<DecisionProjectionReceipt>,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
@@ -234,6 +239,12 @@ type ExecutionQuoteLoader = Box<
dyn FnMut(ExecutionQuoteRequest) -> Result<Vec<IntradayExecutionQuote>, BacktestError> + Send,
>;
struct DecisionProjectionLoader {
contract: DecisionProjectionContract,
load: Box<dyn FnMut(DecisionProjectionRequest) -> Result<DecisionProjectionBatch, BacktestError> + Send>,
requested_dates: BTreeSet<NaiveDate>,
}
#[derive(Debug, Clone, Serialize)]
pub struct AnalyzerTradeRow {
#[serde(with = "date_format")]
@@ -303,6 +314,8 @@ pub struct AnalyzerReport {
pub benchmark_series: Vec<BenchmarkSnapshot>,
pub metrics: BacktestMetrics,
pub terminal_audit: BacktestTerminalAudit,
#[serde(skip_serializing_if = "Vec::is_empty")]
pub decision_projection_receipts: Vec<DecisionProjectionReceipt>,
}
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<S, C, R> {
futures_cost_model: FuturesTransactionCostModel,
futures_validation_config: FuturesValidationConfig,
execution_quote_loader: Option<ExecutionQuoteLoader>,
decision_projection_loader: Option<DecisionProjectionLoader>,
execution_quote_request_cache:
BTreeSet<(NaiveDate, String, Option<NaiveTime>, Option<NaiveTime>)>,
execution_absence_notes: BTreeMap<NaiveDate, Vec<String>>,
@@ -598,11 +613,12 @@ pub fn backtest_execution_dates_with_rules(
impl<S, C, R> BacktestEngine<S, C, R> {
pub fn new(
data: DataSet,
mut data: DataSet,
strategy: S,
broker: BrokerSimulator<C, R>,
config: BacktestConfig,
) -> Self {
data.clear_decision_projection();
Self {
data,
strategy,
@@ -623,6 +639,7 @@ impl<S, C, R> BacktestEngine<S, C, R> {
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<S, C, R> BacktestEngine<S, C, R> {
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<S, C, R> BacktestEngine<S, C, R> {
self
}
pub fn with_decision_projection_loader<F>(
mut self, contract: DecisionProjectionContract, loader: F,
) -> Result<Self, BacktestError>
where
F: FnMut(DecisionProjectionRequest) -> Result<DecisionProjectionBatch, BacktestError> + 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<Option<DecisionProjectionReceipt>, 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::<BTreeSet<_>>();
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::<NaiveDate, f64>::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<Mutex<Vec<(NaiveDate, f64)>>> }
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<StrategyDecision, crate::BacktestError> {
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<Mutex<Vec<(NaiveDate, f64)>>>)
-> BacktestEngine<ProjectionProbe, ChinaAShareCostModel, ChinaEquityRuleHooks>
{
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,
+6
View File
@@ -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,