zsb/fidc-backtest-engine
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

修复迟到成交跨公司行为的经济账本校正

Browse Source
This commit is contained in:
boris
2026-09-14 21:14:20 +08:00
parent ef9cc39882
commit 05f1cbbe00
11 changed files with 1447 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/fidc-core/src/broker.rs
+9
View File
@@ -2766,6 +2766,15 @@ where
if execution.side == OrderSide::Sell {
let date = execution.executed_at.with_timezone(&chrono::FixedOffset::east_opt(8 * 3600).unwrap()).date_naive();
self.mark_same_day_sold(date, &execution.symbol);
if let Some(adjustment) = &execution.corporate_adjustment {
for successor in adjustment
.actions
.iter()
.filter_map(|action| action.successor_symbol.as_deref())
{
self.mark_same_day_sold(date, successor);
}
}
}
}
crates/fidc-core/src/corporate_book.rs
+161
View File
@@ -0,0 +1,161 @@
use crate::{
AccountEvent, BacktestError, BrokerExecutionReport, CashReceivable, DataSet, PortfolioState,
PositionEvent,
};
use chrono::NaiveDate;
/// One corporate-action calculation for normal processing and audited replay.
pub(crate) fn apply(
date: NaiveDate,
data: &DataSet,
portfolio: &mut PortfolioState,
notes: &mut Vec<String>,
cash_dividends_enabled: bool,
cash_dividend_adjusts_cost_basis: bool,
) -> Result<BrokerExecutionReport, BacktestError> {
let mut report = BrokerExecutionReport::default();
for action in data.corporate_actions_on(date) {
if !action.has_effect() {
continue;
}
let Some(existing_position) = portfolio.position(&action.symbol) else {
continue;
};
if existing_position.quantity == 0 {
continue;
}
if cash_dividends_enabled && action.share_cash.abs() > f64::EPSILON {
let cash_before = portfolio.cash();
let (cash_delta, quantity_after, average_cost) = {
let position = portfolio
.position_mut_if_exists(&action.symbol)
.expect("position exists for dividend action");
let cash_delta = if cash_dividend_adjusts_cost_basis {
position.apply_cash_dividend(action.share_cash)
} else {
position.apply_cash_dividend_preserve_cost_basis(action.share_cash)
};
(cash_delta, position.quantity, position.average_cost)
};
if cash_delta.abs() > f64::EPSILON {
let payable_date = action.payable_date.unwrap_or(date);
portfolio.add_cash_receivable(CashReceivable {
symbol: action.symbol.clone(),
ex_date: date,
payable_date,
amount: cash_delta,
reason: format!("cash_dividend {:.6}", action.share_cash),
});
let note = format!(
"cash_dividend_receivable {} share_cash={:.6} quantity={} payable_date={} cash={:.2}",
action.symbol, action.share_cash, quantity_after, payable_date, cash_delta
);
notes.push(note.clone());
report.account_events.push(AccountEvent {
date,
cash_before,
cash_after: portfolio.cash(),
total_equity: portfolio.total_equity(),
note,
});
report.position_events.push(PositionEvent {
date,
symbol: action.symbol.clone(),
delta_quantity: 0,
quantity_after,
average_cost,
realized_pnl_delta: 0.0,
reason: format!("cash_dividend {:.6}", action.share_cash),
});
}
}
let split_ratio = action.split_ratio();
if (split_ratio - 1.0).abs() > f64::EPSILON {
portfolio
.adjust_stock_pool_split(&action.symbol, split_ratio)
.map_err(BacktestError::Execution)?;
let (delta_quantity, quantity_after, average_cost) = {
let position = portfolio
.position_mut_if_exists(&action.symbol)
.expect("position exists for split action");
let delta_quantity = position.apply_split_ratio(split_ratio);
(delta_quantity, position.quantity, position.average_cost)
};
if delta_quantity != 0 {
let note = format!(
"stock_split {} ratio={:.6} delta_qty={}",
action.symbol, split_ratio, delta_quantity
);
notes.push(note);
report.position_events.push(PositionEvent {
date,
symbol: action.symbol.clone(),
delta_quantity,
quantity_after,
average_cost,
realized_pnl_delta: 0.0,
reason: format!("stock_split {:.6}", split_ratio),
});
}
}
if action.has_successor_conversion() {
let successor_symbol = action
.successor_symbol
.as_deref()
.expect("successor symbol checked");
let Some(outcome) = portfolio.apply_successor_conversion(
&action.symbol,
successor_symbol,
action.successor_ratio_value(),
action.successor_cash_value(),
) else {
continue;
};
let reason = format!(
"successor_conversion {}->{} ratio={:.6} cash_per_share={:.6}",
outcome.old_symbol,
outcome.new_symbol,
action.successor_ratio_value(),
action.successor_cash_value()
);
notes.push(reason.clone());
report.position_events.push(PositionEvent {
date,
symbol: outcome.old_symbol.clone(),
delta_quantity: -(outcome.old_quantity as i32),
quantity_after: 0,
average_cost: 0.0,
realized_pnl_delta: 0.0,
reason: reason.clone(),
});
report.position_events.push(PositionEvent {
date,
symbol: outcome.new_symbol.clone(),
delta_quantity: outcome.new_quantity_delta,
quantity_after: outcome.new_quantity_after,
average_cost: outcome.new_average_cost_after,
realized_pnl_delta: 0.0,
reason: reason.clone(),
});
if outcome.cash_delta.abs() > f64::EPSILON {
let cash_before = portfolio.cash();
portfolio
.apply_cash_delta(outcome.cash_delta)
.map_err(BacktestError::Execution)?;
report.account_events.push(AccountEvent {
date,
cash_before,
cash_after: portfolio.cash(),
total_equity: portfolio.total_equity(),
note: format!("{} cash={:.2}", reason, outcome.cash_delta),
});
}
}
}
portfolio.prune_flat_positions();
Ok(report)
}
crates/fidc-core/src/engine.rs
+137 -153
View File
@@ -17,7 +17,7 @@ use crate::futures::{
FuturesTransactionCostModel,
};
use crate::metrics::{BacktestMetrics, RiskFreeRateContract, compute_backtest_metrics_with_manual};
use crate::portfolio::{CashReceivable, HoldingSummary, PortfolioState};
use crate::portfolio::{HoldingSummary, PortfolioState};
use crate::risk_control::{FidcRiskDecisionAudit, RiskCheckScope};
use crate::rules::EquityRuleHooks;
use crate::scheduler::{ScheduleRule, ScheduleStage, Scheduler, default_stage_time};
@@ -490,6 +490,7 @@ pub struct BacktestEngine<S, C, R> {
execution_lifecycle_reported: BTreeSet<(String, String)>,
risk_free_rate_contract: Option<RiskFreeRateContract>,
manual_execution_source: Option<std::sync::Arc<crate::manual_execution::ManualExecutionReplay>>,
manual_corporate_replay: Option<crate::manual_corporate_replay::ManualCorporateReplay>,
deferred_session_decisions: Vec<DeferredSessionDecision>,
}
@@ -625,6 +626,7 @@ impl<S, C, R> BacktestEngine<S, C, R> {
execution_lifecycle_reported: BTreeSet::new(),
risk_free_rate_contract: None,
manual_execution_source: None,
manual_corporate_replay: None,
deferred_session_decisions: Vec::new(),
}
}
@@ -815,8 +817,52 @@ where
let cash_before = portfolio.cash();
let conflict = self.has_open_orders() || self.broker.has_pending_stock_pool_execution()
|| self.broker.pending_etf_target_count() > 0;
let application = cursor.advance_next(portfolio, &self.data, conflict)
.map_err(BacktestError::Execution)?.expect("next observation checked");
let requires_corporate_replay = self
.manual_corporate_replay
.as_ref()
.map(|replay| {
replay.required(
cursor.next_observation().expect("next observation checked"),
&self.data,
)
})
.transpose()
.map_err(BacktestError::Execution)?
.unwrap_or(false);
let application = if requires_corporate_replay {
if self.futures_account.is_some() {
return Err(BacktestError::Execution(
"manual corporate replay requires an equity-only financial ledger".into(),
));
}
let source = cursor.frozen_source();
let applied_count = cursor.applied_count();
let replay = self
.manual_corporate_replay
.as_ref()
.expect("corporate replay checked");
let application = cursor
.advance_next_projected(portfolio, |observation, next| {
replay.project(
&source,
applied_count,
observation,
next,
&self.data,
&result.fills,
conflict,
)
})
.map_err(BacktestError::Execution)?
.expect("next observation checked");
replay.committed(cursor.applied_count());
application
} else {
cursor
.advance_next(portfolio, &self.data, conflict)
.map_err(BacktestError::Execution)?
.expect("next observation checked")
};
self.broker.record_observed_manual_execution(&application);
self.strategy.on_observed_manual_execution(&application)?;
result.account_events.push(AccountEvent { date: observed.date(), cash_before,
@@ -1629,10 +1675,20 @@ where
reason,
} => {
let cash_before = portfolio.cash();
let fixed_before = portfolio.cash_fixed();
if receiving_days == 0 {
portfolio
.deposit_withdraw(amount)
.map_err(BacktestError::Execution)?;
if let Some(replay) = &self.manual_corporate_replay {
replay
.record_cash(
callback_datetime,
fixed_before,
portfolio.cash_fixed(),
)
.map_err(BacktestError::Execution)?;
}
directive_report.account_events.push(AccountEvent {
date: execution_date,
cash_before,
@@ -1694,10 +1750,16 @@ where
}
crate::strategy::OrderIntent::FinanceRepay { amount, reason } => {
let cash_before = portfolio.cash();
let fixed_before = portfolio.cash_fixed();
let liabilities_before = portfolio.cash_liabilities();
portfolio
.finance_repay(amount)
.map_err(BacktestError::Execution)?;
if let Some(replay) = &self.manual_corporate_replay {
replay
.record_cash(callback_datetime, fixed_before, portfolio.cash_fixed())
.map_err(BacktestError::Execution)?;
}
directive_report.account_events.push(AccountEvent {
date: execution_date,
cash_before,
@@ -2979,9 +3041,38 @@ where
.iter()
.map(|(execution_date, _)| *execution_date)
.collect::<Vec<_>>();
if let (Some(first), Some(observed)) = (execution_dates.first(), manual_cursor.as_ref().and_then(|cursor| cursor.next_observation_at())) {
if observed.with_timezone(&chrono::FixedOffset::east_opt(8 * 3600).unwrap()).date_naive() < *first {
return Err(BacktestError::Execution("manual observations precede the declared initial portfolio period".into()));
self.manual_corporate_replay = if manual_has_fills
&& execution_dates
.iter()
.any(|date| !self.data.corporate_actions_on(*date).is_empty())
{
execution_dates.first().copied().map(|first| {
crate::manual_corporate_replay::ManualCorporateReplay::new(
first,
self.cash_dividends_enabled,
self.cash_dividend_adjusts_cost_basis,
self.broker.matching_type(),
self.broker.execution_price_field(),
self.broker.same_day_buy_close_mark_at_fill(),
)
})
} else {
None
};
if let (Some(first), Some(observed)) = (
execution_dates.first(),
manual_cursor
.as_ref()
.and_then(|cursor| cursor.next_observation_at()),
) {
if observed
.with_timezone(&chrono::FixedOffset::east_opt(8 * 3600).unwrap())
.date_naive()
< *first
{
return Err(BacktestError::Execution(
"manual observations precede the declared initial portfolio period".into(),
));
}
}
let mut result = BacktestResult {
@@ -3060,6 +3151,9 @@ where
&mut portfolio,
&mut corporate_action_notes,
)?;
if let Some(replay) = &self.manual_corporate_replay {
replay.record_session(execution_date);
}
self.extend_result(
&mut result,
receivable_report,
@@ -3116,6 +3210,9 @@ where
PriceField::Close,
self.broker.same_day_buy_close_mark_at_fill(),
)?;
if let Some(replay) = &self.manual_corporate_replay {
replay.record_close(execution_date);
}
let close_report = self.broker.after_trading(execution_date);
merge_broker_report(&mut report, close_report);
let futures_daily_settlement_report = self.settle_futures_daily(execution_date);
@@ -4139,6 +4236,9 @@ where
PriceField::Close,
self.broker.same_day_buy_close_mark_at_fill(),
)?;
if let Some(replay) = &self.manual_corporate_replay {
replay.record_close(execution_date);
}
let post_trade_open_orders = self.open_order_views();
let mut broker_diagnostics = std::mem::take(&mut report.diagnostics);
@@ -4566,150 +4666,14 @@ where
portfolio: &mut PortfolioState,
notes: &mut Vec<String>,
) -> Result<BrokerExecutionReport, BacktestError> {
let mut report = BrokerExecutionReport::default();
for action in self.data.corporate_actions_on(date) {
if !action.has_effect() {
continue;
}
let Some(existing_position) = portfolio.position(&action.symbol) else {
continue;
};
if existing_position.quantity == 0 {
continue;
}
if self.cash_dividends_enabled && action.share_cash.abs() > f64::EPSILON {
let cash_before = portfolio.cash();
let (cash_delta, quantity_after, average_cost) = {
let position = portfolio
.position_mut_if_exists(&action.symbol)
.expect("position exists for dividend action");
let cash_delta = if self.cash_dividend_adjusts_cost_basis {
position.apply_cash_dividend(action.share_cash)
} else {
position.apply_cash_dividend_preserve_cost_basis(action.share_cash)
};
(cash_delta, position.quantity, position.average_cost)
};
if cash_delta.abs() > f64::EPSILON {
let payable_date = action.payable_date.unwrap_or(date);
portfolio.add_cash_receivable(CashReceivable {
symbol: action.symbol.clone(),
ex_date: date,
payable_date,
amount: cash_delta,
reason: format!("cash_dividend {:.6}", action.share_cash),
});
let note = format!(
"cash_dividend_receivable {} share_cash={:.6} quantity={} payable_date={} cash={:.2}",
action.symbol, action.share_cash, quantity_after, payable_date, cash_delta
);
notes.push(note.clone());
report.account_events.push(AccountEvent {
date,
cash_before,
cash_after: portfolio.cash(),
total_equity: portfolio.total_equity(),
note,
});
report.position_events.push(PositionEvent {
date,
symbol: action.symbol.clone(),
delta_quantity: 0,
quantity_after,
average_cost,
realized_pnl_delta: 0.0,
reason: format!("cash_dividend {:.6}", action.share_cash),
});
}
}
let split_ratio = action.split_ratio();
if (split_ratio - 1.0).abs() > f64::EPSILON {
portfolio.adjust_stock_pool_split(&action.symbol, split_ratio)
.map_err(BacktestError::Execution)?;
let (delta_quantity, quantity_after, average_cost) = {
let position = portfolio
.position_mut_if_exists(&action.symbol)
.expect("position exists for split action");
let delta_quantity = position.apply_split_ratio(split_ratio);
(delta_quantity, position.quantity, position.average_cost)
};
if delta_quantity != 0 {
let note = format!(
"stock_split {} ratio={:.6} delta_qty={}",
action.symbol, split_ratio, delta_quantity
);
notes.push(note);
report.position_events.push(PositionEvent {
date,
symbol: action.symbol.clone(),
delta_quantity,
quantity_after,
average_cost,
realized_pnl_delta: 0.0,
reason: format!("stock_split {:.6}", split_ratio),
});
}
}
if action.has_successor_conversion() {
let successor_symbol = action
.successor_symbol
.as_deref()
.expect("successor symbol checked");
let Some(outcome) = portfolio.apply_successor_conversion(
&action.symbol,
successor_symbol,
action.successor_ratio_value(),
action.successor_cash_value(),
) else {
continue;
};
let reason = format!(
"successor_conversion {}->{} ratio={:.6} cash_per_share={:.6}",
outcome.old_symbol,
outcome.new_symbol,
action.successor_ratio_value(),
action.successor_cash_value()
);
notes.push(reason.clone());
report.position_events.push(PositionEvent {
date,
symbol: outcome.old_symbol.clone(),
delta_quantity: -(outcome.old_quantity as i32),
quantity_after: 0,
average_cost: 0.0,
realized_pnl_delta: 0.0,
reason: reason.clone(),
});
report.position_events.push(PositionEvent {
date,
symbol: outcome.new_symbol.clone(),
delta_quantity: outcome.new_quantity_delta,
quantity_after: outcome.new_quantity_after,
average_cost: outcome.new_average_cost_after,
realized_pnl_delta: 0.0,
reason: reason.clone(),
});
if outcome.cash_delta.abs() > f64::EPSILON {
let cash_before = portfolio.cash();
portfolio
.apply_cash_delta(outcome.cash_delta)
.map_err(BacktestError::Execution)?;
report.account_events.push(AccountEvent {
date,
cash_before,
cash_after: portfolio.cash(),
total_equity: portfolio.total_equity(),
note: format!("{} cash={:.2}", reason, outcome.cash_delta),
});
}
}
}
portfolio.prune_flat_positions();
Ok(report)
crate::corporate_book::apply(
date,
&self.data,
portfolio,
notes,
self.cash_dividends_enabled,
self.cash_dividend_adjusts_cost_basis,
)
}
fn settle_cash_receivables(
@@ -4846,10 +4810,20 @@ where
notes: &mut Vec<String>,
) -> Result<BrokerExecutionReport, BacktestError> {
let mut report = BrokerExecutionReport::default();
for flow in portfolio
let fixed_before = portfolio.cash_fixed();
let flows = portfolio
.settle_pending_cash_flows(date)
.map_err(BacktestError::Execution)?
{
.map_err(BacktestError::Execution)?;
if let Some(replay) = &self.manual_corporate_replay {
replay
.record_cash(
date.and_hms_opt(0, 0, 0),
fixed_before,
portfolio.cash_fixed(),
)
.map_err(BacktestError::Execution)?;
}
for flow in flows {
let cash_before = portfolio.cash() - flow.amount;
let note = format!(
"deposit_withdraw_settled amount={:.2} payable_date={} reason={}",
@@ -4989,9 +4963,19 @@ where
}
let cash_before = portfolio.cash();
let fixed_before = portfolio.cash_fixed();
portfolio
.apply_management_fee(fee)
.map_err(BacktestError::Execution)?;
if let Some(replay) = &self.manual_corporate_replay {
replay
.record_cash(
stage_datetime(execution_date, callback_time),
fixed_before,
portfolio.cash_fixed(),
)
.map_err(BacktestError::Execution)?;
}
let mut report = BrokerExecutionReport::default();
report.account_events.push(AccountEvent {
date: execution_date,
crates/fidc-core/src/lib.rs
+2
View File
@@ -1,6 +1,7 @@
pub mod broker;
pub mod calendar;
pub mod cost;
mod corporate_book;
pub mod data;
mod numeric_factors;
pub mod daily_patterns;
@@ -22,6 +23,7 @@ pub mod futures;
pub mod instrument;
pub mod metrics;
pub mod manual_execution;
mod manual_corporate_replay;
mod numeric_expr_vm;
pub mod platform_expr_strategy;
pub mod platform_runtime_schema;
crates/fidc-core/src/manual_corporate_replay.rs
+431
View File
@@ -0,0 +1,431 @@
use std::cell::{Cell, RefCell};
use std::collections::{BTreeMap, BTreeSet};
use chrono::{DateTime, FixedOffset, NaiveDate, NaiveDateTime, Utc};
use sha2::{Digest, Sha256};
use crate::manual_execution::{
AppliedManualFill, ManualCorporateActionReference, ManualCorporateAdjustment,
ManualCorporatePositionChange, ManualExecutionReplay, ManualFillObservation,
};
use crate::{DataSet, FillEvent, FixedMoney, MatchingType, OrderSide, PortfolioState, PriceField};
#[derive(Debug, Clone)]
struct CashEffect {
at: NaiveDateTime,
amount: FixedMoney,
}
/// Replays issued financial facts, never strategy callbacks or hypothetical orders.
/// The ordinary book remains observable until an actual receipt is delivered.
pub(crate) struct ManualCorporateReplay {
first_date: NaiveDate,
sessions: RefCell<BTreeSet<NaiveDate>>,
closed: RefCell<BTreeSet<NaiveDate>>,
cash: RefCell<Vec<CashEffect>>,
reconciled_count: Cell<usize>,
cash_dividends: bool,
adjust_cost: bool,
matching: MatchingType,
daily_price: PriceField,
same_day_mark_at_fill: bool,
}
impl ManualCorporateReplay {
pub(crate) fn new(
first_date: NaiveDate,
cash_dividends: bool,
adjust_cost: bool,
matching: MatchingType,
daily_price: PriceField,
same_day_mark_at_fill: bool,
) -> Self {
Self {
first_date,
sessions: RefCell::new(BTreeSet::new()),
closed: RefCell::new(BTreeSet::new()),
cash: RefCell::new(Vec::new()),
reconciled_count: Cell::new(0),
cash_dividends,
adjust_cost,
matching,
daily_price,
same_day_mark_at_fill,
}
}
pub(crate) fn record_session(&self, date: NaiveDate) {
self.sessions.borrow_mut().insert(date);
}
pub(crate) fn record_close(&self, date: NaiveDate) {
self.closed.borrow_mut().insert(date);
}
pub(crate) fn committed(&self, count: usize) {
self.reconciled_count.set(count);
}
pub(crate) fn record_cash(
&self,
at: Option<NaiveDateTime>,
before: FixedMoney,
after: FixedMoney,
) -> Result<(), String> {
let amount = after
.checked_sub(before)
.ok_or("manual corporate cash observation overflow")?;
if amount != FixedMoney::ZERO {
self.cash.borrow_mut().push(CashEffect {
at: at.ok_or("manual corporate cash observation has no execution clock")?,
amount,
});
}
Ok(())
}
fn action_references(
&self,
observation: ManualFillObservation<'_>,
data: &DataSet,
) -> Result<Vec<ManualCorporateActionReference>, String> {
let mut symbols = BTreeSet::from([observation.order.symbol.clone()]);
let observed_date = local(observation.fill.observed_at).date();
let mut actions = Vec::new();
for date in self.sessions.borrow().range((
std::ops::Bound::Excluded(observation.fill.trade_date),
std::ops::Bound::Included(observed_date),
)) {
for action in data.corporate_actions_on(*date) {
if !symbols.contains(&action.symbol) {
continue;
}
crate::finite_serialization::validate(action).map_err(|error| error.to_string())?;
let effective = (action.split_ratio() - 1.).abs() > f64::EPSILON
|| action.has_successor_conversion()
|| (self.cash_dividends && action.share_cash.abs() > f64::EPSILON);
if !effective {
continue;
}
if let Some(successor) = action
.successor_symbol
.as_ref()
.filter(|_| action.has_successor_conversion())
{
if data.instrument(successor).is_none() {
return Err(format!(
"manual corporate successor is absent from frozen source data: symbol={successor} action_date={date}"
));
}
symbols.insert(successor.clone());
}
actions.push(ManualCorporateActionReference {
date: *date,
symbol: action.symbol.clone(),
successor_symbol: action.successor_symbol.clone(),
share_cash: action.share_cash.to_string(),
split_ratio: action.split_ratio().to_string(),
successor_ratio: action.successor_ratio.map(|value| value.to_string()),
successor_cash: action.successor_cash.map(|value| value.to_string()),
sha256: digest(
&serde_json::to_value(action).map_err(|error| error.to_string())?,
)?,
});
}
}
Ok(actions)
}
pub(crate) fn required(
&self,
observation: ManualFillObservation<'_>,
data: &DataSet,
) -> Result<bool, String> {
Ok(!self.action_references(observation, data)?.is_empty())
}
pub(crate) fn project(
&self,
source: &ManualExecutionReplay,
applied_count: usize,
observation: ManualFillObservation<'_>,
current: &mut PortfolioState,
data: &DataSet,
fills: &[FillEvent],
has_pending: bool,
) -> Result<(AppliedManualFill, ManualCorporateAdjustment), String> {
if has_pending {
return Err("manual observation conflicts with pending shadow orders".into());
}
let all = source.observations()?;
if all.get(applied_count).is_none_or(|next| {
next.fill.observation_event_id != observation.fill.observation_event_id
}) {
return Err(
"manual corporate observation prefix differs from the immutable trace".into(),
);
}
let actions = self.action_references(observation, data)?;
if actions.is_empty() {
return Err("manual corporate projection has no processed corporate action".into());
}
let at = local(observation.fill.observed_at);
let reference = self.replay(
current.initial_cash_fixed(),
&all[..applied_count],
self.reconciled_count.get(),
data,
fills,
at,
)?;
let expected = current.financial_replay_identity();
let reference_identity = reference.financial_replay_identity();
if reference_identity != expected {
return Err(format!(
"manual corporate ledger coverage mismatch: symbol={} observed_at={} expected={} replayed={}",
observation.order.symbol,
observation.fill.observed_at,
digest(&expected)?,
digest(&reference_identity)?
));
}
let replayed = self.replay(
current.initial_cash_fixed(),
&all[..=applied_count],
applied_count + 1,
data,
fills,
at,
)?;
let replayed_identity = replayed.financial_replay_identity();
let gross = FixedMoney::from_decimal_str(&observation.fill.gross_amount()?.to_string())?;
let fees = FixedMoney::from_decimal_str(&observation.fill.total_fees()?.to_string())?;
let cash_delta = match observation.order.side {
OrderSide::Buy => gross.checked_add(fees).and_then(FixedMoney::checked_neg),
OrderSide::Sell => gross.checked_sub(fees),
}
.ok_or("manual corporate trade cash overflow")?;
let before = current.cash_fixed();
let after = replayed.cash_fixed();
let corporate_cash = after
.checked_sub(before)
.and_then(|delta| delta.checked_sub(cash_delta))
.ok_or("manual corporate adjustment overflow")?;
let symbols = current
.positions()
.keys()
.chain(replayed.positions().keys())
.cloned()
.chain(std::iter::once(observation.order.symbol.clone()))
.collect::<BTreeSet<_>>();
let mut positions = BTreeMap::new();
for symbol in symbols {
let change = ManualCorporatePositionChange {
quantity_before: current
.position(&symbol)
.map_or(0, |position| position.quantity),
quantity_after: replayed
.position(&symbol)
.map_or(0, |position| position.quantity),
cost_basis_before: current
.financial_position_basis(&symbol)
.to_decimal_string(),
cost_basis_after: replayed
.financial_position_basis(&symbol)
.to_decimal_string(),
};
if change.quantity_before != change.quantity_after
|| change.cost_basis_before != change.cost_basis_after
{
positions.insert(symbol, change);
}
}
let quantity_after = replayed
.position(&observation.order.symbol)
.map_or(0, |position| position.quantity);
let adjustment = ManualCorporateAdjustment {
schema: "fidc.manual-corporate-adjustment/v1".into(),
observed_at: observation.fill.observed_at,
cash_dividends_enabled: self.cash_dividends,
dividend_cost_basis_adjustment: self.adjust_cost,
actions,
cash_before: before.to_decimal_string(),
cash_after: after.to_decimal_string(),
corporate_cash_delta: corporate_cash.to_decimal_string(),
positions,
reference_sha256: digest(&reference_identity)?,
replayed_sha256: digest(&replayed_identity)?,
};
current.replace_replayed_financial_book(replayed)?;
Ok((
AppliedManualFill {
gross,
fees,
cash_delta,
quantity_after,
},
adjustment,
))
}
fn replay(
&self,
initial_cash: FixedMoney,
manual: &[ManualFillObservation<'_>],
economic_count: usize,
data: &DataSet,
fills: &[FillEvent],
at: NaiveDateTime,
) -> Result<PortfolioState, String> {
enum Event<'a> {
Session,
Cash(&'a CashEffect),
Corporate(NaiveDate),
Settle(NaiveDate),
Manual(ManualFillObservation<'a>),
Simulated(&'a FillEvent),
Close(NaiveDate),
}
let mut events = Vec::new();
let sessions = self.sessions.borrow();
let closed = self.closed.borrow();
let cash = self.cash.borrow();
for date in &*sessions {
let clock = date.and_hms_opt(0, 0, 0).unwrap();
events.push((clock, 0, 0, Event::Session));
events.push((clock, 2, 0, Event::Corporate(*date)));
events.push((clock, 3, 0, Event::Settle(*date)));
}
for (index, effect) in cash.iter().enumerate() {
events.push((effect.at, 1, index, Event::Cash(effect)));
}
for (index, observation) in manual.iter().enumerate() {
let clock = if index < economic_count {
local(observation.fill.executed_at)
} else {
local(observation.fill.observed_at)
};
if clock.date() < self.first_date {
return Err("manual corporate execution precedes the represented initial ledger; opening facts are required".into());
}
events.push((clock, 4, fills.len() + index, Event::Manual(*observation)));
}
for (index, fill) in fills.iter().enumerate() {
let date = fill.execution_date.unwrap_or(fill.date);
// This is the frozen daily matching model, not a broker timestamp.
// Intraday contracts must supply their actual execution clock.
let clock = match fill.execution_timestamp {
Some(clock) => clock,
None if matches!(
self.matching,
MatchingType::OpenAuction | MatchingType::NextBarOpen
) =>
{
date.and_hms_opt(9, 30, 0).unwrap()
}
None if self.matching == MatchingType::CurrentBarClose
&& self.daily_price == PriceField::Close =>
{
date.and_hms_opt(15, 0, 0).unwrap()
}
None => {
return Err(
"manual corporate replay lacks an intraday fill execution clock".into(),
);
}
};
events.push((clock, 4, index, Event::Simulated(fill)));
}
for date in &*closed {
events.push((
date.and_hms_nano_opt(23, 59, 59, 999_999_999)
.unwrap()
.min(at),
5,
0,
Event::Close(*date),
));
}
events.sort_by_key(|(clock, priority, sequence, _)| (*clock, *priority, *sequence));
let mut book = PortfolioState::from_fixed_initial_cash(initial_cash);
for (clock, _, _, event) in events {
if clock > at {
return Err("manual corporate replay contains a future financial fact".into());
}
match event {
Event::Session => book.begin_trading_day(),
Event::Cash(effect) => {
book.apply_cash_delta_fixed(effect.amount)?;
if book.cash_fixed() < FixedMoney::ZERO {
return Err(
"manual corporate replay conflicts with prior cash facts".into()
);
}
}
Event::Corporate(date) => {
crate::corporate_book::apply(
date,
data,
&mut book,
&mut Vec::new(),
self.cash_dividends,
self.adjust_cost,
)
.map_err(|error| error.to_string())?;
}
Event::Settle(date) => {
for receivable in book.take_due_cash_receivables(date) {
book.settle_cash_receivable(&receivable)?;
}
}
Event::Manual(observation) => {
observation.apply(&mut book, data, false)?;
}
Event::Simulated(fill) => {
let gross = FixedMoney::from_f64(fill.gross_amount)
.ok_or("invalid simulated gross amount")?;
let fees = FixedMoney::checked_sum_f64([
fill.commission,
fill.stamp_tax,
fill.transfer_fee,
])
.ok_or("invalid simulated fee amount")?;
book.apply_observed_manual_fill(
fill.execution_date.unwrap_or(fill.date),
&fill.symbol,
fill.side,
fill.quantity,
fill.price,
fill.price,
gross,
fees,
)?;
book.prune_flat_positions();
}
Event::Close(date) => {
book.update_prices_with_options(
date,
data,
PriceField::Close,
self.same_day_mark_at_fill,
)
.map_err(|error| error.to_string())?;
}
}
if book.cash_fixed() < FixedMoney::ZERO {
return Err("manual corporate replay would borrow unobserved cash".into());
}
}
Ok(book)
}
}
fn local(value: DateTime<Utc>) -> NaiveDateTime {
value
.with_timezone(&FixedOffset::east_opt(8 * 3600).unwrap())
.naive_local()
}
fn digest(value: &serde_json::Value) -> Result<String, String> {
Ok(format!(
"{:x}",
Sha256::digest(serde_json::to_vec(value).map_err(|error| error.to_string())?)
))
}
crates/fidc-core/src/manual_execution.rs
+87
View File
@@ -482,9 +482,80 @@ pub struct ManualReplayApplication {
pub ledger_gross_amount: String,
pub ledger_fees: String,
pub cash_delta: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub corporate_adjustment: Option<ManualCorporateAdjustment>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct ManualCorporateAdjustment {
pub schema: String,
pub observed_at: DateTime<Utc>,
pub cash_dividends_enabled: bool,
pub dividend_cost_basis_adjustment: bool,
pub actions: Vec<ManualCorporateActionReference>,
pub cash_before: String,
pub cash_after: String,
pub corporate_cash_delta: String,
pub positions: BTreeMap<String, ManualCorporatePositionChange>,
pub reference_sha256: String,
pub replayed_sha256: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct ManualCorporateActionReference {
pub date: NaiveDate,
pub symbol: String,
pub successor_symbol: Option<String>,
pub share_cash: String,
pub split_ratio: String,
pub successor_ratio: Option<String>,
pub successor_cash: Option<String>,
pub sha256: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(deny_unknown_fields, rename_all = "camelCase")]
pub struct ManualCorporatePositionChange {
pub quantity_before: u32,
pub quantity_after: u32,
pub cost_basis_before: String,
pub cost_basis_after: String,
}
impl ManualReplayCursor {
pub(crate) fn frozen_source(&self) -> std::sync::Arc<ManualExecutionReplay> {
self.replay.clone()
}
pub(crate) fn next_observation(&self) -> Option<ManualFillObservation<'_>> {
self.indices.get(self.cursor).map(|&(a, o, f)| ManualFillObservation {
action: &self.replay.actions[a], order: &self.replay.actions[a].orders[o],
fill: &self.replay.actions[a].orders[o].fills[f],
})
}
pub(crate) fn advance_next_projected<F>(
&mut self, portfolio: &mut PortfolioState, project: F,
) -> Result<Option<ManualReplayApplication>, String>
where F: FnOnce(ManualFillObservation<'_>, &mut PortfolioState) -> Result<(AppliedManualFill, ManualCorporateAdjustment), String> {
let Some(observation) = self.next_observation() else { return Ok(None); };
let at = observation.fill.observed_at;
if at > self.replay.observation_cutoff || self.clock.is_some_and(|clock| at < clock) {
return Err("manual projected observation clock violates the frozen trace".into());
}
let mut next = portfolio.clone();
let (applied, adjustment) = project(observation, &mut next)?;
let mut application = observation.application(applied)?;
application.corporate_adjustment = Some(adjustment);
crate::finite_serialization::validate(&application).map_err(|error| error.to_string())?;
*portfolio = next;
self.cursor += 1;
self.clock = Some(at);
Ok(Some(application))
}
pub fn new(replay: ManualExecutionReplay) -> Result<Self, String> {
Self::from_shared(std::sync::Arc::new(replay))
}
@@ -594,6 +665,7 @@ impl ManualReplayCursor {
ledger_gross_amount: applied.gross.to_decimal_string(),
ledger_fees: applied.fees.to_decimal_string(),
cash_delta: applied.cash_delta.to_decimal_string(),
corporate_adjustment: None,
});
}
*portfolio = next;
@@ -604,6 +676,21 @@ impl ManualReplayCursor {
}
impl ManualFillObservation<'_> {
fn application(&self, applied: AppliedManualFill) -> Result<ManualReplayApplication, String> {
Ok(ManualReplayApplication {
action_id: self.action.action_id.clone(), order_id: self.order.order_id.clone(),
trade_id: self.fill.trade_id.clone(), observation_event_id: self.fill.observation_event_id.clone(),
observation_sequence: self.fill.observation_sequence, observed_at: self.fill.observed_at,
fee_observation_event_id: self.fill.fee_observation_event_id.clone(), fee_observed_at: self.fill.fee_observed_at,
executed_at: self.fill.executed_at, symbol: self.order.symbol.clone(), side: self.order.side,
quantity: self.fill.quantity, quantity_after: applied.quantity_after, price: self.fill.price.to_string(),
commission: self.fill.commission.map(|fee| fee.to_string()), stamp_tax: self.fill.stamp_tax.map(|fee| fee.to_string()),
transfer_fee: self.fill.transfer_fee.map(|fee| fee.to_string()), source_total_fee: self.fill.total_fee.to_string(),
source_gross_amount: self.fill.gross_amount()?.to_string(), ledger_gross_amount: applied.gross.to_decimal_string(),
ledger_fees: applied.fees.to_decimal_string(), cash_delta: applied.cash_delta.to_decimal_string(), corporate_adjustment: None,
})
}
pub(crate) fn apply(
&self,
portfolio: &mut PortfolioState,
crates/fidc-core/src/manual_execution/tests.rs
+15
View File
@@ -24,6 +24,21 @@ fn reseal(input: &mut ManualExecutionReplay) {
input.content_sha256 = input.content_digest().unwrap();
}
#[test]
fn a_failed_corporate_projection_does_not_change_the_book_or_receipt_cursor() {
let mut cursor = ManualReplayCursor::new(sample()).unwrap();
let mut portfolio = PortfolioState::new(10000.);
let before = portfolio.financial_replay_identity();
let error = cursor.advance_next_projected(&mut portfolio, |_, next| {
next.apply_cash_delta(-50.)?;
next.position_mut("000001.SZ").buy(NaiveDate::from_ymd_opt(2026, 9, 11).unwrap(), 100, 10.);
Err("financial coverage mismatch".into())
}).unwrap_err();
assert_eq!(error, "financial coverage mismatch");
assert_eq!(portfolio.financial_replay_identity(), before);
assert_eq!(cursor.applied_count(), 0);
}
fn delayed_buy_replay() -> ManualExecutionReplay {
let mut input = sample();
let template = input.actions[0].clone();
crates/fidc-core/src/platform_expr_strategy.rs
+5
View File
@@ -12470,6 +12470,11 @@ impl Strategy for PlatformExprStrategy {
.entry(execution.symbol.clone())
.and_modify(|previous| *previous = (*previous).max(date))
.or_insert(date);
if let Some(adjustment) = &execution.corporate_adjustment {
for successor in adjustment.actions.iter().filter_map(|action| action.successor_symbol.as_ref()) {
history.entry(successor.clone()).and_modify(|previous| *previous = (*previous).max(date)).or_insert(date);
}
}
Ok(())
}
fn name(&self) -> &str {
crates/fidc-core/src/portfolio.rs
+51 -1
View File
@@ -734,6 +734,10 @@ impl PortfolioState {
pub fn new(initial_cash: f64) -> Self {
let initial_cash = fixed_money(initial_cash, "initial cash")
.expect("initial cash must be finite fixed-point money");
Self::from_fixed_initial_cash(initial_cash)
}
pub(crate) fn from_fixed_initial_cash(initial_cash: FixedMoney) -> Self {
Self {
initial_cash,
units: initial_cash,
@@ -756,6 +760,8 @@ impl PortfolioState {
self.initial_cash.to_f64()
}
pub(crate) fn initial_cash_fixed(&self) -> FixedMoney { self.initial_cash }
pub(crate) fn stock_pool_execution_state(&self,pool_id:&str)->crate::stock_pool_state::StockPoolExecutionState{
self.stock_pool_states.get(pool_id).cloned().unwrap_or_default()
}
@@ -787,6 +793,46 @@ impl PortfolioState {
self.cash.to_f64()
}
pub(crate) fn cash_fixed(&self) -> FixedMoney { self.cash }
pub(crate) fn financial_replay_identity(&self) -> serde_json::Value {
let positions = self.positions.iter().filter(|(_, position)| position.quantity > 0)
.map(|(symbol, position)| {
let mut lots = position.lots.iter().map(|lot| (lot.acquired_date, lot.quantity,
lot.entry_value.to_decimal_string(), lot.cost_basis.to_decimal_string())).collect::<Vec<_>>();
lots.sort();
(symbol.clone(), serde_json::json!({"quantity":position.quantity,"lots":lots,
"openedDate":position.opened_date,"lastBuyDate":position.last_buy_date}))
}).collect::<std::collections::BTreeMap<_, _>>();
let mut receivables = self.cash_receivables.iter().map(|row| (row.symbol.clone(), row.ex_date,
row.payable_date, fixed_money_or_panic(row.amount, "receivable identity").to_decimal_string(), row.reason.clone())).collect::<Vec<_>>();
receivables.sort();
serde_json::json!({"cash":self.cash.to_decimal_string(),"positions":positions,"receivables":receivables})
}
pub(crate) fn financial_position_basis(&self, symbol: &str) -> FixedMoney {
self.positions.get(symbol).map_or(FixedMoney::ZERO, Position::total_cost_basis)
}
pub(crate) fn replace_replayed_financial_book(&mut self, mut replayed: PortfolioState) -> Result<(), String> {
if replayed.cash < FixedMoney::ZERO || replayed.initial_cash != self.initial_cash {
return Err("manual corporate replay changed initial capital or borrowed cash".into());
}
for (symbol, position) in &mut replayed.positions {
if let Some(current) = self.positions.get(symbol).filter(|current| current.quantity > 0) {
position.last_price = current.last_price;
position.refresh_day_pnl();
}
}
self.cash = replayed.cash;
self.positions = replayed.positions;
self.cash_receivables = replayed.cash_receivables;
self.day_sold_symbols = replayed.day_sold_symbols;
// Existing issued units, explicit cash-flow/financing facts, and task
// target state are observed controls, not counterfactual new orders.
Ok(())
}
pub fn external_cash_flow_total(&self) -> f64 {
self.external_cash_flow_total.to_f64()
}
@@ -822,9 +868,13 @@ impl PortfolioState {
}
pub fn apply_cash_delta(&mut self, delta: f64) -> Result<(), String> {
self.apply_cash_delta_fixed(fixed_money(delta, "cash delta")?)
}
pub(crate) fn apply_cash_delta_fixed(&mut self, delta: FixedMoney) -> Result<(), String> {
self.cash = self
.cash
.checked_add(fixed_money(delta, "cash delta")?)
.checked_add(delta)
.ok_or_else(|| "fixed-point cash overflow".to_string())?;
Ok(())
}