From 695fdee4b8ba4b456313f06415c128a828b43f56 Mon Sep 17 00:00:00 2001 From: boris Date: Tue, 15 Sep 2026 01:28:38 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=9D=E7=95=99=E6=A8=A1=E6=8B=9F=E5=99=A8?= =?UTF-8?q?=E5=A4=B1=E8=B4=A5=E8=B0=83=E7=94=A8=E5=89=8D=E7=9A=84=E5=A7=94?= =?UTF-8?q?=E6=89=98=E4=B8=8E=E6=89=A7=E8=A1=8C=E7=8A=B6=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/fidc-core/src/broker.rs | 180 +++++- .../src/broker_order_recovery_tests.rs | 515 ++++++++++++++++++ crates/fidc-core/src/broker_stock_pool.rs | 7 +- crates/fidc-core/src/etf_execution.rs | 2 +- crates/fidc-core/src/portfolio.rs | 73 +++ docs/simulator-order-recovery-20260915.md | 29 + 6 files changed, 776 insertions(+), 30 deletions(-) create mode 100644 crates/fidc-core/src/broker_order_recovery_tests.rs create mode 100644 docs/simulator-order-recovery-20260915.md diff --git a/crates/fidc-core/src/broker.rs b/crates/fidc-core/src/broker.rs index 9c25d75..a6fdc05 100644 --- a/crates/fidc-core/src/broker.rs +++ b/crates/fidc-core/src/broker.rs @@ -100,7 +100,7 @@ struct QuoteLiquidityConsumption { quantity: u32, } -#[derive(Debug, Default)] +#[derive(Debug, Default, Clone)] struct IntradayExecutionLedger { cursors: BTreeMap, depth_consumption: BTreeMap; 2]>, @@ -235,7 +235,7 @@ enum BrokerCallbackPhase { BeforeStrategy, } -#[derive(Debug, Default)] +#[derive(Debug, Default, Clone)] struct BrokerExecutionSession { date: Option, intraday_turnover: BTreeMap, @@ -441,6 +441,43 @@ impl Drop for RestoreCell<'_, T> { fn drop(&mut self) { self.0.set(self.1); } } +struct RestoreRefCell<'a, T>(&'a RefCell, Option); +impl Drop for RestoreRefCell<'_, T> { + fn drop(&mut self) { + if let Some(value) = self.1.take() { self.0.replace(value); } + } +} + +macro_rules! execution_context_checkpoint { + ($($field:ident : $kind:ty),* $(,)?) => { + struct BrokerExecutionContext { $($field: $kind),* } + impl BrokerExecutionContext { + fn capture(broker: &BrokerSimulator) -> Self { + Self { $($field: broker.$field.get()),* } + } + fn restore(self, broker: &BrokerSimulator) { + $(broker.$field.set(self.$field);)* + } + } + }; +} +execution_context_checkpoint! { + runtime_etf_daily_open: bool, + runtime_stock_pool_followup: bool, + runtime_intraday_start_time: Option, + runtime_intraday_end_time: Option, + runtime_execution_clock: Option, + runtime_callback_phase: BrokerCallbackPhase, + runtime_algo_schedule: Option, + runtime_unprocessed_algorithm_cash: FixedMoney, + runtime_decision_date: Option, + runtime_order_created_date: Option, + runtime_resting_order_origin: Option, + runtime_decision_total_equity: Option, + runtime_target_position_limit: Option, + runtime_time_in_force: Option, +} + pub struct BrokerSimulator { historical_etf_open_fallback: bool, verified_etf_minute_absences: RefCell>, @@ -485,9 +522,96 @@ pub struct BrokerSimulator { next_order_id: Cell, open_orders: RefCell>, execution_session: RefCell, + execution_transaction_depth: Cell, +} + +/// Only unpublished simulator state is transactional. Broker observations and +/// results returned successfully by earlier calls are outside this checkpoint. +struct BrokerExecutionCheckpoint { + portfolio: crate::portfolio::PortfolioCheckpoint, + orders: Vec, + etf_targets: crate::etf_execution::DeferredEtfTargets, + pool_targets: BTreeMap, + sold: BTreeMap>, + session: BrokerExecutionSession, + next_order_id: u64, + context: BrokerExecutionContext, +} + +impl BrokerExecutionCheckpoint { + fn capture(broker: &BrokerSimulator, portfolio: &PortfolioState, symbols: Option<&BTreeSet>) -> Self { + Self { + portfolio: portfolio.checkpoint(symbols), orders: broker.open_orders.borrow().clone(), + etf_targets: broker.deferred_etf_targets.borrow().clone(), + pool_targets: broker.deferred_stock_pools.borrow().clone(), + sold: broker.same_day_sold_symbols.borrow().clone(), + session: broker.execution_session.borrow().clone(), + next_order_id: broker.next_order_id.get(), + context: BrokerExecutionContext::capture(broker), + } + } + fn restore(self, broker: &BrokerSimulator, portfolio: &mut PortfolioState) { + self.portfolio.restore(portfolio); + *broker.open_orders.borrow_mut() = self.orders; + *broker.deferred_etf_targets.borrow_mut() = self.etf_targets; + *broker.deferred_stock_pools.borrow_mut() = self.pool_targets; + *broker.same_day_sold_symbols.borrow_mut() = self.sold; + *broker.execution_session.borrow_mut() = self.session; + broker.next_order_id.set(self.next_order_id); + self.context.restore(broker); + } +} + +struct BrokerExecutionTransaction<'a, C, R> { + broker: &'a BrokerSimulator, + portfolio: &'a mut PortfolioState, + checkpoint: Option, +} + +impl Drop for BrokerExecutionTransaction<'_, C, R> { + fn drop(&mut self) { + if let Some(checkpoint) = self.checkpoint.take() { + checkpoint.restore(self.broker, self.portfolio); + } + } } impl BrokerSimulator { + fn execution_transaction(&self, portfolio: &mut PortfolioState, needed: bool, symbols: Option<&BTreeSet>, execute: F) + -> Result + where F: FnOnce(&mut PortfolioState) -> Result { + if !needed || self.execution_transaction_depth.get() > 0 { return execute(portfolio); } + let checkpoint = BrokerExecutionCheckpoint::capture(self, portfolio, symbols); + let mut transaction = BrokerExecutionTransaction { broker: self, portfolio, checkpoint: Some(checkpoint) }; + let _depth = RestoreCell(&self.execution_transaction_depth, + self.execution_transaction_depth.replace(1)); + let result = execute(transaction.portfolio); + if result.is_ok() { transaction.checkpoint = None; } + result + } + + fn checkpoint_symbols(&self, decision: &StrategyDecision) -> Option> { + if decision.rebalance || !self.deferred_stock_pools.borrow().is_empty() { return None; } + let mut symbols = self.open_orders.borrow().iter().map(|order| order.symbol.clone()).collect::>(); + symbols.extend(decision.exit_symbols.iter().cloned()); + for intent in &decision.order_intents { + let symbol = match intent.unwrapped() { + OrderIntent::Shares { symbol, .. } | OrderIntent::LimitShares { symbol, .. } + | OrderIntent::Lots { symbol, .. } | OrderIntent::LimitLots { symbol, .. } + | OrderIntent::TargetShares { symbol, .. } | OrderIntent::LimitTargetShares { symbol, .. } + | OrderIntent::Value { symbol, .. } | OrderIntent::LimitValue { symbol, .. } + | OrderIntent::TargetValue { symbol, .. } | OrderIntent::LimitTargetValue { symbol, .. } + | OrderIntent::TimedTargetValue { symbol, .. } | OrderIntent::AlgoValue { symbol, .. } + | OrderIntent::Percent { symbol, .. } | OrderIntent::LimitPercent { symbol, .. } + | OrderIntent::TargetPercent { symbol, .. } | OrderIntent::LimitTargetPercent { symbol, .. } => symbol, + // Unknown/new/whole-portfolio controls must retain everything. + _ => return None, + }; + symbols.insert(symbol.clone()); + } + Some(symbols) + } + pub fn new(cost_model: C, rules: R) -> Self { Self { historical_etf_open_fallback: false, @@ -531,6 +655,7 @@ impl BrokerSimulator { runtime_target_position_limit: Cell::new(None), runtime_time_in_force: Cell::new(None), next_order_id: Cell::new(1), + execution_transaction_depth: Cell::new(0), open_orders: RefCell::new(Vec::new()), execution_session: RefCell::new(BrokerExecutionSession::default()), } @@ -583,6 +708,7 @@ impl BrokerSimulator { runtime_target_position_limit: Cell::new(None), runtime_time_in_force: Cell::new(None), next_order_id: Cell::new(1), + execution_transaction_depth: Cell::new(0), open_orders: RefCell::new(Vec::new()), execution_session: RefCell::new(BrokerExecutionSession::default()), } @@ -1602,30 +1728,22 @@ where data: &DataSet, decision: &StrategyDecision, ) -> Result { - let previous_decision_date = self.runtime_decision_date.get(); - let previous_buy_denials = self.runtime_buy_denials.replace(decision.buy_denials.clone()); + let _buy_denials = RestoreRefCell(&self.runtime_buy_denials, + Some(self.runtime_buy_denials.replace(decision.buy_denials.clone()))); let protection_denials = |scope| decision.risk_decisions.iter() .filter(|row| !row.accepted && row.stage == "automatic_trade_protection" && row.scope == scope) .map(|row| (row.symbol.clone(), row.reason.clone())).collect(); - let previous_auto_buy_denials = self.runtime_auto_buy_denials.replace(protection_denials(crate::risk_control::RiskCheckScope::Buy)); - let previous_auto_sell_denials = self.runtime_auto_sell_denials.replace(protection_denials(crate::risk_control::RiskCheckScope::Sell)); - let previous_order_created_date = self.runtime_order_created_date.get(); - let previous_decision_total_equity = self.runtime_decision_total_equity.get(); - self.runtime_decision_date.set(Some(decision_date)); - self.runtime_order_created_date - .set(Some(order_created_date)); - self.runtime_decision_total_equity - .set(decision_total_equity.filter(|equity| equity.is_finite() && *equity >= 0.0)); - let result = self.execute_with_runtime_dates(date, portfolio, data, decision); - self.runtime_buy_denials.replace(previous_buy_denials); - self.runtime_auto_buy_denials.replace(previous_auto_buy_denials); - self.runtime_auto_sell_denials.replace(previous_auto_sell_denials); - self.runtime_decision_date.set(previous_decision_date); - self.runtime_order_created_date - .set(previous_order_created_date); - self.runtime_decision_total_equity - .set(previous_decision_total_equity); - result + let _auto_buy = RestoreRefCell(&self.runtime_auto_buy_denials, Some(self.runtime_auto_buy_denials + .replace(protection_denials(crate::risk_control::RiskCheckScope::Buy)))); + let _auto_sell = RestoreRefCell(&self.runtime_auto_sell_denials, Some(self.runtime_auto_sell_denials + .replace(protection_denials(crate::risk_control::RiskCheckScope::Sell)))); + let _decision_date = RestoreCell(&self.runtime_decision_date, + self.runtime_decision_date.replace(Some(decision_date))); + let _created_date = RestoreCell(&self.runtime_order_created_date, + self.runtime_order_created_date.replace(Some(order_created_date))); + let _equity = RestoreCell(&self.runtime_decision_total_equity, + self.runtime_decision_total_equity.replace(decision_total_equity.filter(|equity| equity.is_finite() && *equity >= 0.0))); + self.execute_with_runtime_dates(date, portfolio, data, decision) } fn execute_with_runtime_dates( @@ -1638,11 +1756,16 @@ where if self.volume_limit { self.volume_rate.map_err(|error| BacktestError::Execution(error.to_string()))?; } - let mut session = std::mem::take(&mut *self.execution_session.borrow_mut()); - session.activate(date); - let result = self.execute_with_daily_session(date, portfolio, data, decision, &mut session); - *self.execution_session.borrow_mut() = session; - result + self.execution_session.borrow_mut().activate(date); + let may_execute = self.has_open_orders() || !self.deferred_stock_pools.borrow().is_empty() + || decision.rebalance || !decision.order_intents.is_empty() || !decision.exit_symbols.is_empty(); + let symbols = may_execute.then(|| self.checkpoint_symbols(decision)).flatten(); + self.execution_transaction(portfolio, may_execute, symbols.as_ref(), |portfolio| { + let mut session = std::mem::take(&mut *self.execution_session.borrow_mut()); + let result = self.execute_with_daily_session(date, portfolio, data, decision, &mut session); + *self.execution_session.borrow_mut() = session; + result + }) } fn execute_with_daily_session( @@ -8749,6 +8872,7 @@ mod tests { use crate::strategy::{AlgoOrderStyle, OrderIntent, OrderTimeInForce, StrategyDecision}; include!("broker_stock_pool_batch_tests.rs"); + include!("broker_order_recovery_tests.rs"); #[test] fn queued_order_retains_the_real_creation_clock_when_retried() { diff --git a/crates/fidc-core/src/broker_order_recovery_tests.rs b/crates/fidc-core/src/broker_order_recovery_tests.rs new file mode 100644 index 0000000..0c50317 --- /dev/null +++ b/crates/fidc-core/src/broker_order_recovery_tests.rs @@ -0,0 +1,515 @@ +// Kept inside broker::tests to inspect internal accepted-order identity as +// well as the public report. These are simulator states, never GT requests. +fn recovery_test_data(missing_previous: Option, intraday: bool) -> DataSet { + let previous = chrono::NaiveDate::from_ymd_opt(2025, 1, 2).unwrap(); + let date = chrono::NaiveDate::from_ymd_opt(2025, 1, 3).unwrap(); + let mut instruments = Vec::new(); + let mut rows = Vec::new(); + let mut candidates = Vec::new(); + let mut quotes = Vec::new(); + for index in 1..=2 { + let symbol = format!("{index:06}.SZ"); + let mut instrument = limit_test_instrument(); + instrument.symbol = symbol.clone(); + instruments.push(instrument); + for day in [previous, date] { + if day == previous && missing_previous == Some(index) { + continue; + } + let mut row = dated_limit_test_snapshot(day); + row.symbol = symbol.clone().into(); + rows.push(row); + let mut candidate = dated_limit_test_candidate(day, false, false, true, true); + candidate.symbol = symbol.clone().into(); + candidates.push(candidate); + } + if intraday { + let mut quote = limit_test_quote(10., 10., 10.); + quote.symbol = symbol; + quote.date = date; + quote.timestamp = date.and_hms_opt(9, 33, 0).unwrap(); + quotes.push(quote); + } + } + DataSet::from_components_with_actions_and_quotes( + instruments, + rows, + vec![], + candidates, + vec![ + dated_limit_test_benchmark(previous), + dated_limit_test_benchmark(date), + ], + vec![], + quotes, + ) + .unwrap() +} + +fn recovery_test_broker( + intraday: bool, + first_side: OrderSide, +) -> ( + BrokerSimulator, + PortfolioState, +) { + let mut broker = BrokerSimulator::new(ChinaAShareCostModel::default(), ChinaEquityRuleHooks) + .with_matching_type(if intraday { + MatchingType::MinuteLast + } else { + MatchingType::CurrentBarClose + }) + .with_volume_limit(false) + .with_liquidity_limit(false) + .with_slippage_model(SlippageModel::HistoricalVolumeVolatility( + super::DynamicSlippageConfig::new(0., 0., 0.1), + )); + if intraday { + broker = + broker.with_intraday_execution_start_time(NaiveTime::from_hms_opt(9, 33, 0).unwrap()); + } + let mut first = test_open_order(1); + first.filled_quantity = 100; + first.remaining_quantity = 100; + first.commission_remaining = Some(0.); + first.side = first_side; + let mut second = test_open_order(2); + second.symbol = "000002.SZ".into(); + broker.upsert_open_order(first); + broker.upsert_open_order(second); + broker.next_order_id.set(3); + let mut account = PortfolioState::new(9000.); + account.position_mut("000001.SZ").buy( + chrono::NaiveDate::from_ymd_opt(2025, 1, 2).unwrap(), + if first_side == OrderSide::Buy { + 100 + } else { + 200 + }, + 10., + ); + account.begin_trading_day(); + (broker, account) +} + +fn recovery_report_value(report: &BrokerExecutionReport) -> serde_json::Value { + serde_json::json!({"orders":report.order_events,"fills":report.fill_events, + "positions":report.position_events,"accounts":report.account_events, + "events":report.process_events,"diagnostics":report.diagnostics}) +} + +#[test] +fn failed_resting_order_batch_keeps_accepted_orders_and_unpublished_financial_state() { + let date = chrono::NaiveDate::from_ymd_opt(2025, 1, 3).unwrap(); + for intraday in [false, true] { + for first_side in [OrderSide::Buy, OrderSide::Sell] { + for missing in [1, 2] { + let (broker, mut account) = recovery_test_broker(intraday, first_side); + let orders = format!("{:?}", broker.open_orders.borrow()); + let ledger = account.financial_replay_identity(); + let error = broker + .execute( + date, + &mut account, + &recovery_test_data(Some(missing), intraday), + &StrategyDecision::default(), + ) + .unwrap_err(); + assert!( + error + .to_string() + .contains("historical_slippage_calibration_missing") + ); + assert_eq!( + format!("{:?}", broker.open_orders.borrow()), + orders, + "intraday={intraday} first={first_side:?} missing={missing}" + ); + assert_eq!(account.financial_replay_identity(), ledger); + assert!(broker.same_day_sold_symbols.borrow().is_empty()); + let recovered = broker + .execute( + date, + &mut account, + &recovery_test_data(None, intraday), + &StrategyDecision::default(), + ) + .unwrap(); + let (clean, mut clean_account) = recovery_test_broker(intraday, first_side); + let reference = clean + .execute( + date, + &mut clean_account, + &recovery_test_data(None, intraday), + &StrategyDecision::default(), + ) + .unwrap(); + assert_eq!( + recovery_report_value(&recovered), + recovery_report_value(&reference) + ); + assert_eq!( + account.financial_replay_identity(), + clean_account.financial_replay_identity() + ); + assert!(broker.open_order_views().is_empty()); + assert_eq!(recovered.fill_events.len(), 2); + } + } + } +} + +#[test] +fn failed_new_batch_does_not_erase_prior_success_or_double_charge_on_retry() { + let date = chrono::NaiveDate::from_ymd_opt(2025, 1, 3).unwrap(); + let (broker, mut account) = recovery_test_broker(true, OrderSide::Buy); + let good = recovery_test_data(None, true); + let prior = broker + .execute(date, &mut account, &good, &StrategyDecision::default()) + .unwrap(); + assert_eq!(prior.fill_events.len(), 2); + let initial = account.financial_replay_identity(); + let id = broker.next_order_id.get(); + let decision = StrategyDecision { + order_intents: vec![ + OrderIntent::Shares { + symbol: "000001.SZ".into(), + quantity: 100, + reason: "next-batch-a".into(), + }, + OrderIntent::Shares { + symbol: "000002.SZ".into(), + quantity: 100, + reason: "next-batch-b".into(), + }, + ], + ..Default::default() + }; + // A later quote lets this batch execute independently of the prior fills. + let mut parts = good.snapshot_components(); + for quote in &mut parts.execution_quotes { + quote.timestamp += chrono::Duration::minutes(1); + } + broker + .runtime_execution_clock + .set(Some(NaiveTime::from_hms_opt(9, 34, 0).unwrap())); + let restored = DataSet::from_components_with_actions_and_quotes( + parts.instruments.clone(), + parts.market.clone(), + parts.factors.clone(), + parts.candidates.clone(), + parts.benchmarks.clone(), + vec![], + parts.execution_quotes.clone(), + ) + .unwrap(); + parts + .market + .retain(|row| !(row.symbol.as_str() == "000002.SZ" && row.date < date)); + let broken = DataSet::from_components_with_actions_and_quotes( + parts.instruments, + parts.market, + parts.factors, + parts.candidates, + parts.benchmarks, + vec![], + parts.execution_quotes, + ) + .unwrap(); + assert!( + broker + .execute(date, &mut account, &broken, &decision) + .is_err() + ); + assert_eq!(account.financial_replay_identity(), initial); + assert_eq!(broker.next_order_id.get(), id); + assert!(broker.open_orders.borrow().is_empty()); + let result = broker + .execute(date, &mut account, &restored, &decision) + .unwrap(); + assert_eq!(result.fill_events.len(), 2); + assert_eq!(result.fill_events[0].order_id, Some(id)); + assert_eq!(result.fill_events[1].order_id, Some(id + 1)); + assert_eq!(account.position("000001.SZ").unwrap().quantity, 300); + assert_eq!(account.position("000002.SZ").unwrap().quantity, 300); + assert_eq!( + prior.fill_events.len(), + 2, + "previously returned report remains intact" + ); +} + +#[test] +fn unwinding_an_unpublished_simulator_transaction_restores_its_state() { + let (broker, mut account) = recovery_test_broker(false, OrderSide::Sell); + let initial = account.financial_replay_identity(); + let orders = format!("{:?}", broker.open_orders.borrow()); + let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let _ = broker.execution_transaction(&mut account, true, None, |account| { + account.apply_cash_delta(500.).unwrap(); + broker.open_orders.borrow_mut().clear(); + panic!("isolated simulator callback unwind"); + }); + })); + assert!(result.is_err()); + assert_eq!(account.financial_replay_identity(), initial); + assert_eq!(format!("{:?}", broker.open_orders.borrow()), orders); + assert_eq!(broker.execution_transaction_depth.get(), 0); +} + +#[test] +fn deferred_etf_batch_failure_keeps_both_targets_and_prior_generation_progress() { + let date = chrono::NaiveDate::from_ymd_opt(2025, 1, 3).unwrap(); + let previous = chrono::NaiveDate::from_ymd_opt(2025, 1, 2).unwrap(); + let mut parts = recovery_test_data(None, false).snapshot_components(); + for instrument in &mut parts.instruments { + instrument.board = "ETF".into(); + } + let good = DataSet::from_components_with_actions_and_quotes( + parts.instruments.clone(), + parts.market.clone(), + parts.factors.clone(), + parts.candidates.clone(), + parts.benchmarks.clone(), + vec![], + vec![], + ) + .unwrap(); + parts + .market + .retain(|row| !(row.date == date && row.symbol.as_str() == "000002.SZ")); + let bad = DataSet::from_components_with_actions_and_quotes( + parts.instruments, + parts.market, + parts.factors, + parts.candidates, + parts.benchmarks, + vec![], + vec![], + ) + .unwrap(); + let broker = BrokerSimulator::new(ChinaAShareCostModel::default(), ChinaEquityRuleHooks) + .with_volume_limit(false) + .with_liquidity_limit(false); + let members = std::sync::Arc::new( + (1..=2) + .map(|index| crate::stock_pool_execution::StockPoolMemberSpec { + symbol: format!("{index:06}.SZ"), + requested_order: index, + recommendation_reason: String::new(), + target_weight_bps: None, + stop_loss: None, + take_profit: None, + }) + .collect::>(), + ); + broker + .deferred_etf_targets + .borrow_mut() + .replace_generation("pool", "latest"); + for index in 1..=2 { + broker + .deferred_etf_targets + .borrow_mut() + .upsert(crate::etf_execution::DeferredEtfTarget { + pool_id: "pool".into(), + generation: "latest".into(), + symbol: format!("{index:06}.SZ"), + signal_date: previous, + signal_at: previous.and_hms_opt(13, 0, 0).unwrap(), + execute_on: Some(date), + target_value: 1000.into(), + target_weight_bps: 5000, + side: crate::stock_pool_execution::OrderSide::Buy, + max_positions: 2, + rule: Default::default(), + members: std::sync::Arc::clone(&members), + reason: "deferred recovery fixture".into(), + }); + } + let queue = format!("{:?}", broker.deferred_etf_targets.borrow()); + let mut account = PortfolioState::new(10000.); + let state = account.stock_pool_execution_state("pool"); + assert!( + broker + .execute_deferred_etf_targets(date, &mut account, &bad) + .is_err() + ); + assert_eq!(account.cash(), 10000.); + assert!(account.positions().is_empty()); + assert_eq!(account.stock_pool_execution_state("pool"), state); + assert_eq!(format!("{:?}", broker.deferred_etf_targets.borrow()), queue); + assert_eq!(broker.next_order_id.get(), 1); + assert_eq!(broker.execution_transaction_depth.get(), 0); + let result = broker + .execute_deferred_etf_targets(date, &mut account, &good) + .unwrap(); + assert_eq!(result.fill_events.len(), 2, "{result:?}"); + assert_eq!(broker.pending_etf_target_count(), 0); + assert_eq!(account.position("000001.SZ").unwrap().quantity, 100); + assert_eq!(account.position("000002.SZ").unwrap().quantity, 100); +} + +#[test] +fn public_callback_unwind_does_not_leak_order_context_or_authoritative_prior_state() { + struct PanicRules; + impl crate::rules::EquityRuleHooks for PanicRules { + fn can_buy( + &self, + _: chrono::NaiveDate, + _: &DailyMarketSnapshot, + _: &CandidateEligibility, + _: PriceField, + ) -> crate::rules::RuleCheck { + panic!("isolated rule callback panic") + } + fn can_sell( + &self, + _: chrono::NaiveDate, + _: &DailyMarketSnapshot, + _: &CandidateEligibility, + _: &crate::portfolio::Position, + _: PriceField, + ) -> crate::rules::RuleCheck { + unreachable!() + } + } + let date = chrono::NaiveDate::from_ymd_opt(2025, 1, 3).unwrap(); + let prior = chrono::NaiveDate::from_ymd_opt(2025, 1, 2).unwrap(); + let broker = BrokerSimulator::new(ChinaAShareCostModel::default(), PanicRules) + .with_volume_limit(false) + .with_liquidity_limit(false); + broker.runtime_decision_date.set(Some(prior)); + broker + .runtime_buy_denials + .borrow_mut() + .insert("unrelated".into(), "prior".into()); + let mut account = PortfolioState::new(10000.); + let decision = StrategyDecision { + buy_denials: BTreeMap::from([("another".into(), "temporary".into())]), + order_intents: vec![ + OrderIntent::LimitShares { + symbol: "000001.SZ".into(), + quantity: 100, + limit_price: 10., + reason: "panic fixture".into(), + } + .with_time_in_force(OrderTimeInForce::Gtc), + ], + ..Default::default() + }; + assert!( + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + let _ = broker.execute( + date, + &mut account, + &recovery_test_data(None, false), + &decision, + ); + })) + .is_err() + ); + assert_eq!(account.cash(), 10000.); + assert!(account.positions().is_empty()); + assert!(broker.open_order_views().is_empty()); + assert_eq!(broker.runtime_decision_date.get(), Some(prior)); + assert_eq!( + *broker.runtime_buy_denials.borrow(), + BTreeMap::from([("unrelated".into(), "prior".into())]) + ); + assert_eq!(broker.runtime_time_in_force.get(), None); + assert_eq!(broker.runtime_target_position_limit.get(), None); + assert_eq!(broker.execution_transaction_depth.get(), 0); +} + +#[test] +fn simulator_transaction_profile_preserves_successful_output() { + let date = chrono::NaiveDate::from_ymd_opt(2025, 1, 3).unwrap(); + let previous = chrono::NaiveDate::from_ymd_opt(2025, 1, 2).unwrap(); + let mut instruments = Vec::new(); + let mut market = Vec::new(); + let mut candidates = Vec::new(); + for index in 1..=30 { + let symbol = format!("{index:06}.SZ"); + let mut instrument = limit_test_instrument(); + instrument.symbol = symbol.clone(); + instruments.push(instrument); + for day in [previous, date] { + let mut row = dated_limit_test_snapshot(day); + row.symbol = symbol.clone().into(); + market.push(row); + let mut row = dated_limit_test_candidate(day, false, false, true, true); + row.symbol = symbol.clone().into(); + candidates.push(row); + } + } + let data = DataSet::from_components( + instruments, + market, + vec![], + candidates, + vec![ + dated_limit_test_benchmark(previous), + dated_limit_test_benchmark(date), + ], + ) + .unwrap(); + let mut reference = None; + let mut samples = Vec::new(); + for protected in [false, true, true, false] { + let broker = BrokerSimulator::new(ChinaAShareCostModel::default(), ChinaEquityRuleHooks) + .with_matching_type(MatchingType::CurrentBarClose) + .with_volume_limit(false) + .with_liquidity_limit(false); + // Private comparison only: no runtime option can disable protection. + if !protected { + broker.execution_transaction_depth.set(1); + } + let mut account = PortfolioState::new(10_000_000.); + for index in 1..=30 { + for _ in 0..20 { + account + .position_mut(&format!("{index:06}.SZ")) + .buy(previous, 100, 10.); + } + } + account.begin_trading_day(); + let mut orders = Vec::new(); + let mut fills = Vec::new(); + let start = std::time::Instant::now(); + for index in 0..500 { + let report = broker + .execute( + date, + &mut account, + &data, + &StrategyDecision { + order_intents: vec![OrderIntent::Shares { + symbol: format!("{:06}.SZ", index % 30 + 1), + quantity: 100, + reason: "transaction profile".into(), + }], + ..Default::default() + }, + ) + .unwrap(); + orders.extend(report.order_events); + fills.extend(report.fill_events); + } + samples.push( + serde_json::json!({"protected":protected,"microseconds":start.elapsed().as_micros()}), + ); + assert_eq!(fills.len(), 500); + let outcome = serde_json::json!({"orders":orders,"fills":fills,"ledger":account.financial_replay_identity()}); + if let Some(reference) = &reference { + assert_eq!(&outcome, reference); + } else { + reference = Some(outcome); + } + } + println!( + "simulator_transaction_profile={}", + serde_json::json!({"securities":30,"initial_lots_per_security":20,"calls":500,"samples":samples, + "scope":"isolated broker only; not Source or full backtest throughput"}) + ); +} diff --git a/crates/fidc-core/src/broker_stock_pool.rs b/crates/fidc-core/src/broker_stock_pool.rs index f48aae0..77d78d9 100644 --- a/crates/fidc-core/src/broker_stock_pool.rs +++ b/crates/fidc-core/src/broker_stock_pool.rs @@ -5,7 +5,7 @@ use crate::stock_pool_execution as pool; use rust_decimal::{Decimal, prelude::ToPrimitive}; use chrono::Timelike; -#[derive(Debug)] +#[derive(Debug, Clone)] pub(super) struct DeferredStockPoolExecution { date: NaiveDate, contract: Box, @@ -751,6 +751,11 @@ impl BrokerSimulator { /// Called at the opening clock, after settlement/corporate actions and /// auction callbacks. It never sends a stock order or replays a strategy. pub(crate) fn execute_deferred_etf_targets(&self, date: NaiveDate, portfolio: &mut PortfolioState, data: &DataSet) -> Result { + self.execution_transaction(portfolio, !self.has_open_orders() && self.pending_etf_target_count() > 0, None, + |portfolio| self.execute_deferred_etf_targets_inner(date, portfolio, data)) + } + + fn execute_deferred_etf_targets_inner(&self, date: NaiveDate, portfolio: &mut PortfolioState, data: &DataSet) -> Result { let mut report = BrokerExecutionReport::default(); if self.has_open_orders() { if self.pending_etf_target_count() > 0 { diff --git a/crates/fidc-core/src/etf_execution.rs b/crates/fidc-core/src/etf_execution.rs index 13a9212..fbbb321 100644 --- a/crates/fidc-core/src/etf_execution.rs +++ b/crates/fidc-core/src/etf_execution.rs @@ -61,7 +61,7 @@ pub(crate) struct DeferredEtfTarget { /// Owned by one broker/run. Replacing a full pool generation supersedes older /// queued targets; order of the latest candidate list is retained. -#[derive(Debug, Default)] +#[derive(Debug, Default, Clone)] pub(crate) struct DeferredEtfTargets { generations: std::collections::BTreeMap, rows: Vec, diff --git a/crates/fidc-core/src/portfolio.rs b/crates/fidc-core/src/portfolio.rs index e5bc9c5..bc52548 100644 --- a/crates/fidc-core/src/portfolio.rs +++ b/crates/fidc-core/src/portfolio.rs @@ -715,6 +715,27 @@ pub struct PortfolioState { stock_pool_states: std::collections::BTreeMap, } +pub(crate) struct PortfolioCheckpoint { + saved: PortfolioState, + position_order: Option>, +} + +impl PortfolioCheckpoint { + pub(crate) fn restore(mut self, current: &mut PortfolioState) { + if let Some(order) = self.position_order.take() { + let mut positions = IndexMap::with_capacity(order.len()); + for symbol in order { + let position = self.saved.positions.shift_remove(&symbol) + .or_else(|| current.positions.shift_remove(&symbol)) + .expect("unchanged checkpoint position must remain present"); + positions.insert(symbol, position); + } + self.saved.positions = positions; + } + *current = self.saved; + } +} + #[derive(Debug, Clone)] pub struct PendingCashFlow { pub payable_date: NaiveDate, @@ -734,6 +755,29 @@ pub(crate) struct SuccessorConversionOutcome { } impl PortfolioState { + /// Ordinary single-security orders need not duplicate every other lot. + /// Complex portfolio intents request the complete checkpoint instead. + pub(crate) fn checkpoint(&self, symbols: Option<&BTreeSet>) -> PortfolioCheckpoint { + let Some(symbols) = symbols else { + return PortfolioCheckpoint { saved: self.clone(), position_order: None }; + }; + PortfolioCheckpoint { + saved: Self { + initial_cash: self.initial_cash, units: self.units, cash: self.cash, + external_cash_flow_total: self.external_cash_flow_total, + cash_liabilities: self.cash_liabilities, management_fee_rate: self.management_fee_rate, + management_fees: self.management_fees, + // prune_flat_positions can remove an unrelated zero row. + positions: self.positions.iter().filter(|(symbol, position)| position.quantity == 0 || symbols.contains(*symbol)) + .map(|(symbol, position)| (symbol.clone(), position.clone())).collect(), + cash_receivables: self.cash_receivables.clone(), pending_cash_flows: self.pending_cash_flows.clone(), + day_sold_symbols: self.day_sold_symbols.clone(), corporate_predecessors: self.corporate_predecessors.clone(), + stock_pool_states: self.stock_pool_states.clone(), + }, + position_order: Some(self.positions.keys().cloned().collect()), + } + } + 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"); @@ -1647,6 +1691,35 @@ mod tests { BenchmarkSnapshot, CandidateEligibility, DailyFactorSnapshot, DailyMarketSnapshot, DataSet, PriceField, }; + + #[test] + fn scoped_checkpoint_restores_order_flat_rows_and_progress_without_copying_untouched_lots() { + let date = NaiveDate::from_ymd_opt(2026,9,15).unwrap(); + let mut portfolio = PortfolioState::new(10000.); + portfolio.position_mut("000001.SZ").buy(date,100,10.); + portfolio.position_mut("000002.SZ").buy(date,200,10.); + portfolio.position_mut("000003.SZ").buy(date,100,10.); + portfolio.position_mut("000003.SZ").sell(100,11.).unwrap(); + let flat_realized = portfolio.position("000003.SZ").unwrap().realized_pnl; + let untouched_lots = portfolio.position("000002.SZ").unwrap().lots.as_ptr(); + let before = portfolio.financial_replay_identity(); + let order = portfolio.positions.keys().cloned().collect::>(); + let checkpoint = portfolio.checkpoint(Some(&BTreeSet::from(["000001.SZ".into(), "000004.SZ".into()]))); + assert!(!checkpoint.saved.positions.contains_key("000002.SZ")); + portfolio.position_mut("000001.SZ").sell(100,11.).unwrap(); + portfolio.prune_flat_positions(); + portfolio.position_mut("000004.SZ").buy(date,100,12.); + portfolio.apply_cash_delta(100.).unwrap(); + portfolio.stock_pool_states.insert("changed".into(), Default::default()); + checkpoint.restore(&mut portfolio); + assert_eq!(portfolio.financial_replay_identity(), before); + assert_eq!(portfolio.positions.keys().cloned().collect::>(), order); + assert_eq!(portfolio.position("000002.SZ").unwrap().lots.as_ptr(), untouched_lots); + assert_eq!(portfolio.position("000003.SZ").unwrap().realized_pnl, flat_realized); + assert!(portfolio.stock_pool_states.is_empty()); + assert!(portfolio.position("000004.SZ").is_none()); + } + #[test] fn cash_ledger_accumulates_micro_yuan_exactly() { let mut portfolio = PortfolioState::new(1_000_000.0); diff --git a/docs/simulator-order-recovery-20260915.md b/docs/simulator-order-recovery-20260915.md new file mode 100644 index 0000000..0cd9456 --- /dev/null +++ b/docs/simulator-order-recovery-20260915.md @@ -0,0 +1,29 @@ +# 模拟器异常恢复与活动委托保留 + +2026-09-15。开发候选，未发布生产；不是实际GT/QMT撤改单功能。 + +## 已复现的问题 + +活动委托恢复先用 `mem::take` 取出整批订单。某只证券历史滑点证据缺失而返回错误后，当前订单及后续未处理订单被一起丢掉。原有两笔GTC模拟委托，一笔已累计成交100股，错误后队列直接为空。若前一笔已经在本次调用内部撮合，资金/持仓/手续费和行情消费也可能改变，但整个调用没有返回成功报告，重试将不一致。 + +## 事务边界 + +- 只保护一次模拟器调用中尚未成功返回的内部结果：资金、实际批次持仓、订单及累计成交、股票池/ETF顺延目标、执行游标/成交量消费、手续费状态和内部编号，以及临时运行上下文。 +- 本次调用明确失败或回调解栈时恢复检查点，错误继续向调用方返回；不吞错、不自动重跑。调用方仍应修复输入后重试原请求或发出明确新请求，不能把失败请求当成已接受的新目标。 +- 已成功返回的旧成交和报告、进入本次调用前已确认的手工事实不回滚。真实券商订单和回报不在该内存事务内，不能撤销或伪造实际GT/QMT结果。 +- 正常业务拒绝仍是有效结果：报告成功返回时，其他成功成交与拒绝记录一同保留，不因为有Rejected状态就整批回退。 +- 普通执行及ETF顺延消费共用边界，嵌套调用只保留一次检查点；临时日期、委托有效期、风险限制等在回调异常后不会泄漏到下一次调用。硬件断电、OOM以及用户自定义钩子的外部副作用不属于此证明。 + +## 开销控制 + +初版完整复制全部持仓。现普通单证券指令只保存可能变动的持仓与可被清理的零股记录，保留原位置顺序；完整组合、新类型指令和股票池阶段使用保守的完整检查点。未触及的持仓批次不复制，失败时才重新组装；无工作、无交易的调用不建检查点。 + +隔离未优化编译配置下，30只证券、每只20个初始批次、500次调用：初版保护样例约35.7—40.2ms，缩小持仓范围后约24.8—26.8ms；无保护对照11.5—22.1ms，首轮/并发噪声存在。保护有成本，不能称无性能回退，更不能据此宣称生产整段回测提速。四次对照的委托、成交和经济账本完全相同，正式Source/Runner性能准入仍保留。 + +## 验证 + +七项新增专项覆盖：日线/分钟、买/卖、第一或第二笔失败，已有部分成交；补齐数据后的原编号恢复与正常一次执行逐字段一致；此前成功调用不受后续失败影响；ETF两个顺延目标和进度完整保留；异常解栈；公开回调的临时上下文；局部检查点的位置顺序、零股和未复制批次；成功路径微基准的结果等价。 + +本机Core926、Trading625、Runner463/API129全量通过，原9/63/16项ignore不计。测试初次String/CompactString赋值错误及筛选名匹配0项已纠正，0项不当作通过；新增批次对照也改用实际更晚的执行时钟，不把未来报价当可立即成交。 + +未修改UI、在线账户、交易路由或任务；未做新的私有PG或真实券商验收。Source d5冻结保持，跨公司行为的实际券商委托调整、正式换股数据/范围、在线事实重建与真正Source/Runner联合验收仍未完成。Linux使用本轮新只读快照，不复用旧收据；没有release/tag或生产重启。