保留模拟器失败调用前的委托与执行状态

2026-09-15 01:28:38 +08:00
parent 4c6147e2ee
commit 695fdee4b8
6 changed files with 776 additions and 30 deletions
@@ -100,7 +100,7 @@ struct QuoteLiquidityConsumption {
    quantity: u32,
 }

-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 struct IntradayExecutionLedger {
    cursors: BTreeMap<String, NaiveDateTime>,
    depth_consumption: BTreeMap<String, [Option<QuoteDepthConsumption>; 2]>,
@@ -235,7 +235,7 @@ enum BrokerCallbackPhase {
    BeforeStrategy,
 }

-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 struct BrokerExecutionSession {
    date: Option<NaiveDate>,
    intraday_turnover: BTreeMap<String, u32>,
@@ -441,6 +441,43 @@ impl<T: Copy> Drop for RestoreCell<'_, T> {
    fn drop(&mut self) { self.0.set(self.1); }
 }

+struct RestoreRefCell<'a, T>(&'a RefCell<T>, Option<T>);
+impl<T> 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<C, R>(broker: &BrokerSimulator<C, R>) -> Self {
+                Self { $($field: broker.$field.get()),* }
+            }
+            fn restore<C, R>(self, broker: &BrokerSimulator<C, R>) {
+                $(broker.$field.set(self.$field);)*
+            }
+        }
+    };
+}
+execution_context_checkpoint! {
+    runtime_etf_daily_open: bool,
+    runtime_stock_pool_followup: bool,
+    runtime_intraday_start_time: Option<NaiveTime>,
+    runtime_intraday_end_time: Option<NaiveTime>,
+    runtime_execution_clock: Option<NaiveTime>,
+    runtime_callback_phase: BrokerCallbackPhase,
+    runtime_algo_schedule: Option<AlgoExecutionRequest>,
+    runtime_unprocessed_algorithm_cash: FixedMoney,
+    runtime_decision_date: Option<NaiveDate>,
+    runtime_order_created_date: Option<NaiveDate>,
+    runtime_resting_order_origin: Option<RestingOrderOrigin>,
+    runtime_decision_total_equity: Option<f64>,
+    runtime_target_position_limit: Option<usize>,
+    runtime_time_in_force: Option<OrderTimeInForce>,
+}
+
 pub struct BrokerSimulator<C, R> {
    historical_etf_open_fallback: bool,
    verified_etf_minute_absences: RefCell<BTreeSet<(NaiveDate, String)>>,
@@ -485,9 +522,96 @@ pub struct BrokerSimulator<C, R> {
    next_order_id: Cell<u64>,
    open_orders: RefCell<Vec<OpenOrder>>,
    execution_session: RefCell<BrokerExecutionSession>,
+    execution_transaction_depth: Cell<usize>,
+}
+
+/// 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<OpenOrder>,
+    etf_targets: crate::etf_execution::DeferredEtfTargets,
+    pool_targets: BTreeMap<String, stock_pool::DeferredStockPoolExecution>,
+    sold: BTreeMap<NaiveDate, BTreeSet<String>>,
+    session: BrokerExecutionSession,
+    next_order_id: u64,
+    context: BrokerExecutionContext,
+}
+
+impl BrokerExecutionCheckpoint {
+    fn capture<C, R>(broker: &BrokerSimulator<C, R>, portfolio: &PortfolioState, symbols: Option<&BTreeSet<String>>) -> 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<C, R>(self, broker: &BrokerSimulator<C, R>, 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<C, R>,
+    portfolio: &'a mut PortfolioState,
+    checkpoint: Option<BrokerExecutionCheckpoint>,
+}
+
+impl<C, R> Drop for BrokerExecutionTransaction<'_, C, R> {
+    fn drop(&mut self) {
+        if let Some(checkpoint) = self.checkpoint.take() {
+            checkpoint.restore(self.broker, self.portfolio);
+        }
+    }
 }

 impl<C, R> BrokerSimulator<C, R> {
+    fn execution_transaction<F>(&self, portfolio: &mut PortfolioState, needed: bool, symbols: Option<&BTreeSet<String>>, execute: F)
+        -> Result<BrokerExecutionReport, BacktestError>
+    where F: FnOnce(&mut PortfolioState) -> Result<BrokerExecutionReport, BacktestError> {
+        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<BTreeSet<String>> {
+        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::<BTreeSet<_>>();
+        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<C, R> BrokerSimulator<C, R> {
            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<C, R> BrokerSimulator<C, R> {
            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<BrokerExecutionReport, BacktestError> {
-        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() {
@@ -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<usize>, 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<ChinaAShareCostModel, ChinaEquityRuleHooks>,
+    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::<Vec<_>>(),
+    );
+    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"})
+    );
+}
@@ -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<pool::FrozenStockPoolIntent>,
@@ -751,6 +751,11 @@ impl<C: CostModel, R: EquityRuleHooks> BrokerSimulator<C, R> {
    /// 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<BrokerExecutionReport, BacktestError> {
+        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<BrokerExecutionReport, BacktestError> {
        let mut report = BrokerExecutionReport::default();
        if self.has_open_orders() {
            if self.pending_etf_target_count() > 0 {
@@ -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<String, String>,
    rows: Vec<DeferredEtfTarget>,
@@ -715,6 +715,27 @@ pub struct PortfolioState {
    stock_pool_states: std::collections::BTreeMap<String,crate::stock_pool_state::StockPoolExecutionState>,
 }

+pub(crate) struct PortfolioCheckpoint {
+    saved: PortfolioState,
+    position_order: Option<Vec<String>>,
+}
+
+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<String>>) -> 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::<Vec<_>>();
+        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::<Vec<_>>(), 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);