fidc-backtest-engine/crates/fidc-core/src/engine.rs

use std::collections::{BTreeMap, BTreeSet};

use chrono::{Datelike, Duration, NaiveDate, NaiveTime};
use serde::Serialize;
use thiserror::Error;

use crate::broker::{BrokerExecutionReport, BrokerSimulator, MatchingType};
use crate::cost::CostModel;
use crate::data::{BenchmarkSnapshot, DataSet, DataSetError, IntradayExecutionQuote, PriceField};
use crate::event_bus::{BacktestProcessMod, BacktestProcessModLoader, ProcessEventBus};
use crate::events::{
    AccountEvent, FillEvent, OrderEvent, OrderSide, OrderStatus, PositionEvent, ProcessEvent,
    ProcessEventKind,
};
use crate::futures::{
    FuturesAccountState, FuturesExecutionReport, FuturesOrderIntent, FuturesPositionEffect,
    FuturesTransactionCostModel,
};
use crate::metrics::{BacktestMetrics, compute_backtest_metrics};
use crate::portfolio::{CashReceivable, HoldingSummary, PortfolioState};
use crate::risk_control::FidcRiskDecisionAudit;
use crate::rules::EquityRuleHooks;
use crate::scheduler::{ScheduleRule, ScheduleStage, Scheduler, default_stage_time};
use crate::strategy::{
    OpenOrderView, OrderIntent, Strategy, StrategyContext, StrategyDecision,
    TargetPortfolioOrderPricing,
};

#[derive(Debug, Error)]
pub enum BacktestError {
    #[error(transparent)]
    Data(#[from] DataSetError),
    #[error("missing {field} price for {symbol} on {date}")]
    MissingPrice {
        date: NaiveDate,
        symbol: String,
        field: &'static str,
    },
    #[error("benchmark snapshot missing for {date}")]
    MissingBenchmark { date: NaiveDate },
    #[error("{0}")]
    Execution(String),
}

#[derive(Debug, Clone)]
pub struct BacktestConfig {
    pub initial_cash: f64,
    pub benchmark_code: String,
    pub start_date: Option<NaiveDate>,
    pub end_date: Option<NaiveDate>,
    pub decision_lag_trading_days: usize,
    pub execution_price_field: PriceField,
}

#[derive(Debug, Clone, Copy)]
pub struct FuturesValidationConfig {
    pub enforce_active_instrument: bool,
    pub enforce_trading_phase: bool,
    pub enforce_limit_price_tick: bool,
    pub enforce_price_limits: bool,
}

impl Default for FuturesValidationConfig {
    fn default() -> Self {
        Self {
            enforce_active_instrument: true,
            enforce_trading_phase: true,
            enforce_limit_price_tick: true,
            enforce_price_limits: true,
        }
    }
}

#[derive(Debug, Clone, Serialize)]
pub struct DailyEquityPoint {
    #[serde(with = "date_format")]
    pub date: NaiveDate,
    pub cash: f64,
    pub market_value: f64,
    pub total_equity: f64,
    pub benchmark_close: f64,
    pub benchmark_prev_close: f64,
    pub notes: String,
    pub diagnostics: String,
}

#[derive(Debug, Clone)]
pub struct BacktestResult {
    pub strategy_name: String,
    pub equity_curve: Vec<DailyEquityPoint>,
    pub benchmark_series: Vec<BenchmarkSnapshot>,
    pub risk_decisions: Vec<FidcRiskDecisionAudit>,
    pub order_events: Vec<OrderEvent>,
    pub fills: Vec<FillEvent>,
    pub position_events: Vec<PositionEvent>,
    pub account_events: Vec<AccountEvent>,
    pub process_events: Vec<ProcessEvent>,
    pub holdings_summary: Vec<HoldingSummary>,
    pub daily_holdings: Vec<HoldingSummary>,
    pub metrics: BacktestMetrics,
}

#[derive(Debug, Clone)]
pub struct ExecutionQuoteRequest {
    pub date: NaiveDate,
    pub start_time: Option<chrono::NaiveTime>,
    pub end_time: Option<chrono::NaiveTime>,
    pub symbols: BTreeSet<String>,
}

type ExecutionQuoteLoader = Box<
    dyn FnMut(ExecutionQuoteRequest) -> Result<Vec<IntradayExecutionQuote>, BacktestError> + Send,
>;

#[derive(Debug, Clone, Serialize)]
pub struct AnalyzerTradeRow {
    #[serde(with = "date_format")]
    pub date: NaiveDate,
    pub order_id: Option<u64>,
    pub symbol: String,
    pub side: OrderSide,
    pub quantity: u32,
    pub price: f64,
    pub gross_amount: f64,
    pub transaction_cost: f64,
    pub net_cash_flow: f64,
    pub reason: String,
}

#[derive(Debug, Clone, Serialize)]
pub struct AnalyzerPositionRow {
    #[serde(with = "date_format")]
    pub date: NaiveDate,
    pub symbol: String,
    pub quantity: u32,
    pub market_value: f64,
    pub weight: f64,
    pub average_cost: f64,
    pub realized_pnl: f64,
    pub unrealized_pnl: f64,
    pub transaction_cost: f64,
}

#[derive(Debug, Clone, Serialize)]
pub struct AnalyzerMonthlyReturnRow {
    pub year: i32,
    pub month: u32,
    pub portfolio_return: f64,
    pub benchmark_return: f64,
    pub excess_return: f64,
}

#[derive(Debug, Clone, Serialize)]
pub struct AnalyzerRiskSummary {
    pub total_return: f64,
    pub annual_return: f64,
    pub benchmark_cumulative_return: f64,
    pub excess_cumulative_return: f64,
    pub alpha: f64,
    pub beta: f64,
    pub sharpe: f64,
    pub sortino: f64,
    pub information_ratio: f64,
    pub tracking_error: f64,
    pub volatility: f64,
    pub max_drawdown: f64,
    pub max_drawdown_duration_days: usize,
    pub win_rate: f64,
    pub excess_win_rate: f64,
}

#[derive(Debug, Clone, Serialize)]
pub struct AnalyzerReport {
    pub strategy_name: String,
    pub trades: Vec<AnalyzerTradeRow>,
    pub positions: Vec<AnalyzerPositionRow>,
    pub monthly_returns: Vec<AnalyzerMonthlyReturnRow>,
    pub risk_summary: AnalyzerRiskSummary,
    pub equity_curve: Vec<DailyEquityPoint>,
    pub benchmark_series: Vec<BenchmarkSnapshot>,
    pub metrics: BacktestMetrics,
}

impl BacktestResult {
    pub fn analyzer_report(&self) -> AnalyzerReport {
        AnalyzerReport {
            strategy_name: self.strategy_name.clone(),
            trades: self
                .fills
                .iter()
                .map(|fill| AnalyzerTradeRow {
                    date: fill.date,
                    order_id: fill.order_id,
                    symbol: fill.symbol.clone(),
                    side: fill.side,
                    quantity: fill.quantity,
                    price: fill.price,
                    gross_amount: fill.gross_amount,
                    transaction_cost: fill.commission + fill.stamp_tax,
                    net_cash_flow: fill.net_cash_flow,
                    reason: fill.reason.clone(),
                })
                .collect(),
            positions: self
                .daily_holdings
                .iter()
                .map(|holding| AnalyzerPositionRow {
                    date: holding.date,
                    symbol: holding.symbol.clone(),
                    quantity: holding.quantity,
                    market_value: holding.market_value,
                    weight: holding.value_percent,
                    average_cost: holding.average_cost,
                    realized_pnl: holding.realized_pnl,
                    unrealized_pnl: holding.unrealized_pnl,
                    transaction_cost: holding.transaction_cost,
                })
                .collect(),
            monthly_returns: self.analyzer_monthly_returns(),
            risk_summary: self.analyzer_risk_summary(),
            equity_curve: self.equity_curve.clone(),
            benchmark_series: self.benchmark_series.clone(),
            metrics: self.metrics.clone(),
        }
    }

    pub fn analyzer_report_json(&self) -> Result<String, serde_json::Error> {
        serde_json::to_string_pretty(&self.analyzer_report())
    }

    pub fn analyzer_monthly_returns(&self) -> Vec<AnalyzerMonthlyReturnRow> {
        let mut month_points = BTreeMap::<(i32, u32), (f64, f64, f64, f64)>::new();
        let mut previous_equity = self.metrics.initial_cash;
        let mut previous_benchmark = self
            .equity_curve
            .first()
            .map(|point| point.benchmark_prev_close)
            .unwrap_or_default();
        for point in &self.equity_curve {
            let key = (point.date.year(), point.date.month());
            month_points
                .entry(key)
                .and_modify(|(_, _, end_equity, end_benchmark)| {
                    *end_equity = point.total_equity;
                    *end_benchmark = point.benchmark_close;
                })
                .or_insert((
                    previous_equity,
                    previous_benchmark,
                    point.total_equity,
                    point.benchmark_close,
                ));
            previous_equity = point.total_equity;
            previous_benchmark = point.benchmark_close;
        }
        month_points
            .into_iter()
            .map(
                |((year, month), (start_equity, start_benchmark, end_equity, end_benchmark))| {
                    let portfolio_return = analyzer_ratio_change(start_equity, end_equity);
                    let benchmark_return = analyzer_ratio_change(start_benchmark, end_benchmark);
                    AnalyzerMonthlyReturnRow {
                        year,
                        month,
                        portfolio_return,
                        benchmark_return,
                        excess_return: portfolio_return - benchmark_return,
                    }
                },
            )
            .collect()
    }

    pub fn analyzer_risk_summary(&self) -> AnalyzerRiskSummary {
        AnalyzerRiskSummary {
            total_return: self.metrics.total_return,
            annual_return: self.metrics.annual_return,
            benchmark_cumulative_return: self.metrics.benchmark_cumulative_return,
            excess_cumulative_return: self.metrics.excess_cumulative_return,
            alpha: self.metrics.alpha,
            beta: self.metrics.beta,
            sharpe: self.metrics.sharpe,
            sortino: self.metrics.sortino,
            information_ratio: self.metrics.information_ratio,
            tracking_error: self.metrics.tracking_error,
            volatility: self.metrics.volatility,
            max_drawdown: self.metrics.max_drawdown,
            max_drawdown_duration_days: self.metrics.max_drawdown_duration_days,
            win_rate: self.metrics.win_rate,
            excess_win_rate: self.metrics.excess_win_rate,
        }
    }
}

#[derive(Debug, Clone, Serialize)]
pub struct BacktestDayProgress {
    #[serde(with = "date_format")]
    pub date: NaiveDate,
    pub cash: f64,
    pub market_value: f64,
    pub total_equity: f64,
    pub unit_nav: f64,
    pub total_return: f64,
    pub benchmark_close: f64,
    pub daily_fill_count: usize,
    pub cumulative_trade_count: usize,
    pub holding_count: usize,
    pub notes: String,
    pub diagnostics: String,
    pub orders: Vec<OrderEvent>,
    pub fills: Vec<FillEvent>,
    pub holdings: Vec<HoldingSummary>,
    pub process_events: Vec<ProcessEvent>,
}

#[derive(Debug, Clone)]
struct FuturesOpenOrder {
    order_id: u64,
    intent: FuturesOrderIntent,
    requested_quantity: u32,
    filled_quantity: u32,
    remaining_quantity: u32,
    limit_price: f64,
    reason: String,
}

pub struct BacktestEngine<S, C, R> {
    data: DataSet,
    strategy: S,
    broker: BrokerSimulator<C, R>,
    config: BacktestConfig,
    dividend_reinvestment: bool,
    cash_dividends_enabled: bool,
    cash_dividend_adjusts_cost_basis: bool,
    process_event_bus: ProcessEventBus,
    dynamic_universe: Option<BTreeSet<String>>,
    subscriptions: BTreeSet<String>,
    futures_account: Option<FuturesAccountState>,
    next_futures_order_id: u64,
    futures_open_orders: Vec<FuturesOpenOrder>,
    futures_expirations: BTreeMap<NaiveDate, BTreeMap<String, f64>>,
    futures_settlement_price_mode: String,
    futures_cost_model: FuturesTransactionCostModel,
    futures_validation_config: FuturesValidationConfig,
    execution_quote_loader: Option<ExecutionQuoteLoader>,
    execution_quote_request_cache:
        BTreeSet<(NaiveDate, String, Option<NaiveTime>, Option<NaiveTime>)>,
}

impl<S, C, R> BacktestEngine<S, C, R> {
    pub fn new(
        data: DataSet,
        strategy: S,
        broker: BrokerSimulator<C, R>,
        config: BacktestConfig,
    ) -> Self {
        Self {
            data,
            strategy,
            broker,
            config,
            dividend_reinvestment: false,
            cash_dividends_enabled: true,
            cash_dividend_adjusts_cost_basis: true,
            process_event_bus: ProcessEventBus::new(),
            dynamic_universe: None,
            subscriptions: BTreeSet::new(),
            futures_account: None,
            next_futures_order_id: 9_000_000_000,
            futures_open_orders: Vec::new(),
            futures_expirations: BTreeMap::new(),
            futures_settlement_price_mode: "close".to_string(),
            futures_cost_model: FuturesTransactionCostModel::default(),
            futures_validation_config: FuturesValidationConfig::default(),
            execution_quote_loader: None,
            execution_quote_request_cache: BTreeSet::new(),
        }
    }

    pub fn into_data(self) -> DataSet {
        self.data
    }

    pub fn with_execution_quote_loader<F>(mut self, loader: F) -> Self
    where
        F: FnMut(ExecutionQuoteRequest) -> Result<Vec<IntradayExecutionQuote>, BacktestError>
            + Send
            + 'static,
    {
        self.execution_quote_loader = Some(Box::new(loader));
        self
    }

    pub fn with_dividend_reinvestment(mut self, enabled: bool) -> Self {
        self.dividend_reinvestment = enabled;
        self
    }

    pub fn with_cash_dividends(mut self, enabled: bool) -> Self {
        self.cash_dividends_enabled = enabled;
        self
    }

    pub fn with_cash_dividend_cost_basis_adjustment(mut self, enabled: bool) -> Self {
        self.cash_dividend_adjusts_cost_basis = enabled;
        self
    }

    pub fn with_futures_account(mut self, account: FuturesAccountState) -> Self {
        self.futures_account = Some(account);
        self
    }

    pub fn with_futures_initial_cash(self, initial_cash: f64) -> Self {
        self.with_futures_account(FuturesAccountState::new(initial_cash))
    }

    pub fn futures_account(&self) -> Option<&FuturesAccountState> {
        self.futures_account.as_ref()
    }

    pub fn futures_account_mut(&mut self) -> Option<&mut FuturesAccountState> {
        self.futures_account.as_mut()
    }

    pub fn with_futures_expiration(
        mut self,
        date: NaiveDate,
        symbol: impl Into<String>,
        settlement_price: f64,
    ) -> Self {
        self.futures_expirations
            .entry(date)
            .or_default()
            .insert(symbol.into(), settlement_price);
        self
    }

    pub fn with_futures_expirations(
        mut self,
        expirations: BTreeMap<NaiveDate, BTreeMap<String, f64>>,
    ) -> Self {
        self.futures_expirations = expirations;
        self
    }

    pub fn with_futures_settlement_price_mode(mut self, mode: impl Into<String>) -> Self {
        self.futures_settlement_price_mode = mode.into();
        self
    }

    pub fn with_futures_transaction_cost_model(
        mut self,
        cost_model: FuturesTransactionCostModel,
    ) -> Self {
        self.futures_cost_model = cost_model;
        self
    }

    pub fn with_futures_validation_config(mut self, config: FuturesValidationConfig) -> Self {
        self.futures_validation_config = config;
        self
    }

    pub fn process_event_bus_mut(&mut self) -> &mut ProcessEventBus {
        &mut self.process_event_bus
    }

    pub fn add_process_listener<F>(&mut self, kind: ProcessEventKind, listener: F)
    where
        F: FnMut(&ProcessEvent) + 'static,
    {
        self.process_event_bus.add_listener(kind, listener);
    }

    pub fn add_any_process_listener<F>(&mut self, listener: F)
    where
        F: FnMut(&ProcessEvent) + 'static,
    {
        self.process_event_bus.add_any_listener(listener);
    }

    pub fn install_process_mod<M>(&mut self, module: &mut M)
    where
        M: BacktestProcessMod,
    {
        self.process_event_bus.install_mod(module);
    }

    pub fn install_process_mod_loader(
        &mut self,
        loader: &mut BacktestProcessModLoader,
    ) -> Vec<String> {
        self.process_event_bus.install_mod_loader(loader)
    }

    pub fn install_enabled_process_mods(
        &mut self,
        loader: &mut BacktestProcessModLoader,
        enabled_names: &[String],
    ) -> Vec<String> {
        self.process_event_bus
            .install_enabled_mods(loader, enabled_names)
    }
}

impl<S, C, R> BacktestEngine<S, C, R>
where
    S: Strategy,
    C: CostModel,
    R: EquityRuleHooks,
{
    fn ensure_execution_quotes_for_decision(
        &mut self,
        execution_date: NaiveDate,
        portfolio: &PortfolioState,
        open_orders: &[OpenOrderView],
        decision: &StrategyDecision,
        start_time: Option<chrono::NaiveTime>,
        end_time: Option<chrono::NaiveTime>,
    ) -> Result<(), BacktestError> {
        if self.execution_quote_loader.is_none() {
            return Ok(());
        }
        if self.broker.execution_price_field() != PriceField::Last
            && !decision_has_algo_execution(decision)
        {
            return Ok(());
        }

        let caller_start_time = start_time;
        let caller_end_time = end_time;
        let start_time = caller_start_time.or_else(|| self.broker.intraday_execution_start_time());
        let mut symbols = execution_quote_symbols_for_decision(decision, portfolio, open_orders);
        self.load_missing_execution_quotes(execution_date, start_time, end_time, &mut symbols)?;

        if caller_start_time.is_none() && caller_end_time.is_none() {
            for ((intent_start_time, intent_end_time), mut intent_symbols) in
                algo_execution_quote_windows_for_decision(decision, portfolio)
            {
                self.load_missing_execution_quotes(
                    execution_date,
                    intent_start_time,
                    intent_end_time,
                    &mut intent_symbols,
                )?;
            }
        }

        Ok(())
    }

    fn load_missing_execution_quotes(
        &mut self,
        execution_date: NaiveDate,
        start_time: Option<NaiveTime>,
        end_time: Option<NaiveTime>,
        symbols: &mut BTreeSet<String>,
    ) -> Result<(), BacktestError> {
        symbols.retain(|symbol| {
            let request_key = (execution_date, symbol.clone(), start_time, end_time);
            if self.execution_quote_request_cache.contains(&request_key) {
                return false;
            }
            if start_time.is_some() && end_time.is_none() {
                return !has_execution_quote_near_start_time(
                    &self.data,
                    execution_date,
                    symbol,
                    start_time.expect("checked start_time"),
                );
            }
            !has_execution_quote_in_window(&self.data, execution_date, symbol, start_time, end_time)
        });
        if symbols.is_empty() {
            return Ok(());
        }

        let requested_symbols = symbols.iter().cloned().collect::<Vec<_>>();
        let request = ExecutionQuoteRequest {
            date: execution_date,
            start_time,
            end_time,
            symbols: std::mem::take(symbols),
        };
        let quotes = self
            .execution_quote_loader
            .as_mut()
            .expect("checked execution quote loader")
            .as_mut()(request)?;
        self.data.add_execution_quotes(quotes);
        for symbol in requested_symbols {
            self.execution_quote_request_cache.insert((
                execution_date,
                symbol,
                start_time,
                end_time,
            ));
        }
        Ok(())
    }

    fn ensure_execution_quotes_for_portfolio_times(
        &mut self,
        execution_date: NaiveDate,
        portfolio: &PortfolioState,
        quote_times: &[NaiveTime],
    ) -> Result<(), BacktestError> {
        if self.execution_quote_loader.is_none() || quote_times.is_empty() {
            return Ok(());
        }
        let base_symbols = portfolio
            .positions()
            .keys()
            .cloned()
            .collect::<BTreeSet<_>>();
        if base_symbols.is_empty() {
            return Ok(());
        }
        for quote_time in quote_times {
            let mut symbols = base_symbols.clone();
            self.load_missing_execution_quotes(
                execution_date,
                Some(*quote_time),
                None,
                &mut symbols,
            )?;
        }
        Ok(())
    }

    fn ensure_execution_quotes_for_symbols_at_times(
        &mut self,
        execution_date: NaiveDate,
        symbols: &BTreeSet<String>,
        quote_times: &[NaiveTime],
    ) -> Result<(), BacktestError> {
        if self.execution_quote_loader.is_none() || quote_times.is_empty() || symbols.is_empty() {
            return Ok(());
        }
        let base_symbols = symbols
            .iter()
            .filter(|symbol| !symbol.trim().is_empty())
            .cloned()
            .collect::<BTreeSet<_>>();
        if base_symbols.is_empty() {
            return Ok(());
        }
        for quote_time in quote_times {
            let mut symbols = base_symbols.clone();
            self.load_missing_execution_quotes(
                execution_date,
                Some(*quote_time),
                None,
                &mut symbols,
            )?;
        }
        Ok(())
    }

    fn apply_strategy_directives(
        &mut self,
        execution_date: NaiveDate,
        decision_date: NaiveDate,
        decision_index: usize,
        portfolio: &mut PortfolioState,
        open_orders: &[crate::strategy::OpenOrderView],
        process_events: &mut Vec<ProcessEvent>,
        decision: &mut crate::strategy::StrategyDecision,
        directive_report: &mut BrokerExecutionReport,
    ) -> Result<(), BacktestError> {
        if decision.order_intents.is_empty() {
            return Ok(());
        }

        let mut retained = Vec::with_capacity(decision.order_intents.len());
        for intent in decision.order_intents.drain(..) {
            match intent {
                crate::strategy::OrderIntent::UpdateUniverse { symbols, reason } => {
                    let symbol_count = symbols.len();
                    self.dynamic_universe = Some(symbols.clone());
                    decision
                        .diagnostics
                        .push(format!("dynamic_universe_updated count={symbol_count}"));
                    publish_custom_process_event(
                        &mut self.strategy,
                        &mut self.process_event_bus,
                        execution_date,
                        decision_date,
                        decision_index,
                        &self.data,
                        portfolio,
                        self.futures_account.as_ref(),
                        open_orders,
                        self.dynamic_universe.as_ref(),
                        &self.subscriptions,
                        process_events,
                        ProcessEvent {
                            date: execution_date,
                            kind: ProcessEventKind::UniverseUpdated,
                            order_id: None,
                            symbol: (symbol_count == 1)
                                .then(|| symbols.iter().next().cloned())
                                .flatten(),
                            side: None,
                            detail: format!(
                                "reason={reason} count={symbol_count} symbols={}",
                                symbols.iter().cloned().collect::<Vec<_>>().join(",")
                            ),
                        },
                    )?;
                }
                crate::strategy::OrderIntent::Subscribe { symbols, reason } => {
                    let mut added = Vec::new();
                    for symbol in symbols {
                        if self.subscriptions.insert(symbol.clone()) {
                            added.push(symbol);
                        }
                    }
                    if !added.is_empty() {
                        decision.diagnostics.push(format!(
                            "subscriptions_added count={} total={}",
                            added.len(),
                            self.subscriptions.len()
                        ));
                        publish_custom_process_event(
                            &mut self.strategy,
                            &mut self.process_event_bus,
                            execution_date,
                            decision_date,
                            decision_index,
                            &self.data,
                            portfolio,
                            self.futures_account.as_ref(),
                            open_orders,
                            self.dynamic_universe.as_ref(),
                            &self.subscriptions,
                            process_events,
                            ProcessEvent {
                                date: execution_date,
                                kind: ProcessEventKind::UniverseSubscribed,
                                order_id: None,
                                symbol: (added.len() == 1).then(|| added[0].clone()),
                                side: None,
                                detail: format!(
                                    "reason={reason} count={} symbols={}",
                                    added.len(),
                                    added.join(",")
                                ),
                            },
                        )?;
                    }
                }
                crate::strategy::OrderIntent::Unsubscribe { symbols, reason } => {
                    let mut removed = Vec::new();
                    for symbol in symbols {
                        if self.subscriptions.remove(&symbol) {
                            removed.push(symbol);
                        }
                    }
                    if !removed.is_empty() {
                        decision.diagnostics.push(format!(
                            "subscriptions_removed count={} total={}",
                            removed.len(),
                            self.subscriptions.len()
                        ));
                        publish_custom_process_event(
                            &mut self.strategy,
                            &mut self.process_event_bus,
                            execution_date,
                            decision_date,
                            decision_index,
                            &self.data,
                            portfolio,
                            self.futures_account.as_ref(),
                            open_orders,
                            self.dynamic_universe.as_ref(),
                            &self.subscriptions,
                            process_events,
                            ProcessEvent {
                                date: execution_date,
                                kind: ProcessEventKind::UniverseUnsubscribed,
                                order_id: None,
                                symbol: (removed.len() == 1).then(|| removed[0].clone()),
                                side: None,
                                detail: format!(
                                    "reason={reason} count={} symbols={}",
                                    removed.len(),
                                    removed.join(",")
                                ),
                            },
                        )?;
                    }
                }
                crate::strategy::OrderIntent::DepositWithdraw {
                    amount,
                    receiving_days,
                    reason,
                } => {
                    let cash_before = portfolio.cash();
                    if receiving_days == 0 {
                        portfolio
                            .deposit_withdraw(amount)
                            .map_err(BacktestError::Execution)?;
                        directive_report.account_events.push(AccountEvent {
                            date: execution_date,
                            cash_before,
                            cash_after: portfolio.cash(),
                            total_equity: portfolio.total_equity(),
                            note: format!("deposit_withdraw amount={amount:.2} reason={reason}"),
                        });
                    } else {
                        let payable_date = self
                            .data
                            .next_trading_date(execution_date, receiving_days)
                            .ok_or_else(|| {
                                BacktestError::Execution(format!(
                                    "no trading date for deposit_withdraw receiving_days={receiving_days} from {execution_date}"
                                ))
                            })?;
                        portfolio
                            .schedule_deposit_withdraw(payable_date, amount, reason.clone())
                            .map_err(BacktestError::Execution)?;
                        directive_report.account_events.push(AccountEvent {
                            date: execution_date,
                            cash_before,
                            cash_after: portfolio.cash(),
                            total_equity: portfolio.total_equity(),
                            note: format!(
                                "deposit_withdraw_scheduled amount={amount:.2} payable_date={payable_date} reason={reason}"
                            ),
                        });
                    }
                    decision.diagnostics.push(format!(
                        "account_deposit_withdraw amount={amount:.2} receiving_days={receiving_days}"
                    ));
                    publish_custom_process_event(
                        &mut self.strategy,
                        &mut self.process_event_bus,
                        execution_date,
                        decision_date,
                        decision_index,
                        &self.data,
                        &*portfolio,
                        self.futures_account.as_ref(),
                        open_orders,
                        self.dynamic_universe.as_ref(),
                        &self.subscriptions,
                        process_events,
                        ProcessEvent {
                            date: execution_date,
                            kind: ProcessEventKind::AccountDepositWithdraw,
                            order_id: None,
                            symbol: None,
                            side: None,
                            detail: format!(
                                "reason={reason} amount={amount:.2} receiving_days={receiving_days} cash_before={cash_before:.2} cash_after={:.2}",
                                portfolio.cash()
                            ),
                        },
                    )?;
                }
                crate::strategy::OrderIntent::FinanceRepay { amount, reason } => {
                    let cash_before = portfolio.cash();
                    let liabilities_before = portfolio.cash_liabilities();
                    portfolio
                        .finance_repay(amount)
                        .map_err(BacktestError::Execution)?;
                    directive_report.account_events.push(AccountEvent {
                        date: execution_date,
                        cash_before,
                        cash_after: portfolio.cash(),
                        total_equity: portfolio.total_equity(),
                        note: format!(
                            "finance_repay amount={amount:.2} liabilities_before={liabilities_before:.2} liabilities_after={:.2} reason={reason}",
                            portfolio.cash_liabilities()
                        ),
                    });
                    decision.diagnostics.push(format!(
                        "account_finance_repay amount={amount:.2} liabilities={:.2}",
                        portfolio.cash_liabilities()
                    ));
                    publish_custom_process_event(
                        &mut self.strategy,
                        &mut self.process_event_bus,
                        execution_date,
                        decision_date,
                        decision_index,
                        &self.data,
                        &*portfolio,
                        self.futures_account.as_ref(),
                        open_orders,
                        self.dynamic_universe.as_ref(),
                        &self.subscriptions,
                        process_events,
                        ProcessEvent {
                            date: execution_date,
                            kind: ProcessEventKind::AccountFinanceRepay,
                            order_id: None,
                            symbol: None,
                            side: None,
                            detail: format!(
                                "reason={reason} amount={amount:.2} cash_before={cash_before:.2} cash_after={:.2} liabilities_before={liabilities_before:.2} liabilities_after={:.2}",
                                portfolio.cash(),
                                portfolio.cash_liabilities()
                            ),
                        },
                    )?;
                }
                crate::strategy::OrderIntent::SetManagementFeeRate { rate, reason } => {
                    portfolio
                        .set_management_fee_rate(rate)
                        .map_err(BacktestError::Execution)?;
                    decision
                        .diagnostics
                        .push(format!("account_management_fee_rate rate={rate:.6}"));
                    publish_custom_process_event(
                        &mut self.strategy,
                        &mut self.process_event_bus,
                        execution_date,
                        decision_date,
                        decision_index,
                        &self.data,
                        &*portfolio,
                        self.futures_account.as_ref(),
                        open_orders,
                        self.dynamic_universe.as_ref(),
                        &self.subscriptions,
                        process_events,
                        ProcessEvent {
                            date: execution_date,
                            kind: ProcessEventKind::AccountManagementFee,
                            order_id: None,
                            symbol: None,
                            side: None,
                            detail: format!(
                                "reason={reason} rate={rate:.6} management_fees={:.2}",
                                portfolio.management_fees()
                            ),
                        },
                    )?;
                }
                crate::strategy::OrderIntent::CancelOrder { order_id, reason } => {
                    let report = self.cancel_futures_open_order(execution_date, order_id, &reason);
                    if report.order_events.is_empty() && report.process_events.is_empty() {
                        retained
                            .push(crate::strategy::OrderIntent::CancelOrder { order_id, reason });
                    } else {
                        merge_futures_report(directive_report, report);
                    }
                }
                crate::strategy::OrderIntent::CancelSymbol { symbol, reason } => {
                    let report = self.cancel_futures_open_orders_for_symbol(
                        execution_date,
                        &symbol,
                        &reason,
                    );
                    if report.order_events.is_empty() && report.process_events.is_empty() {
                        retained
                            .push(crate::strategy::OrderIntent::CancelSymbol { symbol, reason });
                    } else {
                        merge_futures_report(directive_report, report);
                    }
                }
                crate::strategy::OrderIntent::CancelAll { reason } => {
                    let report = self.cancel_all_futures_open_orders(execution_date, &reason);
                    let has_stock_open_orders = !self.broker.open_order_views().is_empty();
                    if has_stock_open_orders || report.order_events.is_empty() {
                        retained.push(crate::strategy::OrderIntent::CancelAll {
                            reason: reason.clone(),
                        });
                    }
                    merge_futures_report(directive_report, report);
                }
                crate::strategy::OrderIntent::Futures { intent } => {
                    let order_id = self.next_futures_order_id;
                    self.next_futures_order_id += 1;
                    let report = self.submit_futures_order(execution_date, order_id, intent, false);
                    decision.diagnostics.push(format!(
                        "futures_order order_id={order_id} events={}",
                        report.order_events.len()
                    ));
                    merge_futures_report(directive_report, report);
                }
                other => retained.push(other),
            }
        }
        decision.order_intents = retained;
        Ok(())
    }

    fn open_order_views(&self) -> Vec<crate::strategy::OpenOrderView> {
        let mut views = self.broker.open_order_views();
        views.extend(
            self.futures_open_orders
                .iter()
                .map(|order| crate::strategy::OpenOrderView {
                    order_id: order.order_id,
                    symbol: order.intent.symbol.clone(),
                    side: order.intent.side(),
                    requested_quantity: order.requested_quantity,
                    filled_quantity: order.filled_quantity,
                    remaining_quantity: order.remaining_quantity,
                    unfilled_quantity: order.remaining_quantity,
                    status: OrderStatus::Pending,
                    avg_price: 0.0,
                    transaction_cost: 0.0,
                    limit_price: order.limit_price,
                    reason: order.reason.clone(),
                }),
        );
        views.sort_by_key(|order| order.order_id);
        views
    }

    fn aggregate_initial_cash(&self) -> f64 {
        self.config.initial_cash
            + self
                .futures_account
                .as_ref()
                .map(FuturesAccountState::starting_cash)
                .unwrap_or(0.0)
    }

    fn aggregate_cash(&self, portfolio: &PortfolioState) -> f64 {
        portfolio.cash()
            + self
                .futures_account
                .as_ref()
                .map(FuturesAccountState::cash)
                .unwrap_or(0.0)
    }

    fn aggregate_market_value(&self, portfolio: &PortfolioState) -> f64 {
        portfolio.market_value()
            + self
                .futures_account
                .as_ref()
                .map(FuturesAccountState::position_equity)
                .unwrap_or(0.0)
    }

    fn aggregate_total_equity(&self, portfolio: &PortfolioState) -> f64 {
        portfolio.total_equity()
            + self
                .futures_account
                .as_ref()
                .map(FuturesAccountState::total_value)
                .unwrap_or(0.0)
    }

    fn submit_futures_order(
        &mut self,
        date: NaiveDate,
        order_id: u64,
        intent: FuturesOrderIntent,
        from_pending: bool,
    ) -> FuturesExecutionReport {
        let Some(_) = self.futures_account.as_ref() else {
            return self.reject_futures_order(
                date,
                order_id,
                intent,
                "futures account is not enabled".to_string(),
            );
        };

        let original_requested = intent.quantity;
        let mut intent = self.resolve_futures_trading_parameters(date, intent);
        if let Some(reason) = self.validate_futures_submission(date, &intent) {
            return self.reject_futures_order(date, order_id, intent, reason);
        }

        let fill = self.resolve_futures_fill(date, &intent);
        let Some((execution_price, fill_quantity)) = fill else {
            if intent.allow_pending || intent.limit_price.is_some() {
                return self.queue_futures_order(
                    date,
                    order_id,
                    intent,
                    original_requested,
                    0,
                    from_pending,
                    "limit not matched or no executable futures price",
                );
            }
            return self.reject_futures_order(
                date,
                order_id,
                intent,
                "missing executable futures price".to_string(),
            );
        };
        if fill_quantity == 0 {
            if intent.allow_pending || intent.limit_price.is_some() {
                return self.queue_futures_order(
                    date,
                    order_id,
                    intent,
                    original_requested,
                    0,
                    from_pending,
                    "futures liquidity unavailable",
                );
            }
            return self.reject_futures_order(
                date,
                order_id,
                intent,
                "futures liquidity unavailable".to_string(),
            );
        }

        let remaining = original_requested.saturating_sub(fill_quantity);
        intent.price = execution_price;
        intent.quantity = fill_quantity;
        intent = self.resolve_futures_transaction_cost(date, intent);
        let mut report = self
            .futures_account
            .as_mut()
            .expect("checked futures account")
            .execute_order(date, Some(order_id), intent.clone());

        if remaining > 0 && (intent.allow_pending || intent.limit_price.is_some()) {
            for event in &mut report.order_events {
                if event.order_id == Some(order_id) {
                    event.requested_quantity = original_requested;
                    event.filled_quantity = fill_quantity;
                    event.status = OrderStatus::PartiallyFilled;
                }
            }
            let mut remaining_intent = intent.clone();
            remaining_intent.quantity = remaining;
            remaining_intent.transaction_cost = 0.0;
            let queued = self.queue_futures_order(
                date,
                order_id,
                remaining_intent,
                original_requested,
                fill_quantity,
                true,
                "partial fill remaining quantity pending",
            );
            report.order_events.extend(queued.order_events);
            report.process_events.extend(queued.process_events);
            report.diagnostics.extend(queued.diagnostics);
        } else if remaining > 0 {
            for event in &mut report.order_events {
                if event.order_id == Some(order_id) {
                    event.requested_quantity = original_requested;
                    event.filled_quantity = fill_quantity;
                    event.status = OrderStatus::PartiallyFilled;
                    event.reason.push_str(": remaining quantity canceled");
                }
            }
        }
        report
    }

    fn process_futures_open_orders(&mut self, date: NaiveDate) -> BrokerExecutionReport {
        let pending = std::mem::take(&mut self.futures_open_orders);
        let mut combined = BrokerExecutionReport::default();
        for mut order in pending {
            order.intent.quantity = order.remaining_quantity;
            let report = self.submit_futures_order(date, order.order_id, order.intent, true);
            merge_futures_report(&mut combined, report);
        }
        combined
    }

    fn queue_futures_order(
        &mut self,
        date: NaiveDate,
        order_id: u64,
        intent: FuturesOrderIntent,
        requested_quantity: u32,
        filled_quantity: u32,
        _from_pending: bool,
        reason: &str,
    ) -> FuturesExecutionReport {
        let mut report = FuturesExecutionReport::default();
        let side = intent.side();
        let limit_price = intent.limit_price.unwrap_or(intent.price);
        self.futures_open_orders.push(FuturesOpenOrder {
            order_id,
            requested_quantity,
            filled_quantity,
            remaining_quantity: intent.quantity,
            limit_price,
            reason: format!("{}: {reason}", intent.reason),
            intent,
        });
        report.order_events.push(OrderEvent {
            date,
            order_id: Some(order_id),
            symbol: self
                .futures_open_orders
                .last()
                .map(|order| order.intent.symbol.clone())
                .unwrap_or_default(),
            side,
            requested_quantity,
            filled_quantity,
            status: OrderStatus::Pending,
            reason: reason.to_string(),
        });
        report.process_events.push(ProcessEvent {
            date,
            kind: ProcessEventKind::OrderCreationPass,
            order_id: Some(order_id),
            symbol: self
                .futures_open_orders
                .last()
                .map(|order| order.intent.symbol.clone()),
            side: Some(side),
            detail: format!("futures pending limit_price={limit_price:.6} reason={reason}"),
        });
        report
    }

    fn reject_futures_order(
        &self,
        date: NaiveDate,
        order_id: u64,
        intent: FuturesOrderIntent,
        reason: String,
    ) -> FuturesExecutionReport {
        let side = intent.side();
        let mut report = FuturesExecutionReport::default();
        report.order_events.push(OrderEvent {
            date,
            order_id: Some(order_id),
            symbol: intent.symbol.clone(),
            side,
            requested_quantity: intent.quantity,
            filled_quantity: 0,
            status: OrderStatus::Rejected,
            reason: format!(
                "{}: {reason} direction={} effect={}",
                intent.reason,
                intent.direction.as_str(),
                intent.effect.as_str()
            ),
        });
        report.process_events.push(ProcessEvent {
            date,
            kind: ProcessEventKind::OrderCreationReject,
            order_id: Some(order_id),
            symbol: Some(intent.symbol),
            side: Some(side),
            detail: reason,
        });
        report
    }

    fn validate_futures_submission(
        &self,
        date: NaiveDate,
        intent: &FuturesOrderIntent,
    ) -> Option<String> {
        if intent.quantity == 0 {
            return Some("zero futures quantity".to_string());
        }
        if self.futures_validation_config.enforce_active_instrument {
            if let Some(instrument) = self.data.instrument(&intent.symbol) {
                if !instrument.is_active_on(date) {
                    return Some(format!(
                        "inactive futures instrument symbol={} date={date}",
                        intent.symbol
                    ));
                }
            }
        }
        if self.futures_validation_config.enforce_trading_phase {
            if let Some(snapshot) = self.data.market(date, &intent.symbol) {
                if snapshot.paused {
                    return Some(format!(
                        "paused futures instrument symbol={}",
                        intent.symbol
                    ));
                }
                if !futures_trading_phase_allows_orders(snapshot.trading_phase.as_deref()) {
                    return Some(format!(
                        "futures trading phase does not allow orders symbol={} phase={}",
                        intent.symbol,
                        snapshot.trading_phase.as_deref().unwrap_or("")
                    ));
                }
            }
        }
        if let Some(limit_price) = intent.limit_price {
            if !limit_price.is_finite() || limit_price <= 0.0 {
                return Some("invalid futures limit price".to_string());
            }
            if self.futures_validation_config.enforce_limit_price_tick {
                let tick = self.futures_price_tick(date, &intent.symbol);
                if !price_is_tick_aligned(limit_price, tick) {
                    return Some(format!(
                        "futures limit price not aligned to tick symbol={} price={limit_price:.6} tick={tick:.6}",
                        intent.symbol
                    ));
                }
            }
            if self.futures_validation_config.enforce_price_limits {
                if let Some(snapshot) = self.data.market(date, &intent.symbol) {
                    if snapshot.upper_limit.is_finite()
                        && snapshot.upper_limit > 0.0
                        && limit_price > snapshot.upper_limit + 1e-9
                    {
                        return Some(format!(
                            "futures limit price above upper limit symbol={} price={limit_price:.6} upper={:.6}",
                            intent.symbol, snapshot.upper_limit
                        ));
                    }
                    if snapshot.lower_limit.is_finite()
                        && snapshot.lower_limit > 0.0
                        && limit_price < snapshot.lower_limit - 1e-9
                    {
                        return Some(format!(
                            "futures limit price below lower limit symbol={} price={limit_price:.6} lower={:.6}",
                            intent.symbol, snapshot.lower_limit
                        ));
                    }
                }
            }
            for order in &self.futures_open_orders {
                if order.intent.symbol != intent.symbol || order.intent.side() == intent.side() {
                    continue;
                }
                let existing_limit = order.limit_price;
                let crosses = match intent.side() {
                    OrderSide::Buy => limit_price >= existing_limit,
                    OrderSide::Sell => limit_price <= existing_limit,
                };
                if crosses {
                    return Some(format!(
                        "self-trade risk with futures open order {}",
                        order.order_id
                    ));
                }
            }
        }
        None
    }

    fn futures_price_tick(&self, date: NaiveDate, symbol: &str) -> f64 {
        self.data
            .futures_trading_parameter(date, symbol)
            .map(|params| params.price_tick)
            .filter(|tick| tick.is_finite() && *tick > 0.0)
            .or_else(|| {
                self.data
                    .market(date, symbol)
                    .map(|snapshot| snapshot.effective_price_tick())
            })
            .unwrap_or(1.0)
            .max(1e-9)
    }

    fn resolve_futures_trading_parameters(
        &self,
        date: NaiveDate,
        mut intent: FuturesOrderIntent,
    ) -> FuturesOrderIntent {
        if let Some(params) = self.data.futures_trading_parameter(date, &intent.symbol) {
            intent.spec = params.spec();
        }
        intent
    }

    fn resolve_futures_transaction_cost(
        &self,
        date: NaiveDate,
        mut intent: FuturesOrderIntent,
    ) -> FuturesOrderIntent {
        if intent.transaction_cost > 0.0 {
            return intent;
        }
        if let Some(params) = self.data.futures_trading_parameter(date, &intent.symbol) {
            let close_today_quantity = self.futures_close_today_quantity(&intent);
            intent.transaction_cost = self.futures_cost_model.calculate(
                params,
                intent.effect,
                intent.price,
                intent.quantity,
                close_today_quantity,
            );
        }
        intent
    }

    fn futures_close_today_quantity(&self, intent: &FuturesOrderIntent) -> u32 {
        match intent.effect {
            FuturesPositionEffect::Open | FuturesPositionEffect::CloseYesterday => 0,
            FuturesPositionEffect::CloseToday => intent.quantity,
            FuturesPositionEffect::Close => self
                .futures_account
                .as_ref()
                .and_then(|account| account.position(&intent.symbol, intent.direction))
                .map(|position| intent.quantity.saturating_sub(position.old_quantity))
                .unwrap_or(0),
        }
    }

    fn resolve_futures_fill(
        &self,
        date: NaiveDate,
        intent: &FuturesOrderIntent,
    ) -> Option<(f64, u32)> {
        if self.broker.execution_price_field() == PriceField::Last {
            if let Some(fill) = self.resolve_futures_intraday_fill(date, intent) {
                return Some(fill);
            }
        }
        if let Some(snapshot) = self.data.market(date, &intent.symbol) {
            if snapshot.paused {
                return None;
            }
            let price = match self.broker.execution_price_field() {
                PriceField::DayOpen => snapshot.day_open,
                PriceField::Open => snapshot.open,
                PriceField::Close => snapshot.close,
                PriceField::Last => match intent.side() {
                    OrderSide::Buy => snapshot.buy_price(PriceField::Last),
                    OrderSide::Sell => snapshot.sell_price(PriceField::Last),
                },
            };
            if !self.futures_price_can_trade(snapshot, intent.side(), price, intent.limit_price) {
                return None;
            }
            return Some((price, intent.quantity));
        }
        if intent.price.is_finite() && intent.price > 0.0 {
            if futures_limit_satisfied(intent.side(), intent.price, intent.limit_price) {
                return Some((intent.price, intent.quantity));
            }
        }
        None
    }

    fn resolve_futures_intraday_fill(
        &self,
        date: NaiveDate,
        intent: &FuturesOrderIntent,
    ) -> Option<(f64, u32)> {
        let snapshot = self.data.market(date, &intent.symbol);
        if matches!(
            self.broker.matching_type(),
            MatchingType::MinuteBestCounterparty
        ) {
            let depth = self.data.order_book_depth_on(date, &intent.symbol);
            if !depth.is_empty() {
                return self.resolve_futures_depth_fill(date, intent, snapshot);
            }
        }
        let quotes = self.data.execution_quotes_on(date, &intent.symbol);
        for quote in quotes {
            let price = match self.broker.matching_type() {
                MatchingType::MinuteBestOwn => match intent.side() {
                    OrderSide::Buy => {
                        if quote.bid1.is_finite() && quote.bid1 > 0.0 {
                            quote.bid1
                        } else {
                            quote.last_price
                        }
                    }
                    OrderSide::Sell => {
                        if quote.ask1.is_finite() && quote.ask1 > 0.0 {
                            quote.ask1
                        } else {
                            quote.last_price
                        }
                    }
                },
                MatchingType::MinuteBestCounterparty => match intent.side() {
                    OrderSide::Buy => quote.buy_price().unwrap_or(quote.last_price),
                    OrderSide::Sell => quote.sell_price().unwrap_or(quote.last_price),
                },
                _ => quote.last_price,
            };
            if let Some(snapshot) = snapshot {
                if !self.futures_price_can_trade(snapshot, intent.side(), price, intent.limit_price)
                {
                    continue;
                }
            } else if !futures_limit_satisfied(intent.side(), price, intent.limit_price) {
                continue;
            }
            let top_level_quantity = match intent.side() {
                OrderSide::Buy => quote.ask1_volume,
                OrderSide::Sell => quote.bid1_volume,
            }
            .max(quote.volume_delta)
            .min(u32::MAX as u64) as u32;
            let fill_quantity = if top_level_quantity == 0 {
                intent.quantity
            } else {
                intent.quantity.min(top_level_quantity)
            };
            if price.is_finite() && price > 0.0 && fill_quantity > 0 {
                return Some((price, fill_quantity));
            }
        }
        None
    }

    fn resolve_futures_depth_fill(
        &self,
        date: NaiveDate,
        intent: &FuturesOrderIntent,
        snapshot: Option<&crate::data::DailyMarketSnapshot>,
    ) -> Option<(f64, u32)> {
        let depth = self.data.order_book_depth_on(date, &intent.symbol);
        let mut cursor = 0usize;
        while cursor < depth.len() {
            let timestamp = depth[cursor].timestamp;
            let start = cursor;
            while cursor < depth.len() && depth[cursor].timestamp == timestamp {
                cursor += 1;
            }
            let mut levels = depth[start..cursor].iter().collect::<Vec<_>>();
            levels.sort_by(|left, right| left.level.cmp(&right.level));

            let mut filled_quantity = 0_u32;
            let mut gross_amount = 0.0_f64;
            for level in levels {
                let Some(price) = level.executable_price(intent.side()) else {
                    continue;
                };
                let can_trade = if let Some(snapshot) = snapshot {
                    self.futures_price_can_trade(snapshot, intent.side(), price, intent.limit_price)
                } else {
                    futures_limit_satisfied(intent.side(), price, intent.limit_price)
                };
                if !can_trade {
                    if intent.limit_price.is_some() {
                        break;
                    }
                    continue;
                }
                let available_quantity =
                    level.executable_volume(intent.side()).min(u32::MAX as u64) as u32;
                if available_quantity == 0 {
                    continue;
                }
                let remaining = intent.quantity.saturating_sub(filled_quantity);
                if remaining == 0 {
                    break;
                }
                let take_quantity = remaining.min(available_quantity);
                gross_amount += price * take_quantity as f64;
                filled_quantity += take_quantity;
                if filled_quantity >= intent.quantity {
                    break;
                }
            }
            if filled_quantity > 0 {
                return Some((gross_amount / filled_quantity as f64, filled_quantity));
            }
        }
        None
    }

    fn futures_price_can_trade(
        &self,
        snapshot: &crate::data::DailyMarketSnapshot,
        side: OrderSide,
        price: f64,
        limit_price: Option<f64>,
    ) -> bool {
        if !price.is_finite() || price <= 0.0 {
            return false;
        }
        if !futures_limit_satisfied(side, price, limit_price) {
            return false;
        }
        match side {
            OrderSide::Buy => !snapshot.is_at_upper_limit_price(price),
            OrderSide::Sell => !snapshot.is_at_lower_limit_price(price),
        }
    }

    fn cancel_futures_open_order(
        &mut self,
        date: NaiveDate,
        order_id: u64,
        reason: &str,
    ) -> FuturesExecutionReport {
        let Some(index) = self
            .futures_open_orders
            .iter()
            .position(|order| order.order_id == order_id)
        else {
            return FuturesExecutionReport::default();
        };
        let order = self.futures_open_orders.remove(index);
        futures_cancel_report(date, order, reason)
    }

    fn cancel_futures_open_orders_for_symbol(
        &mut self,
        date: NaiveDate,
        symbol: &str,
        reason: &str,
    ) -> FuturesExecutionReport {
        let mut report = FuturesExecutionReport::default();
        let mut retained = Vec::with_capacity(self.futures_open_orders.len());
        let mut canceled = Vec::new();
        for order in self.futures_open_orders.drain(..) {
            if order.intent.symbol == symbol {
                canceled.push(order);
            } else {
                retained.push(order);
            }
        }
        self.futures_open_orders = retained;
        for order in canceled {
            merge_futures_execution_report(&mut report, futures_cancel_report(date, order, reason));
        }
        report
    }

    fn cancel_all_futures_open_orders(
        &mut self,
        date: NaiveDate,
        reason: &str,
    ) -> FuturesExecutionReport {
        let mut report = FuturesExecutionReport::default();
        for order in std::mem::take(&mut self.futures_open_orders) {
            merge_futures_execution_report(&mut report, futures_cancel_report(date, order, reason));
        }
        report
    }

    pub fn run(&mut self) -> Result<BacktestResult, BacktestError> {
        self.run_with_progress(|_| {})
    }

    pub fn run_with_progress<F>(
        &mut self,
        mut on_progress: F,
    ) -> Result<BacktestResult, BacktestError>
    where
        F: FnMut(&BacktestDayProgress),
    {
        let mut portfolio = PortfolioState::new(self.config.initial_cash);
        let scheduler_calendar = self.data.calendar().clone();
        let scheduler = Scheduler::new(&scheduler_calendar);
        let execution_dates = self
            .data
            .calendar()
            .iter()
            .filter(|date| {
                self.config
                    .start_date
                    .map(|start| *date >= start)
                    .unwrap_or(true)
            })
            .filter(|date| self.config.end_date.map(|end| *date <= end).unwrap_or(true))
            .filter(|date| {
                !self.data.factor_snapshots_on(*date).is_empty()
                    && !self.data.candidate_snapshots_on(*date).is_empty()
            })
            .collect::<Vec<_>>();
        let mut result = BacktestResult {
            strategy_name: self.strategy.name().to_string(),
            benchmark_series: self
                .data
                .benchmark_series()
                .into_iter()
                .filter(|row| {
                    self.config
                        .start_date
                        .map(|start| row.date >= start)
                        .unwrap_or(true)
                })
                .filter(|row| {
                    self.config
                        .end_date
                        .map(|end| row.date <= end)
                        .unwrap_or(true)
                })
                .collect(),
            risk_decisions: Vec::new(),
            order_events: Vec::new(),
            fills: Vec::new(),
            position_events: Vec::new(),
            account_events: Vec::new(),
            process_events: Vec::new(),
            equity_curve: Vec::new(),
            holdings_summary: Vec::new(),
            daily_holdings: Vec::new(),
            metrics: BacktestMetrics::default(),
        };

        for (execution_idx, execution_date) in execution_dates.iter().copied().enumerate() {
            let mut corporate_action_notes = Vec::new();
            portfolio.begin_trading_day();
            if let Some(account) = self.futures_account.as_mut() {
                account.begin_trading_day();
            }
            let pending_cash_flow_report = self.settle_pending_cash_flows(
                execution_date,
                &mut portfolio,
                &mut corporate_action_notes,
            );
            self.extend_result(&mut result, pending_cash_flow_report);
            let corporate_action_report = self.apply_corporate_actions(
                execution_date,
                &mut portfolio,
                &mut corporate_action_notes,
            )?;
            self.extend_result(&mut result, corporate_action_report);
            let receivable_report = self.settle_cash_receivables(
                execution_date,
                &mut portfolio,
                &mut corporate_action_notes,
            )?;
            self.extend_result(&mut result, receivable_report);
            let delisting_report = self.settle_delisted_positions(
                execution_date,
                &mut portfolio,
                &mut corporate_action_notes,
            )?;
            self.extend_result(&mut result, delisting_report);
            let futures_open_order_report = self.process_futures_open_orders(execution_date);
            self.extend_result(&mut result, futures_open_order_report);

            let decision_slot = execution_idx
                .checked_sub(self.config.decision_lag_trading_days)
                .map(|decision_idx| (decision_idx, execution_dates[decision_idx]));
            let Some((decision_index, decision_date)) = decision_slot else {
                let mut process_events = Vec::new();
                let mut report = BrokerExecutionReport::default();
                portfolio.update_prices_with_options(
                    execution_date,
                    &self.data,
                    PriceField::Close,
                    self.broker.same_day_buy_close_mark_at_fill(),
                )?;
                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);
                merge_broker_report(&mut report, futures_daily_settlement_report);
                let futures_expiration_report = self.settle_futures_expirations(execution_date);
                merge_broker_report(&mut report, futures_expiration_report);

                let daily_fill_count = report.fill_events.len();
                let day_orders = report.order_events.clone();
                let day_fills = report.fill_events.clone();
                let broker_diagnostics = report.diagnostics.clone();
                self.extend_result(&mut result, report);

                let benchmark =
                    self.data
                        .benchmark(execution_date)
                        .ok_or(BacktestError::MissingBenchmark {
                            date: execution_date,
                        })?;
                let notes = corporate_action_notes.join(" | ");
                let diagnostics = std::iter::once(format!(
                    "decision_lag_warmup lag_days={} execution_index={}",
                    self.config.decision_lag_trading_days, execution_idx
                ))
                .chain(broker_diagnostics.into_iter())
                .collect::<Vec<_>>()
                .join(" | ");
                let holdings_for_day = portfolio.holdings_summary(execution_date);
                let day_process_events = process_events.clone();
                let aggregate_initial_cash = self.aggregate_initial_cash();
                let aggregate_cash = self.aggregate_cash(&portfolio);
                let aggregate_market_value = self.aggregate_market_value(&portfolio);
                let aggregate_total_equity = self.aggregate_total_equity(&portfolio);

                result.equity_curve.push(DailyEquityPoint {
                    date: execution_date,
                    cash: aggregate_cash,
                    market_value: aggregate_market_value,
                    total_equity: aggregate_total_equity,
                    benchmark_close: benchmark.close,
                    benchmark_prev_close: benchmark.prev_close,
                    notes,
                    diagnostics,
                });
                result.daily_holdings.extend(holdings_for_day.clone());
                let latest = result
                    .equity_curve
                    .last()
                    .expect("equity point pushed for progress event");
                on_progress(&BacktestDayProgress {
                    date: execution_date,
                    cash: latest.cash,
                    market_value: latest.market_value,
                    total_equity: latest.total_equity,
                    unit_nav: if aggregate_initial_cash.abs() < f64::EPSILON {
                        0.0
                    } else {
                        latest.total_equity / aggregate_initial_cash
                    },
                    total_return: if aggregate_initial_cash.abs() < f64::EPSILON {
                        0.0
                    } else {
                        (latest.total_equity / aggregate_initial_cash) - 1.0
                    },
                    benchmark_close: latest.benchmark_close,
                    daily_fill_count,
                    cumulative_trade_count: result.fills.len(),
                    holding_count: holdings_for_day.len(),
                    notes: latest.notes.clone(),
                    diagnostics: latest.diagnostics.clone(),
                    orders: day_orders,
                    fills: day_fills,
                    holdings: holdings_for_day,
                    process_events: day_process_events,
                });
                result.process_events.append(&mut process_events);
                continue;
            };
            let mut process_events = Vec::new();
            let mut directive_report = BrokerExecutionReport::default();
            let pre_open_orders = self.open_order_views();
            let schedule_rules = self.strategy.schedule_rules();
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &pre_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::PreBeforeTrading,
                "before_trading:pre",
            )?;
            self.strategy.before_trading(&StrategyContext {
                execution_date,
                decision_date,
                decision_index,
                data: &self.data,
                portfolio: &portfolio,
                futures_account: self.futures_account.as_ref(),
                open_orders: &pre_open_orders,
                dynamic_universe: self.dynamic_universe.as_ref(),
                subscriptions: &self.subscriptions,
                process_events: &process_events,
                active_process_event: None,
                active_datetime: stage_datetime(
                    decision_date,
                    default_stage_time(ScheduleStage::BeforeTrading),
                ),
                order_events: result.order_events.as_slice(),
                fills: result.fills.as_slice(),
            })?;
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &pre_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::BeforeTrading,
                "before_trading",
            )?;
            let mut before_trading_decision = collect_scheduled_decisions(
                &mut self.strategy,
                &scheduler,
                execution_date,
                ScheduleStage::BeforeTrading,
                &schedule_rules,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &pre_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                &mut self.process_event_bus,
                default_stage_time(ScheduleStage::BeforeTrading),
                result.order_events.as_slice(),
                result.fills.as_slice(),
            )?;
            self.apply_strategy_directives(
                execution_date,
                decision_date,
                decision_index,
                &mut portfolio,
                &pre_open_orders,
                &mut process_events,
                &mut before_trading_decision,
                &mut directive_report,
            )?;
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &pre_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::PostBeforeTrading,
                "before_trading:post",
            )?;
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &pre_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::PreOpenAuction,
                "open_auction:pre",
            )?;
            let mut auction_decision = collect_scheduled_decisions(
                &mut self.strategy,
                &scheduler,
                execution_date,
                ScheduleStage::OpenAuction,
                &schedule_rules,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &pre_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                &mut self.process_event_bus,
                default_stage_time(ScheduleStage::OpenAuction),
                result.order_events.as_slice(),
                result.fills.as_slice(),
            )?;
            auction_decision.merge_from(self.strategy.open_auction(&StrategyContext {
                execution_date,
                decision_date,
                decision_index,
                data: &self.data,
                portfolio: &portfolio,
                futures_account: self.futures_account.as_ref(),
                open_orders: &pre_open_orders,
                dynamic_universe: self.dynamic_universe.as_ref(),
                subscriptions: &self.subscriptions,
                process_events: &process_events,
                active_process_event: None,
                active_datetime: stage_datetime(
                    decision_date,
                    default_stage_time(ScheduleStage::OpenAuction),
                ),
                order_events: result.order_events.as_slice(),
                fills: result.fills.as_slice(),
            })?);
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &pre_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::OpenAuction,
                "open_auction",
            )?;
            self.apply_strategy_directives(
                execution_date,
                decision_date,
                decision_index,
                &mut portfolio,
                &pre_open_orders,
                &mut process_events,
                &mut auction_decision,
                &mut directive_report,
            )?;
            let pre_auction_execution_orders = self.open_order_views();
            self.ensure_execution_quotes_for_decision(
                execution_date,
                &portfolio,
                &pre_auction_execution_orders,
                &auction_decision,
                None,
                None,
            )?;
            let mut report = self.broker.execute(
                execution_date,
                &mut portfolio,
                &self.data,
                &auction_decision,
            )?;
            let post_auction_open_orders = self.open_order_views();
            publish_process_events(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_auction_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                &mut report.process_events,
            )?;
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_auction_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::PostOpenAuction,
                "open_auction:post",
            )?;

            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_auction_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::PreOnDay,
                "on_day:pre",
            )?;
            let on_day_open_orders = self.open_order_views();
            let decision_quote_times = self.strategy.decision_quote_times();
            if !decision_quote_times.is_empty() {
                let decision_quote_symbols =
                    self.strategy.decision_quote_symbols(&StrategyContext {
                        execution_date,
                        decision_date,
                        decision_index,
                        data: &self.data,
                        portfolio: &portfolio,
                        futures_account: self.futures_account.as_ref(),
                        open_orders: &on_day_open_orders,
                        dynamic_universe: self.dynamic_universe.as_ref(),
                        subscriptions: &self.subscriptions,
                        process_events: &process_events,
                        active_process_event: None,
                        active_datetime: stage_datetime(
                            decision_date,
                            default_stage_time(ScheduleStage::OnDay),
                        ),
                        order_events: result.order_events.as_slice(),
                        fills: result.fills.as_slice(),
                    })?;
                self.ensure_execution_quotes_for_symbols_at_times(
                    execution_date,
                    &decision_quote_symbols,
                    &decision_quote_times,
                )?;
            }
            self.ensure_execution_quotes_for_portfolio_times(
                execution_date,
                &portfolio,
                &decision_quote_times,
            )?;
            let mut decision = decision_slot
                .map(|(decision_idx, decision_date)| {
                    self.strategy.on_day(&StrategyContext {
                        execution_date,
                        decision_date,
                        decision_index: decision_idx,
                        data: &self.data,
                        portfolio: &portfolio,
                        futures_account: self.futures_account.as_ref(),
                        open_orders: &on_day_open_orders,
                        dynamic_universe: self.dynamic_universe.as_ref(),
                        subscriptions: &self.subscriptions,
                        process_events: &process_events,
                        active_process_event: None,
                        active_datetime: stage_datetime(
                            decision_date,
                            default_stage_time(ScheduleStage::OnDay),
                        ),
                        order_events: result.order_events.as_slice(),
                        fills: result.fills.as_slice(),
                    })
                })
                .transpose()?
                .unwrap_or_default();
            decision.merge_from(collect_scheduled_decisions(
                &mut self.strategy,
                &scheduler,
                execution_date,
                ScheduleStage::OnDay,
                &schedule_rules,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &on_day_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                &mut self.process_event_bus,
                default_stage_time(ScheduleStage::OnDay),
                result.order_events.as_slice(),
                result.fills.as_slice(),
            )?);
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &on_day_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::OnDay,
                "on_day",
            )?;
            let bar_open_orders = self.open_order_views();
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &bar_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::PreBar,
                "bar:pre",
            )?;
            decision.merge_from(collect_scheduled_decisions(
                &mut self.strategy,
                &scheduler,
                execution_date,
                ScheduleStage::Bar,
                &schedule_rules,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &bar_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                &mut self.process_event_bus,
                default_stage_time(ScheduleStage::Bar),
                result.order_events.as_slice(),
                result.fills.as_slice(),
            )?);
            decision.merge_from(self.strategy.on_bar(&StrategyContext {
                execution_date,
                decision_date,
                decision_index,
                data: &self.data,
                portfolio: &portfolio,
                futures_account: self.futures_account.as_ref(),
                open_orders: &bar_open_orders,
                dynamic_universe: self.dynamic_universe.as_ref(),
                subscriptions: &self.subscriptions,
                process_events: &process_events,
                active_process_event: None,
                active_datetime: stage_datetime(
                    decision_date,
                    default_stage_time(ScheduleStage::Bar),
                ),
                order_events: result.order_events.as_slice(),
                fills: result.fills.as_slice(),
            })?);
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &bar_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::Bar,
                "bar",
            )?;
            self.apply_strategy_directives(
                execution_date,
                decision_date,
                decision_index,
                &mut portfolio,
                &on_day_open_orders,
                &mut process_events,
                &mut decision,
                &mut directive_report,
            )?;

            let pre_intraday_execution_orders = self.open_order_views();
            self.ensure_execution_quotes_for_decision(
                execution_date,
                &portfolio,
                &pre_intraday_execution_orders,
                &decision,
                None,
                None,
            )?;
            let mut intraday_report =
                self.broker
                    .execute(execution_date, &mut portfolio, &self.data, &decision)?;
            let post_intraday_open_orders = self.open_order_views();
            publish_process_events(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_intraday_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                &mut intraday_report.process_events,
            )?;
            report.order_events.extend(intraday_report.order_events);
            report.fill_events.extend(intraday_report.fill_events);
            report
                .position_events
                .extend(intraday_report.position_events);
            report.account_events.extend(intraday_report.account_events);
            report.diagnostics.extend(intraday_report.diagnostics);
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_intraday_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::PostOnDay,
                "on_day:post",
            )?;
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_intraday_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::PostBar,
                "bar:post",
            )?;

            if should_run_minute_events(&schedule_rules, &self.subscriptions) {
                let filter_by_subscription = !self.subscriptions.is_empty();
                let minute_quotes = self
                    .data
                    .execution_quotes_on_date(execution_date)
                    .into_iter()
                    .filter(|quote| {
                        !filter_by_subscription || self.subscriptions.contains(&quote.symbol)
                    })
                    .collect::<Vec<_>>();
                for quote in minute_quotes {
                    let minute_time = quote.timestamp.time();
                    let minute_open_orders = self.open_order_views();
                    publish_phase_event(
                        &mut self.strategy,
                        &mut self.process_event_bus,
                        execution_date,
                        decision_date,
                        decision_index,
                        &self.data,
                        &portfolio,
                        self.futures_account.as_ref(),
                        &minute_open_orders,
                        self.dynamic_universe.as_ref(),
                        &self.subscriptions,
                        &mut process_events,
                        execution_date,
                        ProcessEventKind::PreMinute,
                        format!("minute:{}:{}:pre", quote.symbol, quote.timestamp),
                    )?;
                    let mut minute_decision = collect_scheduled_decisions(
                        &mut self.strategy,
                        &scheduler,
                        execution_date,
                        ScheduleStage::Minute,
                        &schedule_rules,
                        decision_date,
                        decision_index,
                        &self.data,
                        &portfolio,
                        self.futures_account.as_ref(),
                        &minute_open_orders,
                        self.dynamic_universe.as_ref(),
                        &self.subscriptions,
                        &mut process_events,
                        &mut self.process_event_bus,
                        Some(minute_time),
                        result.order_events.as_slice(),
                        result.fills.as_slice(),
                    )?;
                    minute_decision.merge_from(self.strategy.on_minute(
                        &StrategyContext {
                            execution_date,
                            decision_date,
                            decision_index,
                            data: &self.data,
                            portfolio: &portfolio,
                            futures_account: self.futures_account.as_ref(),
                            open_orders: &minute_open_orders,
                            dynamic_universe: self.dynamic_universe.as_ref(),
                            subscriptions: &self.subscriptions,
                            process_events: &process_events,
                            active_process_event: None,
                            active_datetime: Some(quote.timestamp),
                            order_events: result.order_events.as_slice(),
                            fills: result.fills.as_slice(),
                        },
                        &quote,
                    )?);
                    publish_phase_event(
                        &mut self.strategy,
                        &mut self.process_event_bus,
                        execution_date,
                        decision_date,
                        decision_index,
                        &self.data,
                        &portfolio,
                        self.futures_account.as_ref(),
                        &minute_open_orders,
                        self.dynamic_universe.as_ref(),
                        &self.subscriptions,
                        &mut process_events,
                        execution_date,
                        ProcessEventKind::Minute,
                        format!("minute:{}:{}", quote.symbol, quote.timestamp),
                    )?;
                    self.apply_strategy_directives(
                        execution_date,
                        decision_date,
                        decision_index,
                        &mut portfolio,
                        &minute_open_orders,
                        &mut process_events,
                        &mut minute_decision,
                        &mut directive_report,
                    )?;
                    let pre_minute_execution_orders = self.open_order_views();
                    self.ensure_execution_quotes_for_decision(
                        execution_date,
                        &portfolio,
                        &pre_minute_execution_orders,
                        &minute_decision,
                        Some(minute_time),
                        Some(minute_time),
                    )?;
                    let mut minute_report = self.broker.execute_between(
                        execution_date,
                        &mut portfolio,
                        &self.data,
                        &minute_decision,
                        Some(minute_time),
                        Some(minute_time),
                    )?;
                    let post_minute_open_orders = self.open_order_views();
                    publish_process_events(
                        &mut self.strategy,
                        &mut self.process_event_bus,
                        execution_date,
                        decision_date,
                        decision_index,
                        &self.data,
                        &portfolio,
                        self.futures_account.as_ref(),
                        &post_minute_open_orders,
                        self.dynamic_universe.as_ref(),
                        &self.subscriptions,
                        &mut process_events,
                        &mut minute_report.process_events,
                    )?;
                    merge_broker_report(&mut report, minute_report);
                    publish_phase_event(
                        &mut self.strategy,
                        &mut self.process_event_bus,
                        execution_date,
                        decision_date,
                        decision_index,
                        &self.data,
                        &portfolio,
                        self.futures_account.as_ref(),
                        &post_minute_open_orders,
                        self.dynamic_universe.as_ref(),
                        &self.subscriptions,
                        &mut process_events,
                        execution_date,
                        ProcessEventKind::PostMinute,
                        format!("minute:{}:{}:post", quote.symbol, quote.timestamp),
                    )?;
                }
            }

            portfolio.update_prices_with_options(
                execution_date,
                &self.data,
                PriceField::Close,
                self.broker.same_day_buy_close_mark_at_fill(),
            )?;

            let post_trade_open_orders = self.open_order_views();
            let visible_order_events = result
                .order_events
                .iter()
                .cloned()
                .chain(report.order_events.iter().cloned())
                .collect::<Vec<_>>();
            let visible_fills = result
                .fills
                .iter()
                .cloned()
                .chain(report.fill_events.iter().cloned())
                .collect::<Vec<_>>();
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_trade_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::PreAfterTrading,
                "after_trading:pre",
            )?;
            self.strategy.after_trading(&StrategyContext {
                execution_date,
                decision_date,
                decision_index,
                data: &self.data,
                portfolio: &portfolio,
                futures_account: self.futures_account.as_ref(),
                open_orders: &post_trade_open_orders,
                dynamic_universe: self.dynamic_universe.as_ref(),
                subscriptions: &self.subscriptions,
                process_events: &process_events,
                active_process_event: None,
                active_datetime: stage_datetime(
                    decision_date,
                    default_stage_time(ScheduleStage::AfterTrading),
                ),
                order_events: visible_order_events.as_slice(),
                fills: visible_fills.as_slice(),
            })?;
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_trade_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::AfterTrading,
                "after_trading",
            )?;
            let mut after_trading_decision = collect_scheduled_decisions(
                &mut self.strategy,
                &scheduler,
                execution_date,
                ScheduleStage::AfterTrading,
                &schedule_rules,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_trade_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                &mut self.process_event_bus,
                default_stage_time(ScheduleStage::AfterTrading),
                visible_order_events.as_slice(),
                visible_fills.as_slice(),
            )?;
            self.apply_strategy_directives(
                execution_date,
                decision_date,
                decision_index,
                &mut portfolio,
                &post_trade_open_orders,
                &mut process_events,
                &mut after_trading_decision,
                &mut directive_report,
            )?;
            let mut close_report = self.broker.after_trading(execution_date);
            publish_process_events(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_trade_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                &mut close_report.process_events,
            )?;
            report.order_events.extend(close_report.order_events);
            report.fill_events.extend(close_report.fill_events);
            report.position_events.extend(close_report.position_events);
            report.account_events.extend(close_report.account_events);
            report.diagnostics.extend(close_report.diagnostics);
            let post_close_open_orders = self.open_order_views();
            let visible_order_events_after_close = result
                .order_events
                .iter()
                .cloned()
                .chain(report.order_events.iter().cloned())
                .collect::<Vec<_>>();
            let visible_fills_after_close = result
                .fills
                .iter()
                .cloned()
                .chain(report.fill_events.iter().cloned())
                .collect::<Vec<_>>();
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_close_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::PostAfterTrading,
                "after_trading:post",
            )?;
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_close_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::PreSettlement,
                "settlement:pre",
            )?;
            self.strategy.on_settlement(&StrategyContext {
                execution_date,
                decision_date,
                decision_index,
                data: &self.data,
                portfolio: &portfolio,
                futures_account: self.futures_account.as_ref(),
                open_orders: &post_close_open_orders,
                dynamic_universe: self.dynamic_universe.as_ref(),
                subscriptions: &self.subscriptions,
                process_events: &process_events,
                active_process_event: None,
                active_datetime: stage_datetime(
                    decision_date,
                    default_stage_time(ScheduleStage::Settlement),
                ),
                order_events: visible_order_events_after_close.as_slice(),
                fills: visible_fills_after_close.as_slice(),
            })?;
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_close_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::Settlement,
                "settlement",
            )?;
            let mut settlement_decision = collect_scheduled_decisions(
                &mut self.strategy,
                &scheduler,
                execution_date,
                ScheduleStage::Settlement,
                &schedule_rules,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_close_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                &mut self.process_event_bus,
                default_stage_time(ScheduleStage::Settlement),
                visible_order_events_after_close.as_slice(),
                visible_fills_after_close.as_slice(),
            )?;
            self.apply_strategy_directives(
                execution_date,
                decision_date,
                decision_index,
                &mut portfolio,
                &post_close_open_orders,
                &mut process_events,
                &mut settlement_decision,
                &mut directive_report,
            )?;
            let futures_daily_settlement_report = self.settle_futures_daily(execution_date);
            merge_broker_report(&mut directive_report, futures_daily_settlement_report);
            let futures_expiration_report = self.settle_futures_expirations(execution_date);
            merge_broker_report(&mut directive_report, futures_expiration_report);
            let dynamic_universe_snapshot = self.dynamic_universe.clone();
            let subscriptions_snapshot = self.subscriptions.clone();
            let management_fee_report = self.apply_management_fee(
                execution_date,
                decision_date,
                decision_index,
                &mut portfolio,
                &post_close_open_orders,
                dynamic_universe_snapshot.as_ref(),
                &subscriptions_snapshot,
                &mut process_events,
                visible_order_events_after_close.as_slice(),
                visible_fills_after_close.as_slice(),
            )?;
            merge_broker_report(&mut directive_report, management_fee_report);
            publish_phase_event(
                &mut self.strategy,
                &mut self.process_event_bus,
                execution_date,
                decision_date,
                decision_index,
                &self.data,
                &portfolio,
                self.futures_account.as_ref(),
                &post_close_open_orders,
                self.dynamic_universe.as_ref(),
                &self.subscriptions,
                &mut process_events,
                execution_date,
                ProcessEventKind::PostSettlement,
                "settlement:post",
            )?;
            merge_broker_report(&mut report, directive_report);
            let daily_fill_count = report.fill_events.len();
            let day_orders = report.order_events.clone();
            let day_fills = report.fill_events.clone();
            let broker_diagnostics = report.diagnostics.clone();
            self.extend_result(&mut result, report);
            result.risk_decisions.extend(decision.risk_decisions);

            let benchmark =
                self.data
                    .benchmark(execution_date)
                    .ok_or(BacktestError::MissingBenchmark {
                        date: execution_date,
                    })?;
            let notes = corporate_action_notes
                .into_iter()
                .chain(decision.notes.into_iter())
                .collect::<Vec<_>>()
                .join(" | ");
            let diagnostics = decision
                .diagnostics
                .into_iter()
                .chain(broker_diagnostics.into_iter())
                .collect::<Vec<_>>()
                .join(" | ");
            let holdings_for_day = portfolio.holdings_summary(execution_date);
            let day_process_events = process_events.clone();
            let aggregate_initial_cash = self.aggregate_initial_cash();
            let aggregate_cash = self.aggregate_cash(&portfolio);
            let aggregate_market_value = self.aggregate_market_value(&portfolio);
            let aggregate_total_equity = self.aggregate_total_equity(&portfolio);

            result.equity_curve.push(DailyEquityPoint {
                date: execution_date,
                cash: aggregate_cash,
                market_value: aggregate_market_value,
                total_equity: aggregate_total_equity,
                benchmark_close: benchmark.close,
                benchmark_prev_close: benchmark.prev_close,
                notes,
                diagnostics,
            });
            result.daily_holdings.extend(holdings_for_day.clone());
            let latest = result
                .equity_curve
                .last()
                .expect("equity point pushed for progress event");
            on_progress(&BacktestDayProgress {
                date: execution_date,
                cash: latest.cash,
                market_value: latest.market_value,
                total_equity: latest.total_equity,
                unit_nav: if aggregate_initial_cash.abs() < f64::EPSILON {
                    0.0
                } else {
                    latest.total_equity / aggregate_initial_cash
                },
                total_return: if aggregate_initial_cash.abs() < f64::EPSILON {
                    0.0
                } else {
                    (latest.total_equity / aggregate_initial_cash) - 1.0
                },
                benchmark_close: latest.benchmark_close,
                daily_fill_count,
                cumulative_trade_count: result.fills.len(),
                holding_count: holdings_for_day.len(),
                notes: latest.notes.clone(),
                diagnostics: latest.diagnostics.clone(),
                orders: day_orders,
                fills: day_fills,
                holdings: holdings_for_day,
                process_events: day_process_events,
            });
            result.process_events.extend(process_events);
        }

        if let Some(last_date) = execution_dates.last().copied() {
            result.holdings_summary = portfolio.holdings_summary(last_date);
        }
        result.metrics = compute_backtest_metrics(
            &result.equity_curve,
            &result.fills,
            &result.daily_holdings,
            self.aggregate_initial_cash(),
        );

        Ok(result)
    }

    fn extend_result(
        &self,
        result: &mut BacktestResult,
        report: BrokerExecutionReport,
    ) -> BrokerExecutionReport {
        result.order_events.extend(report.order_events.clone());
        result.fills.extend(report.fill_events.clone());
        result
            .position_events
            .extend(report.position_events.clone());
        result.account_events.extend(report.account_events.clone());
        result.process_events.extend(report.process_events.clone());
        report
    }

    fn apply_corporate_actions(
        &self,
        date: NaiveDate,
        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 {
                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);
                    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)
    }

    fn settle_cash_receivables(
        &self,
        date: NaiveDate,
        portfolio: &mut PortfolioState,
        notes: &mut Vec<String>,
    ) -> Result<BrokerExecutionReport, BacktestError> {
        let mut report = BrokerExecutionReport::default();
        let settled = portfolio.settle_cash_receivables(date);
        for receivable in settled {
            let mut note = format!(
                "cash_receivable_settled {} ex_date={} payable_date={} cash={:.2}",
                receivable.symbol, receivable.ex_date, receivable.payable_date, receivable.amount
            );
            let cash_before = portfolio.cash() - receivable.amount;

            if self.dividend_reinvestment
                && receivable.reason.starts_with("cash_dividend")
                && receivable.amount > 0.0
            {
                let reinvest_price = portfolio
                    .position(&receivable.symbol)
                    .map(|position| position.last_price)
                    .filter(|price| price.is_finite() && *price > 0.0)
                    .or_else(|| {
                        self.data
                            .calendar()
                            .previous_day(date)
                            .and_then(|prev_date| {
                                self.data.price_on_or_before(
                                    prev_date,
                                    &receivable.symbol,
                                    PriceField::Close,
                                )
                            })
                    });
                let round_lot = self
                    .data
                    .instrument(&receivable.symbol)
                    .map(|instrument| instrument.round_lot.max(1))
                    .unwrap_or(100);
                if let Some(price) = reinvest_price {
                    let raw_quantity = (receivable.amount / price).floor() as u32;
                    let reinvest_quantity = (raw_quantity / round_lot) * round_lot;
                    if reinvest_quantity > 0 {
                        let reinvest_cash = reinvest_quantity as f64 * price;
                        let residual_cash = receivable.amount - reinvest_cash;
                        portfolio.apply_cash_delta(-reinvest_cash);
                        portfolio.position_mut(&receivable.symbol).buy(
                            date,
                            reinvest_quantity,
                            price,
                        );

                        note = format!(
                            "cash_receivable_reinvested {} ex_date={} payable_date={} cash={:.2} reinvest_qty={} reinvest_price={:.4} residual_cash={:.2}",
                            receivable.symbol,
                            receivable.ex_date,
                            receivable.payable_date,
                            receivable.amount,
                            reinvest_quantity,
                            price,
                            residual_cash
                        );
                        report.fill_events.push(FillEvent {
                            date,
                            order_id: None,
                            symbol: receivable.symbol.clone(),
                            side: OrderSide::Buy,
                            quantity: reinvest_quantity,
                            price,
                            gross_amount: reinvest_cash,
                            commission: 0.0,
                            stamp_tax: 0.0,
                            net_cash_flow: -reinvest_cash,
                            reason: "dividend_reinvestment".to_string(),
                        });
                        report.position_events.push(PositionEvent {
                            date,
                            symbol: receivable.symbol.clone(),
                            delta_quantity: reinvest_quantity as i32,
                            quantity_after: portfolio
                                .position(&receivable.symbol)
                                .map(|position| position.quantity)
                                .unwrap_or(0),
                            average_cost: portfolio
                                .position(&receivable.symbol)
                                .map(|position| position.average_cost)
                                .unwrap_or(0.0),
                            realized_pnl_delta: 0.0,
                            reason: "dividend_reinvestment".to_string(),
                        });
                        report.process_events.push(ProcessEvent {
                            date,
                            kind: ProcessEventKind::Trade,
                            order_id: None,
                            symbol: Some(receivable.symbol.clone()),
                            side: Some(OrderSide::Buy),
                            detail: format!(
                                "dividend_reinvestment quantity={} price={}",
                                reinvest_quantity, price
                            ),
                        });
                    }
                }
            }
            notes.push(note.clone());
            report.account_events.push(AccountEvent {
                date,
                cash_before,
                cash_after: portfolio.cash(),
                total_equity: portfolio.total_equity(),
                note,
            });
        }
        Ok(report)
    }

    fn settle_pending_cash_flows(
        &self,
        date: NaiveDate,
        portfolio: &mut PortfolioState,
        notes: &mut Vec<String>,
    ) -> BrokerExecutionReport {
        let mut report = BrokerExecutionReport::default();
        for flow in portfolio.settle_pending_cash_flows(date) {
            let cash_before = portfolio.cash() - flow.amount;
            let note = format!(
                "deposit_withdraw_settled amount={:.2} payable_date={} reason={}",
                flow.amount, flow.payable_date, flow.reason
            );
            notes.push(note.clone());
            report.account_events.push(AccountEvent {
                date,
                cash_before,
                cash_after: portfolio.cash(),
                total_equity: portfolio.total_equity(),
                note,
            });
        }
        report
    }

    fn settle_futures_expirations(&mut self, date: NaiveDate) -> BrokerExecutionReport {
        let mut report = BrokerExecutionReport::default();
        let Some(expirations) = self.futures_expirations.remove(&date) else {
            return report;
        };
        let Some(account) = self.futures_account.as_mut() else {
            report.diagnostics.push(format!(
                "futures_expiration_skipped date={date} reason=no_future_account count={}",
                expirations.len()
            ));
            return report;
        };
        for (symbol, settlement_price) in expirations {
            let futures_report =
                account.expire_contract(date, &symbol, settlement_price, "data_driven_expiration");
            merge_futures_report(&mut report, futures_report);
        }
        report
    }

    fn settle_futures_daily(&mut self, date: NaiveDate) -> BrokerExecutionReport {
        let mut report = BrokerExecutionReport::default();
        let Some(account) = self.futures_account.as_mut() else {
            return report;
        };
        let settlement_prices = account
            .positions()
            .values()
            .filter_map(|position| {
                self.data
                    .futures_settlement_price(
                        date,
                        &position.symbol,
                        &self.futures_settlement_price_mode,
                    )
                    .map(|price| (position.symbol.clone(), price))
            })
            .collect::<BTreeMap<_, _>>();
        if settlement_prices.is_empty() {
            return report;
        }
        let cash_before = account.total_cash();
        let cash_delta = account.settle(&settlement_prices);
        report.account_events.push(AccountEvent {
            date,
            cash_before,
            cash_after: account.total_cash(),
            total_equity: account.total_value(),
            note: format!(
                "futures_daily_settlement mode={} cash_delta={cash_delta:.2} symbols={}",
                self.futures_settlement_price_mode,
                settlement_prices
                    .keys()
                    .cloned()
                    .collect::<Vec<_>>()
                    .join(",")
            ),
        });
        report.process_events.push(ProcessEvent {
            date,
            kind: ProcessEventKind::Settlement,
            order_id: None,
            symbol: None,
            side: None,
            detail: format!(
                "futures_daily_settlement mode={} cash_delta={cash_delta:.2} count={}",
                self.futures_settlement_price_mode,
                settlement_prices.len()
            ),
        });
        report
    }

    fn apply_management_fee(
        &mut self,
        execution_date: NaiveDate,
        decision_date: NaiveDate,
        decision_index: usize,
        portfolio: &mut PortfolioState,
        open_orders: &[crate::strategy::OpenOrderView],
        dynamic_universe: Option<&BTreeSet<String>>,
        subscriptions: &BTreeSet<String>,
        process_events: &mut Vec<ProcessEvent>,
        order_events: &[OrderEvent],
        fills: &[FillEvent],
    ) -> Result<BrokerExecutionReport, BacktestError> {
        let rate = portfolio.management_fee_rate();
        if rate <= 0.0 {
            return Ok(BrokerExecutionReport::default());
        }

        let fee = self
            .strategy
            .management_fee(
                &StrategyContext {
                    execution_date,
                    decision_date,
                    decision_index,
                    data: &self.data,
                    portfolio,
                    futures_account: self.futures_account.as_ref(),
                    open_orders,
                    dynamic_universe,
                    subscriptions,
                    process_events: process_events.as_slice(),
                    active_process_event: None,
                    active_datetime: stage_datetime(
                        decision_date,
                        default_stage_time(ScheduleStage::Settlement),
                    ),
                    order_events,
                    fills,
                },
                rate,
            )?
            .unwrap_or_else(|| portfolio.default_management_fee());
        if fee <= 0.0 {
            return Ok(BrokerExecutionReport::default());
        }

        let cash_before = portfolio.cash();
        portfolio
            .apply_management_fee(fee)
            .map_err(BacktestError::Execution)?;
        let mut report = BrokerExecutionReport::default();
        report.account_events.push(AccountEvent {
            date: execution_date,
            cash_before,
            cash_after: portfolio.cash(),
            total_equity: portfolio.total_equity(),
            note: format!("management_fee rate={rate:.6} fee={fee:.2}"),
        });
        publish_custom_process_event(
            &mut self.strategy,
            &mut self.process_event_bus,
            execution_date,
            decision_date,
            decision_index,
            &self.data,
            &*portfolio,
            self.futures_account.as_ref(),
            open_orders,
            dynamic_universe,
            subscriptions,
            process_events,
            ProcessEvent {
                date: execution_date,
                kind: ProcessEventKind::AccountManagementFee,
                order_id: None,
                symbol: None,
                side: None,
                detail: format!(
                    "rate={rate:.6} fee={fee:.2} cash_before={cash_before:.2} cash_after={:.2} management_fees={:.2}",
                    portfolio.cash(),
                    portfolio.management_fees()
                ),
            },
        )?;
        Ok(report)
    }

    fn settle_delisted_positions(
        &self,
        date: NaiveDate,
        portfolio: &mut PortfolioState,
        notes: &mut Vec<String>,
    ) -> Result<BrokerExecutionReport, BacktestError> {
        let mut report = BrokerExecutionReport::default();
        let symbols = portfolio.positions().keys().cloned().collect::<Vec<_>>();
        for symbol in symbols {
            let Some(position) = portfolio.position(&symbol) else {
                continue;
            };
            if position.quantity == 0 {
                continue;
            }
            let Some(instrument) = self.data.instrument(&symbol) else {
                continue;
            };
            let should_settle = instrument.is_delisted_before(date)
                || (instrument.status.eq_ignore_ascii_case("delisted")
                    && instrument.delisted_at.is_none()
                    && self.data.market(date, &symbol).is_none());
            if !should_settle {
                continue;
            }

            let quantity = position.quantity;
            let settlement_price = if position.last_price.is_finite() && position.last_price > 0.0 {
                position.last_price
            } else if position.average_cost.is_finite() && position.average_cost > 0.0 {
                position.average_cost
            } else {
                0.0
            };
            let effective_delisted_at = instrument
                .delisted_at
                .or_else(|| self.data.calendar().previous_day(date))
                .unwrap_or(date);
            if !settlement_price.is_finite() || settlement_price <= 0.0 {
                return Err(BacktestError::Execution(format!(
                    "missing delisting settlement price for {} on {}",
                    symbol, date
                )));
            }

            let cash_before = portfolio.cash();
            let gross_amount = settlement_price * quantity as f64;
            let realized_pnl_delta = {
                let position = portfolio
                    .position_mut_if_exists(&symbol)
                    .expect("position exists for delisting settlement");
                position
                    .sell(quantity, settlement_price)
                    .map_err(BacktestError::Execution)?
            };
            portfolio.apply_cash_delta(gross_amount);
            portfolio.prune_flat_positions();

            let reason = format!(
                "delisted_cash_settlement effective_date={} status={}",
                effective_delisted_at, instrument.status
            );
            notes.push(reason.clone());
            report.order_events.push(OrderEvent {
                date,
                order_id: None,
                symbol: symbol.clone(),
                side: OrderSide::Sell,
                requested_quantity: quantity,
                filled_quantity: quantity,
                status: OrderStatus::Filled,
                reason: reason.clone(),
            });
            report.fill_events.push(FillEvent {
                date,
                order_id: None,
                symbol: symbol.clone(),
                side: OrderSide::Sell,
                quantity,
                price: settlement_price,
                gross_amount,
                commission: 0.0,
                stamp_tax: 0.0,
                net_cash_flow: gross_amount,
                reason: reason.clone(),
            });
            report.position_events.push(PositionEvent {
                date,
                symbol: symbol.clone(),
                delta_quantity: -(quantity as i32),
                quantity_after: 0,
                average_cost: 0.0,
                realized_pnl_delta,
                reason: reason.clone(),
            });
            report.account_events.push(AccountEvent {
                date,
                cash_before,
                cash_after: portfolio.cash(),
                total_equity: portfolio.total_equity(),
                note: reason,
            });
        }
        Ok(report)
    }
}

fn has_execution_quote_in_window(
    data: &DataSet,
    date: NaiveDate,
    symbol: &str,
    start_time: Option<chrono::NaiveTime>,
    end_time: Option<chrono::NaiveTime>,
) -> bool {
    let start_cursor = start_time.map(|time| date.and_time(time));
    let end_cursor = end_time.map(|time| date.and_time(time));
    if let Some(cursor) = start_cursor
        && end_cursor.is_none()
    {
        return data
            .execution_quotes_on(date, symbol)
            .iter()
            .any(|quote| quote.timestamp <= cursor);
    }
    data.execution_quotes_on(date, symbol).iter().any(|quote| {
        !start_cursor.is_some_and(|cursor| quote.timestamp < cursor)
            && !end_cursor.is_some_and(|cursor| quote.timestamp > cursor)
    })
}

fn has_execution_quote_near_start_time(
    data: &DataSet,
    date: NaiveDate,
    symbol: &str,
    start_time: NaiveTime,
) -> bool {
    let cursor = date.and_time(start_time);
    let Some(latest) = data
        .execution_quotes_on(date, symbol)
        .iter()
        .filter(|quote| quote.timestamp <= cursor)
        .max_by_key(|quote| quote.timestamp)
    else {
        return false;
    };
    cursor.signed_duration_since(latest.timestamp) <= Duration::seconds(90)
}

fn decision_has_algo_execution(decision: &StrategyDecision) -> bool {
    decision.order_intents.iter().any(|intent| {
        matches!(
            intent,
            OrderIntent::AlgoValue { .. }
                | OrderIntent::AlgoPercent { .. }
                | OrderIntent::TimedTargetValue { .. }
                | OrderIntent::TargetPortfolioSmart {
                    order_prices: Some(TargetPortfolioOrderPricing::AlgoOrder { .. }),
                    ..
                }
        )
    })
}

fn execution_quote_symbols_for_decision(
    decision: &StrategyDecision,
    portfolio: &PortfolioState,
    open_orders: &[OpenOrderView],
) -> BTreeSet<String> {
    let mut symbols = BTreeSet::new();
    symbols.extend(open_orders.iter().map(|order| order.symbol.clone()));
    if decision.rebalance {
        symbols.extend(portfolio.positions().keys().cloned());
        symbols.extend(decision.target_weights.keys().cloned());
    }
    if !decision.exit_symbols.is_empty() {
        symbols.extend(decision.exit_symbols.iter().cloned());
    }

    for intent in &decision.order_intents {
        match intent {
            OrderIntent::Shares { symbol, .. }
            | OrderIntent::LimitShares { symbol, .. }
            | OrderIntent::Lots { symbol, .. }
            | OrderIntent::LimitLots { symbol, .. }
            | OrderIntent::TargetShares { symbol, .. }
            | OrderIntent::LimitTargetShares { symbol, .. }
            | OrderIntent::TargetValue { symbol, .. }
            | OrderIntent::TimedTargetValue { symbol, .. }
            | OrderIntent::LimitTargetValue { symbol, .. }
            | OrderIntent::Value { symbol, .. }
            | OrderIntent::LimitValue { symbol, .. }
            | OrderIntent::Percent { symbol, .. }
            | OrderIntent::LimitPercent { symbol, .. }
            | OrderIntent::TargetPercent { symbol, .. }
            | OrderIntent::LimitTargetPercent { symbol, .. }
            | OrderIntent::AlgoValue { symbol, .. }
            | OrderIntent::AlgoPercent { symbol, .. }
            | OrderIntent::CancelSymbol { symbol, .. } => {
                symbols.insert(symbol.clone());
            }
            OrderIntent::TargetPortfolioSmart { target_weights, .. } => {
                symbols.extend(portfolio.positions().keys().cloned());
                symbols.extend(target_weights.keys().cloned());
            }
            OrderIntent::CancelAll { .. } => {
                symbols.extend(open_orders.iter().map(|order| order.symbol.clone()));
            }
            OrderIntent::UpdateUniverse { .. }
            | OrderIntent::Subscribe { .. }
            | OrderIntent::Unsubscribe { .. }
            | OrderIntent::DepositWithdraw { .. }
            | OrderIntent::FinanceRepay { .. }
            | OrderIntent::SetManagementFeeRate { .. }
            | OrderIntent::CancelOrder { .. }
            | OrderIntent::Futures { .. } => {}
        }
    }

    symbols.retain(|symbol| !symbol.trim().is_empty());
    symbols
}

fn algo_execution_quote_windows_for_decision(
    decision: &StrategyDecision,
    portfolio: &PortfolioState,
) -> BTreeMap<(Option<NaiveTime>, Option<NaiveTime>), BTreeSet<String>> {
    let mut groups = BTreeMap::<(Option<NaiveTime>, Option<NaiveTime>), BTreeSet<String>>::new();
    for intent in &decision.order_intents {
        match intent {
            OrderIntent::AlgoValue {
                symbol,
                start_time,
                end_time,
                ..
            }
            | OrderIntent::AlgoPercent {
                symbol,
                start_time,
                end_time,
                ..
            }
            | OrderIntent::TimedTargetValue {
                symbol,
                start_time,
                end_time,
                ..
            } => {
                if start_time.is_some() || end_time.is_some() {
                    groups
                        .entry((*start_time, *end_time))
                        .or_default()
                        .insert(symbol.clone());
                }
            }
            OrderIntent::TargetPortfolioSmart {
                target_weights,
                order_prices:
                    Some(TargetPortfolioOrderPricing::AlgoOrder {
                        start_time,
                        end_time,
                        ..
                    }),
                ..
            } => {
                if start_time.is_some() || end_time.is_some() {
                    let symbols = groups.entry((*start_time, *end_time)).or_default();
                    symbols.extend(portfolio.positions().keys().cloned());
                    symbols.extend(target_weights.keys().cloned());
                }
            }
            _ => {}
        }
    }
    groups
}

fn collect_scheduled_decisions<S: Strategy>(
    strategy: &mut S,
    scheduler: &Scheduler<'_>,
    execution_date: NaiveDate,
    stage: ScheduleStage,
    rules: &[ScheduleRule],
    decision_date: NaiveDate,
    decision_index: usize,
    data: &crate::data::DataSet,
    portfolio: &PortfolioState,
    futures_account: Option<&FuturesAccountState>,
    open_orders: &[crate::strategy::OpenOrderView],
    dynamic_universe: Option<&BTreeSet<String>>,
    subscriptions: &BTreeSet<String>,
    process_events: &mut Vec<ProcessEvent>,
    process_event_bus: &mut ProcessEventBus,
    current_time: Option<chrono::NaiveTime>,
    order_events: &[OrderEvent],
    fills: &[FillEvent],
) -> Result<crate::strategy::StrategyDecision, BacktestError> {
    let mut combined = crate::strategy::StrategyDecision::default();
    // In lagged modes such as next_bar_open, scheduled callbacks generate
    // signals on the decision date while the broker later matches them on the
    // execution date.  Triggering schedules with the execution date would let a
    // T+1 calendar state suppress or create T-day signals.
    for rule in scheduler.triggered_rules_at(decision_date, stage, current_time, rules) {
        publish_phase_event(
            strategy,
            process_event_bus,
            execution_date,
            decision_date,
            decision_index,
            data,
            portfolio,
            futures_account,
            open_orders,
            dynamic_universe,
            subscriptions,
            process_events,
            execution_date,
            ProcessEventKind::PreScheduled,
            format!("scheduled:{}:{}:pre", rule.name, stage_label(stage)),
        )?;
        combined.merge_from(strategy.on_scheduled(
            &StrategyContext {
                execution_date,
                decision_date,
                decision_index,
                data,
                portfolio,
                futures_account,
                open_orders,
                dynamic_universe,
                subscriptions,
                process_events: process_events.as_slice(),
                active_process_event: None,
                active_datetime: stage_datetime(decision_date, current_time),
                order_events,
                fills,
            },
            rule,
        )?);
        publish_phase_event(
            strategy,
            process_event_bus,
            execution_date,
            decision_date,
            decision_index,
            data,
            portfolio,
            futures_account,
            open_orders,
            dynamic_universe,
            subscriptions,
            process_events,
            execution_date,
            ProcessEventKind::PostScheduled,
            format!("scheduled:{}:{}:post", rule.name, stage_label(stage)),
        )?;
    }
    Ok(combined)
}

fn publish_phase_event<S: Strategy>(
    strategy: &mut S,
    process_event_bus: &mut ProcessEventBus,
    execution_date: NaiveDate,
    decision_date: NaiveDate,
    decision_index: usize,
    data: &crate::data::DataSet,
    portfolio: &PortfolioState,
    futures_account: Option<&FuturesAccountState>,
    open_orders: &[crate::strategy::OpenOrderView],
    dynamic_universe: Option<&BTreeSet<String>>,
    subscriptions: &BTreeSet<String>,
    events: &mut Vec<ProcessEvent>,
    date: NaiveDate,
    kind: ProcessEventKind,
    detail: impl Into<String>,
) -> Result<(), BacktestError> {
    let event = ProcessEvent {
        date,
        kind,
        order_id: None,
        symbol: None,
        side: None,
        detail: detail.into(),
    };
    process_event_bus.publish(&event);
    let process_events = events.as_slice();
    let event_ctx = StrategyContext {
        execution_date,
        decision_date,
        decision_index,
        data,
        portfolio,
        futures_account,
        open_orders,
        dynamic_universe,
        subscriptions,
        process_events,
        active_process_event: Some(&event),
        active_datetime: None,
        order_events: &[],
        fills: &[],
    };
    strategy.on_process_event(&event_ctx, &event)?;
    events.push(event);
    Ok(())
}

fn publish_process_events<S: Strategy>(
    strategy: &mut S,
    process_event_bus: &mut ProcessEventBus,
    execution_date: NaiveDate,
    decision_date: NaiveDate,
    decision_index: usize,
    data: &crate::data::DataSet,
    portfolio: &PortfolioState,
    futures_account: Option<&FuturesAccountState>,
    open_orders: &[crate::strategy::OpenOrderView],
    dynamic_universe: Option<&BTreeSet<String>>,
    subscriptions: &BTreeSet<String>,
    target: &mut Vec<ProcessEvent>,
    incoming: &mut Vec<ProcessEvent>,
) -> Result<(), BacktestError> {
    for event in incoming.drain(..) {
        process_event_bus.publish(&event);
        let process_events = target.as_slice();
        let event_ctx = StrategyContext {
            execution_date,
            decision_date,
            decision_index,
            data,
            portfolio,
            futures_account,
            open_orders,
            dynamic_universe,
            subscriptions,
            process_events,
            active_process_event: Some(&event),
            active_datetime: None,
            order_events: &[],
            fills: &[],
        };
        strategy.on_process_event(&event_ctx, &event)?;
        target.push(event);
    }
    Ok(())
}

fn publish_custom_process_event<S: Strategy>(
    strategy: &mut S,
    process_event_bus: &mut ProcessEventBus,
    execution_date: NaiveDate,
    decision_date: NaiveDate,
    decision_index: usize,
    data: &crate::data::DataSet,
    portfolio: &PortfolioState,
    futures_account: Option<&FuturesAccountState>,
    open_orders: &[crate::strategy::OpenOrderView],
    dynamic_universe: Option<&BTreeSet<String>>,
    subscriptions: &BTreeSet<String>,
    target: &mut Vec<ProcessEvent>,
    event: ProcessEvent,
) -> Result<(), BacktestError> {
    process_event_bus.publish(&event);
    let process_events = target.as_slice();
    let event_ctx = StrategyContext {
        execution_date,
        decision_date,
        decision_index,
        data,
        portfolio,
        futures_account,
        open_orders,
        dynamic_universe,
        subscriptions,
        process_events,
        active_process_event: Some(&event),
        active_datetime: None,
        order_events: &[],
        fills: &[],
    };
    strategy.on_process_event(&event_ctx, &event)?;
    target.push(event);
    Ok(())
}

fn stage_label(stage: ScheduleStage) -> &'static str {
    match stage {
        ScheduleStage::BeforeTrading => "before_trading",
        ScheduleStage::OpenAuction => "open_auction",
        ScheduleStage::Bar => "bar",
        ScheduleStage::Minute => "minute",
        ScheduleStage::OnDay => "on_day",
        ScheduleStage::AfterTrading => "after_trading",
        ScheduleStage::Settlement => "settlement",
    }
}

fn stage_datetime(
    date: NaiveDate,
    time: Option<chrono::NaiveTime>,
) -> Option<chrono::NaiveDateTime> {
    time.map(|value| date.and_time(value))
}

fn should_run_minute_events(rules: &[ScheduleRule], subscriptions: &BTreeSet<String>) -> bool {
    !subscriptions.is_empty() || rules.iter().any(|rule| rule.stage == ScheduleStage::Minute)
}

fn merge_broker_report(target: &mut BrokerExecutionReport, incoming: BrokerExecutionReport) {
    target.order_events.extend(incoming.order_events);
    target.fill_events.extend(incoming.fill_events);
    target.position_events.extend(incoming.position_events);
    target.account_events.extend(incoming.account_events);
    target.process_events.extend(incoming.process_events);
    target.diagnostics.extend(incoming.diagnostics);
}

fn merge_futures_report(target: &mut BrokerExecutionReport, incoming: FuturesExecutionReport) {
    target.order_events.extend(incoming.order_events);
    target.fill_events.extend(incoming.fill_events);
    target.position_events.extend(incoming.position_events);
    target.account_events.extend(incoming.account_events);
    target.process_events.extend(incoming.process_events);
    target.diagnostics.extend(incoming.diagnostics);
}

fn merge_futures_execution_report(
    target: &mut FuturesExecutionReport,
    incoming: FuturesExecutionReport,
) {
    target.order_events.extend(incoming.order_events);
    target.fill_events.extend(incoming.fill_events);
    target.position_events.extend(incoming.position_events);
    target.account_events.extend(incoming.account_events);
    target.process_events.extend(incoming.process_events);
    target.diagnostics.extend(incoming.diagnostics);
}

fn analyzer_ratio_change(start: f64, end: f64) -> f64 {
    if start.abs() <= f64::EPSILON {
        0.0
    } else {
        end / start - 1.0
    }
}

fn price_is_tick_aligned(price: f64, tick: f64) -> bool {
    if !price.is_finite() || !tick.is_finite() || tick <= 0.0 {
        return false;
    }
    let ratio = price / tick;
    (ratio - ratio.round()).abs() <= 1e-6
}

fn futures_trading_phase_allows_orders(phase: Option<&str>) -> bool {
    let Some(phase) = phase.map(str::trim).filter(|value| !value.is_empty()) else {
        return true;
    };
    matches!(
        phase.to_ascii_lowercase().as_str(),
        "continuous"
            | "trading"
            | "trade"
            | "open_auction"
            | "auction"
            | "call_auction"
            | "opening_auction"
    )
}

fn futures_limit_satisfied(side: OrderSide, price: f64, limit_price: Option<f64>) -> bool {
    let Some(limit_price) = limit_price else {
        return price.is_finite() && price > 0.0;
    };
    if !price.is_finite() || price <= 0.0 || !limit_price.is_finite() || limit_price <= 0.0 {
        return false;
    }
    match side {
        OrderSide::Buy => price <= limit_price + 1e-9,
        OrderSide::Sell => price + 1e-9 >= limit_price,
    }
}

fn futures_cancel_report(
    date: NaiveDate,
    order: FuturesOpenOrder,
    reason: &str,
) -> FuturesExecutionReport {
    let mut report = FuturesExecutionReport::default();
    let side = order.intent.side();
    report.process_events.push(ProcessEvent {
        date,
        kind: ProcessEventKind::OrderPendingCancel,
        order_id: Some(order.order_id),
        symbol: Some(order.intent.symbol.clone()),
        side: Some(side),
        detail: format!("reason={reason}"),
    });
    report.order_events.push(OrderEvent {
        date,
        order_id: Some(order.order_id),
        symbol: order.intent.symbol.clone(),
        side,
        requested_quantity: order.requested_quantity,
        filled_quantity: order.filled_quantity,
        status: OrderStatus::Canceled,
        reason: format!("{reason}: futures order canceled by user"),
    });
    report.process_events.push(ProcessEvent {
        date,
        kind: ProcessEventKind::OrderCancellationPass,
        order_id: Some(order.order_id),
        symbol: Some(order.intent.symbol),
        side: Some(side),
        detail: format!(
            "requested_quantity={} filled_quantity={} remaining_quantity={}",
            order.requested_quantity, order.filled_quantity, order.remaining_quantity
        ),
    });
    report
}

mod date_format {
    use chrono::NaiveDate;
    use serde::Serializer;

    const FORMAT: &str = "%Y-%m-%d";

    pub fn serialize<S>(date: &NaiveDate, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_str(&date.format(FORMAT).to_string())
    }
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;

    use chrono::NaiveDate;

    use super::{BacktestConfig, BacktestEngine};
    use crate::broker::{BrokerSimulator, MatchingType};
    use crate::cost::ChinaAShareCostModel;
    use crate::data::{
        BenchmarkSnapshot, CandidateEligibility, DailyFactorSnapshot, DailyMarketSnapshot, DataSet,
        PriceField,
    };
    use crate::events::{OrderSide, OrderStatus};
    use crate::instrument::Instrument;
    use crate::risk_control::FidcRiskControlConfig;
    use crate::rules::ChinaEquityRuleHooks;
    use crate::scheduler::{ScheduleRule, ScheduleStage};
    use crate::strategy::{OrderIntent, Strategy, StrategyContext, StrategyDecision};

    const SYMBOL: &str = "000001.SZ";

    #[derive(Debug)]
    struct BuyWhenDecisionDateStrategy {
        decision_date: NaiveDate,
    }

    impl Strategy for BuyWhenDecisionDateStrategy {
        fn name(&self) -> &str {
            "buy_when_decision_date"
        }

        fn on_day(
            &mut self,
            ctx: &StrategyContext<'_>,
        ) -> Result<StrategyDecision, super::BacktestError> {
            if ctx.decision_date == self.decision_date && ctx.portfolio.position(SYMBOL).is_none() {
                return Ok(StrategyDecision {
                    order_intents: vec![OrderIntent::Shares {
                        symbol: SYMBOL.to_string(),
                        quantity: 100,
                        reason: "test_buy".to_string(),
                    }],
                    ..StrategyDecision::default()
                });
            }
            Ok(StrategyDecision::default())
        }
    }

    #[derive(Debug)]
    struct ScheduledBuyStrategy {
        rule: ScheduleRule,
        expected_decision_date: NaiveDate,
    }

    impl Strategy for ScheduledBuyStrategy {
        fn name(&self) -> &str {
            "scheduled_buy"
        }

        fn schedule_rules(&self) -> Vec<ScheduleRule> {
            vec![self.rule.clone()]
        }

        fn on_scheduled(
            &mut self,
            ctx: &StrategyContext<'_>,
            rule: &ScheduleRule,
        ) -> Result<StrategyDecision, super::BacktestError> {
            assert_eq!(rule.name, self.rule.name);
            assert_eq!(ctx.decision_date, self.expected_decision_date);
            assert_eq!(
                ctx.current_datetime().map(|value| value.date()),
                Some(self.expected_decision_date)
            );
            if ctx.portfolio.position(SYMBOL).is_none() {
                return Ok(StrategyDecision {
                    order_intents: vec![OrderIntent::Shares {
                        symbol: SYMBOL.to_string(),
                        quantity: 100,
                        reason: "scheduled_test_buy".to_string(),
                    }],
                    ..StrategyDecision::default()
                });
            }
            Ok(StrategyDecision::default())
        }
    }

    #[derive(Debug)]
    struct ScheduledRoundTripStrategy {
        rule: ScheduleRule,
        buy_decision_date: NaiveDate,
        sell_decision_date: NaiveDate,
    }

    impl Strategy for ScheduledRoundTripStrategy {
        fn name(&self) -> &str {
            "scheduled_round_trip"
        }

        fn schedule_rules(&self) -> Vec<ScheduleRule> {
            vec![self.rule.clone()]
        }

        fn on_scheduled(
            &mut self,
            ctx: &StrategyContext<'_>,
            rule: &ScheduleRule,
        ) -> Result<StrategyDecision, super::BacktestError> {
            assert_eq!(rule.name, self.rule.name);
            assert_eq!(
                ctx.current_datetime().map(|value| value.date()),
                Some(ctx.decision_date)
            );
            if ctx.decision_date == self.buy_decision_date
                && ctx.portfolio.position(SYMBOL).is_none()
            {
                return Ok(StrategyDecision {
                    order_intents: vec![OrderIntent::Shares {
                        symbol: SYMBOL.to_string(),
                        quantity: 100,
                        reason: "round_trip_buy".to_string(),
                    }],
                    ..StrategyDecision::default()
                });
            }
            if ctx.decision_date == self.sell_decision_date {
                if let Some(position) = ctx.portfolio.position(SYMBOL) {
                    return Ok(StrategyDecision {
                        order_intents: vec![OrderIntent::Shares {
                            symbol: SYMBOL.to_string(),
                            quantity: -(position.quantity as i32),
                            reason: "round_trip_sell".to_string(),
                        }],
                        ..StrategyDecision::default()
                    });
                }
            }
            Ok(StrategyDecision::default())
        }
    }

    #[derive(Debug)]
    struct ScheduledSameDayRebuyStrategy {
        rule: ScheduleRule,
        buy_decision_date: NaiveDate,
        rebuy_decision_date: NaiveDate,
    }

    impl Strategy for ScheduledSameDayRebuyStrategy {
        fn name(&self) -> &str {
            "scheduled_same_day_rebuy"
        }

        fn schedule_rules(&self) -> Vec<ScheduleRule> {
            vec![self.rule.clone()]
        }

        fn on_scheduled(
            &mut self,
            ctx: &StrategyContext<'_>,
            rule: &ScheduleRule,
        ) -> Result<StrategyDecision, super::BacktestError> {
            assert_eq!(rule.name, self.rule.name);
            if ctx.decision_date == self.buy_decision_date
                && ctx.portfolio.position(SYMBOL).is_none()
            {
                return Ok(StrategyDecision {
                    order_intents: vec![OrderIntent::Shares {
                        symbol: SYMBOL.to_string(),
                        quantity: 100,
                        reason: "same_day_rebuy_setup_buy".to_string(),
                    }],
                    ..StrategyDecision::default()
                });
            }
            if ctx.decision_date == self.rebuy_decision_date {
                if let Some(position) = ctx.portfolio.position(SYMBOL) {
                    return Ok(StrategyDecision {
                        order_intents: vec![
                            OrderIntent::Shares {
                                symbol: SYMBOL.to_string(),
                                quantity: -(position.quantity as i32),
                                reason: "same_day_rebuy_sell".to_string(),
                            },
                            OrderIntent::Shares {
                                symbol: SYMBOL.to_string(),
                                quantity: 100,
                                reason: "same_day_rebuy_buy".to_string(),
                            },
                        ],
                        ..StrategyDecision::default()
                    });
                }
            }
            Ok(StrategyDecision::default())
        }
    }

    fn d(year: i32, month: u32, day: u32) -> NaiveDate {
        NaiveDate::from_ymd_opt(year, month, day).expect("valid date")
    }

    fn market(date: NaiveDate, open: f64, close: f64) -> DailyMarketSnapshot {
        DailyMarketSnapshot {
            date,
            symbol: SYMBOL.to_string(),
            timestamp: Some(format!("{date} 15:00:00")),
            day_open: open,
            open,
            high: close.max(open) + 1.0,
            low: close.min(open) - 1.0,
            close,
            last_price: close,
            bid1: close - 0.01,
            ask1: close + 0.01,
            prev_close: 10.0,
            volume: 1_000_000,
            minute_volume: 10_000,
            bid1_volume: 10_000,
            ask1_volume: 10_000,
            trading_phase: Some("continuous".to_string()),
            paused: false,
            upper_limit: 200.0,
            lower_limit: 1.0,
            price_tick: 0.01,
        }
    }

    fn market_with_state(
        date: NaiveDate,
        open: f64,
        close: f64,
        paused: bool,
        upper_limit: f64,
        lower_limit: f64,
    ) -> DailyMarketSnapshot {
        DailyMarketSnapshot {
            paused,
            upper_limit,
            lower_limit,
            ..market(date, open, close)
        }
    }

    fn market_with_volume(
        date: NaiveDate,
        open: f64,
        close: f64,
        volume: u64,
    ) -> DailyMarketSnapshot {
        DailyMarketSnapshot {
            volume,
            ..market(date, open, close)
        }
    }

    fn factor(date: NaiveDate) -> DailyFactorSnapshot {
        DailyFactorSnapshot {
            date,
            symbol: SYMBOL.to_string(),
            market_cap_bn: 10.0,
            free_float_cap_bn: 8.0,
            pe_ttm: 12.0,
            turnover_ratio: Some(1.0),
            effective_turnover_ratio: Some(1.0),
            extra_factors: BTreeMap::new(),
        }
    }

    fn candidate(date: NaiveDate) -> CandidateEligibility {
        CandidateEligibility {
            date,
            symbol: SYMBOL.to_string(),
            is_st: false,
            is_star_st: false,
            is_new_listing: false,
            is_paused: false,
            allow_buy: true,
            allow_sell: true,
            is_kcb: false,
            is_one_yuan: false,
            risk_level_code: None,
        }
    }

    fn candidate_with_state(
        date: NaiveDate,
        is_paused: bool,
        allow_buy: bool,
    ) -> CandidateEligibility {
        CandidateEligibility {
            is_paused,
            allow_buy,
            ..candidate(date)
        }
    }

    fn candidate_with_sell_state(
        date: NaiveDate,
        is_paused: bool,
        allow_sell: bool,
    ) -> CandidateEligibility {
        CandidateEligibility {
            is_paused,
            allow_sell,
            ..candidate(date)
        }
    }

    fn st_candidate(date: NaiveDate) -> CandidateEligibility {
        CandidateEligibility {
            is_st: true,
            ..candidate(date)
        }
    }

    fn star_st_candidate(date: NaiveDate) -> CandidateEligibility {
        CandidateEligibility {
            is_star_st: true,
            ..candidate(date)
        }
    }

    fn one_yuan_candidate(date: NaiveDate) -> CandidateEligibility {
        CandidateEligibility {
            is_one_yuan: true,
            ..candidate(date)
        }
    }

    fn benchmark(date: NaiveDate) -> BenchmarkSnapshot {
        BenchmarkSnapshot {
            date,
            benchmark: "000852.SH".to_string(),
            open: 1000.0,
            close: 1000.0,
            prev_close: 1000.0,
            volume: 1_000_000,
        }
    }

    fn dataset() -> DataSet {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        dataset_with(
            market(first, 10.0, 11.5),
            market(second, 12.0, 99.0),
            candidate(first),
            candidate(second),
        )
    }

    fn default_instrument() -> Instrument {
        Instrument {
            symbol: SYMBOL.to_string(),
            name: "Test Stock".to_string(),
            board: "SZ".to_string(),
            round_lot: 100,
            listed_at: Some(d(2020, 1, 1)),
            delisted_at: None,
            status: "active".to_string(),
        }
    }

    fn dataset_with(
        first_market: DailyMarketSnapshot,
        second_market: DailyMarketSnapshot,
        first_candidate: CandidateEligibility,
        second_candidate: CandidateEligibility,
    ) -> DataSet {
        dataset_with_instrument(
            default_instrument(),
            first_market,
            second_market,
            first_candidate,
            second_candidate,
        )
    }

    fn dataset_with_instrument(
        instrument: Instrument,
        first_market: DailyMarketSnapshot,
        second_market: DailyMarketSnapshot,
        first_candidate: CandidateEligibility,
        second_candidate: CandidateEligibility,
    ) -> DataSet {
        let first = first_market.date;
        let second = second_market.date;
        DataSet::from_components(
            vec![instrument],
            vec![first_market, second_market],
            vec![factor(first), factor(second)],
            vec![first_candidate, second_candidate],
            vec![benchmark(first), benchmark(second)],
        )
        .expect("dataset")
    }

    fn dataset_from_market_and_candidates(
        markets: Vec<DailyMarketSnapshot>,
        candidates: Vec<CandidateEligibility>,
    ) -> DataSet {
        let factors = markets
            .iter()
            .map(|market| factor(market.date))
            .collect::<Vec<_>>();
        let benchmarks = markets
            .iter()
            .map(|market| benchmark(market.date))
            .collect::<Vec<_>>();
        DataSet::from_components(
            vec![default_instrument()],
            markets,
            factors,
            candidates,
            benchmarks,
        )
        .expect("dataset")
    }

    fn run_with_matching(
        matching_type: MatchingType,
        execution_price_field: PriceField,
        decision_lag_trading_days: usize,
    ) -> super::BacktestResult {
        let first = d(2025, 1, 2);
        let broker = BrokerSimulator::new_with_execution_price(
            ChinaAShareCostModel::default(),
            ChinaEquityRuleHooks,
            execution_price_field,
        )
        .with_matching_type(matching_type)
        .with_volume_limit(false)
        .with_liquidity_limit(false)
        .with_inactive_limit(false);
        let config = BacktestConfig {
            initial_cash: 100_000.0,
            benchmark_code: "000852.SH".to_string(),
            start_date: Some(first),
            end_date: Some(d(2025, 1, 3)),
            decision_lag_trading_days,
            execution_price_field,
        };

        BacktestEngine::new(
            dataset(),
            BuyWhenDecisionDateStrategy {
                decision_date: first,
            },
            broker,
            config,
        )
        .run()
        .expect("backtest run")
    }

    fn run_scheduled_next_open_with_dataset(dataset: DataSet) -> super::BacktestResult {
        run_scheduled_next_open_with_dataset_and_broker(
            dataset,
            scheduled_next_open_broker(FidcRiskControlConfig::default()),
        )
    }

    fn scheduled_next_open_broker(
        risk_config: FidcRiskControlConfig,
    ) -> BrokerSimulator<ChinaAShareCostModel, ChinaEquityRuleHooks> {
        BrokerSimulator::new_with_execution_price(
            ChinaAShareCostModel::default(),
            ChinaEquityRuleHooks,
            PriceField::Open,
        )
        .with_matching_type(MatchingType::NextBarOpen)
        .with_risk_config(risk_config)
        .with_volume_limit(false)
        .with_liquidity_limit(false)
        .with_inactive_limit(false)
    }

    fn run_scheduled_next_open_with_dataset_and_broker(
        dataset: DataSet,
        broker: BrokerSimulator<ChinaAShareCostModel, ChinaEquityRuleHooks>,
    ) -> super::BacktestResult {
        let first = d(2025, 1, 2);
        let config = BacktestConfig {
            initial_cash: 100_000.0,
            benchmark_code: "000852.SH".to_string(),
            start_date: Some(first),
            end_date: Some(d(2025, 1, 3)),
            decision_lag_trading_days: 1,
            execution_price_field: PriceField::Open,
        };

        BacktestEngine::new(
            dataset,
            ScheduledBuyStrategy {
                rule: ScheduleRule::weekly_by_weekday("weekly_signal", 4, ScheduleStage::OnDay),
                expected_decision_date: first,
            },
            broker,
            config,
        )
        .run()
        .expect("backtest run")
    }

    fn run_scheduled_round_trip_next_open_with_dataset_and_broker(
        dataset: DataSet,
        broker: BrokerSimulator<ChinaAShareCostModel, ChinaEquityRuleHooks>,
    ) -> super::BacktestResult {
        let first = d(2025, 1, 2);
        let third = d(2025, 1, 6);
        let config = BacktestConfig {
            initial_cash: 100_000.0,
            benchmark_code: "000852.SH".to_string(),
            start_date: Some(first),
            end_date: Some(d(2025, 1, 7)),
            decision_lag_trading_days: 1,
            execution_price_field: PriceField::Open,
        };

        BacktestEngine::new(
            dataset,
            ScheduledRoundTripStrategy {
                rule: ScheduleRule::daily("daily_round_trip", ScheduleStage::OnDay),
                buy_decision_date: first,
                sell_decision_date: third,
            },
            broker,
            config,
        )
        .run()
        .expect("backtest run")
    }

    fn run_scheduled_same_day_rebuy_next_open_with_dataset_and_broker(
        dataset: DataSet,
        broker: BrokerSimulator<ChinaAShareCostModel, ChinaEquityRuleHooks>,
    ) -> super::BacktestResult {
        let first = d(2025, 1, 2);
        let third = d(2025, 1, 6);
        let config = BacktestConfig {
            initial_cash: 100_000.0,
            benchmark_code: "000852.SH".to_string(),
            start_date: Some(first),
            end_date: Some(d(2025, 1, 7)),
            decision_lag_trading_days: 1,
            execution_price_field: PriceField::Open,
        };

        BacktestEngine::new(
            dataset,
            ScheduledSameDayRebuyStrategy {
                rule: ScheduleRule::daily("daily_same_day_rebuy", ScheduleStage::OnDay),
                buy_decision_date: first,
                rebuy_decision_date: third,
            },
            broker,
            config,
        )
        .run()
        .expect("backtest run")
    }

    fn assert_next_open_canceled_with_reason(result: &super::BacktestResult, reason: &str) {
        let execution_date = d(2025, 1, 3);
        assert!(result.fills.is_empty());
        assert!(result.order_events.iter().any(|event| {
            event.date == execution_date
                && matches!(event.status, OrderStatus::Canceled | OrderStatus::Rejected)
                && event.reason.contains(reason)
        }));
    }

    fn assert_round_trip_sell_canceled_with_reason(result: &super::BacktestResult, reason: &str) {
        let execution_date = d(2025, 1, 7);
        assert!(result.fills.iter().any(|fill| fill.side == OrderSide::Buy));
        assert!(result.fills.iter().all(|fill| fill.side != OrderSide::Sell));
        assert!(result.order_events.iter().any(|event| {
            event.date == execution_date
                && event.side == OrderSide::Sell
                && matches!(event.status, OrderStatus::Canceled | OrderStatus::Rejected)
                && event.reason.contains(reason)
        }));
    }

    #[test]
    fn current_bar_close_uses_decision_day_close_for_fill() {
        let result = run_with_matching(MatchingType::CurrentBarClose, PriceField::Close, 0);

        assert_eq!(result.fills.len(), 1);
        assert_eq!(result.fills[0].date, d(2025, 1, 2));
        assert_eq!(result.fills[0].price, 11.5);
    }

    #[test]
    fn next_bar_open_skips_unavailable_lag_day_and_fills_next_open() {
        let result = run_with_matching(MatchingType::NextBarOpen, PriceField::Open, 1);

        assert_eq!(result.fills.len(), 1);
        assert_eq!(result.fills[0].date, d(2025, 1, 3));
        assert_eq!(result.fills[0].price, 12.0);
        assert!(
            result.equity_curve[0]
                .diagnostics
                .contains("decision_lag_warmup"),
            "{}",
            result.equity_curve[0].diagnostics
        );
    }

    #[test]
    fn next_bar_open_scheduled_signal_uses_decision_date_not_execution_date() {
        let result = run_scheduled_next_open_with_dataset(dataset());

        assert_eq!(result.fills.len(), 1);
        assert_eq!(result.fills[0].date, d(2025, 1, 3));
        assert_eq!(result.fills[0].price, 12.0);
    }

    #[test]
    fn next_bar_open_execution_risk_ignores_decision_day_paused_state() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let result = run_scheduled_next_open_with_dataset(dataset_with(
            market_with_state(first, 10.0, 11.5, true, 10.0, 9.0),
            market_with_state(second, 12.0, 99.0, false, 200.0, 1.0),
            candidate_with_state(first, true, false),
            candidate(second),
        ));

        assert_eq!(result.fills.len(), 1);
        assert_eq!(result.fills[0].date, second);
        assert_eq!(result.fills[0].price, 12.0);
    }

    #[test]
    fn next_bar_open_execution_risk_ignores_decision_day_upper_limit_state() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let result = run_scheduled_next_open_with_dataset(dataset_with(
            market_with_state(first, 10.0, 10.0, false, 10.0, 1.0),
            market_with_state(second, 12.0, 99.0, false, 200.0, 1.0),
            candidate(first),
            candidate(second),
        ));

        assert_eq!(result.fills.len(), 1);
        assert_eq!(result.fills[0].date, second);
        assert_eq!(result.fills[0].price, 12.0);
    }

    #[test]
    fn next_bar_open_execution_risk_ignores_decision_day_st_state() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let result = run_scheduled_next_open_with_dataset(dataset_with(
            market(first, 10.0, 11.5),
            market(second, 12.0, 99.0),
            st_candidate(first),
            candidate(second),
        ));

        assert_eq!(result.fills.len(), 1);
        assert_eq!(result.fills[0].date, second);
        assert_eq!(result.fills[0].price, 12.0);
    }

    #[test]
    fn next_bar_open_execution_risk_ignores_decision_day_star_st_state() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let result = run_scheduled_next_open_with_dataset(dataset_with(
            market(first, 10.0, 11.5),
            market(second, 12.0, 99.0),
            star_st_candidate(first),
            candidate(second),
        ));

        assert_eq!(result.fills.len(), 1);
        assert_eq!(result.fills[0].date, second);
        assert_eq!(result.fills[0].price, 12.0);
    }

    #[test]
    fn next_bar_open_execution_risk_rejects_execution_day_paused_state() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let result = run_scheduled_next_open_with_dataset(dataset_with(
            market(first, 10.0, 11.5),
            market_with_state(second, 12.0, 99.0, true, 200.0, 1.0),
            candidate(first),
            candidate_with_state(second, true, true),
        ));

        assert_next_open_canceled_with_reason(&result, "paused");
    }

    #[test]
    fn next_bar_open_execution_risk_rejects_execution_day_upper_limit_buy() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let result = run_scheduled_next_open_with_dataset(dataset_with(
            market(first, 10.0, 11.5),
            market_with_state(second, 12.0, 99.0, false, 12.0, 1.0),
            candidate(first),
            candidate(second),
        ));

        assert_next_open_canceled_with_reason(&result, "open at or above upper limit");
    }

    #[test]
    fn next_bar_open_execution_risk_rejects_execution_day_st_state() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let result = run_scheduled_next_open_with_dataset(dataset_with(
            market(first, 10.0, 11.5),
            market(second, 12.0, 99.0),
            candidate(first),
            st_candidate(second),
        ));

        assert_next_open_canceled_with_reason(&result, "st");
    }

    #[test]
    fn next_bar_open_execution_risk_rejects_execution_day_star_st_state() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let result = run_scheduled_next_open_with_dataset(dataset_with(
            market(first, 10.0, 11.5),
            market(second, 12.0, 99.0),
            candidate(first),
            star_st_candidate(second),
        ));

        assert_next_open_canceled_with_reason(&result, "star_st");
    }

    #[test]
    fn next_bar_open_execution_risk_rejects_execution_day_one_yuan_state() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let result = run_scheduled_next_open_with_dataset(dataset_with(
            market(first, 10.0, 11.5),
            market(second, 12.0, 99.0),
            candidate(first),
            one_yuan_candidate(second),
        ));

        assert_next_open_canceled_with_reason(&result, "one_yuan");
    }

    #[test]
    fn next_bar_open_execution_risk_rejects_execution_day_delisted_state() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let mut instrument = default_instrument();
        instrument.delisted_at = Some(second);
        let result = run_scheduled_next_open_with_dataset(dataset_with_instrument(
            instrument,
            market(first, 10.0, 11.5),
            market(second, 12.0, 99.0),
            candidate(first),
            candidate(second),
        ));

        assert_next_open_canceled_with_reason(&result, "inactive_or_delisted");
    }

    #[test]
    fn next_bar_open_execution_risk_rejects_blacklisted_buy_on_execution_day() {
        let mut risk_config = FidcRiskControlConfig::default();
        risk_config
            .static_rules
            .blacklisted_symbols
            .insert(SYMBOL.to_string());
        let result = run_scheduled_next_open_with_dataset_and_broker(
            dataset(),
            scheduled_next_open_broker(risk_config),
        );

        assert_next_open_canceled_with_reason(&result, "blacklisted");
    }

    #[test]
    fn next_bar_open_sell_risk_ignores_decision_day_paused_and_lower_limit_state() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let third = d(2025, 1, 6);
        let fourth = d(2025, 1, 7);
        let result = run_scheduled_round_trip_next_open_with_dataset_and_broker(
            dataset_from_market_and_candidates(
                vec![
                    market(first, 10.0, 10.5),
                    market(second, 11.0, 11.5),
                    market_with_state(third, 9.0, 9.0, true, 20.0, 9.0),
                    market(fourth, 12.0, 12.2),
                ],
                vec![
                    candidate(first),
                    candidate(second),
                    candidate_with_sell_state(third, true, false),
                    candidate(fourth),
                ],
            ),
            scheduled_next_open_broker(FidcRiskControlConfig::default()),
        );

        let sell_fill = result
            .fills
            .iter()
            .find(|fill| fill.side == OrderSide::Sell)
            .expect("sell should execute on actual execution date");
        assert_eq!(sell_fill.date, fourth);
        assert_eq!(sell_fill.price, 12.0);
    }

    #[test]
    fn next_bar_open_sell_risk_rejects_execution_day_lower_limit_state() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let third = d(2025, 1, 6);
        let fourth = d(2025, 1, 7);
        let result = run_scheduled_round_trip_next_open_with_dataset_and_broker(
            dataset_from_market_and_candidates(
                vec![
                    market(first, 10.0, 10.5),
                    market(second, 11.0, 11.5),
                    market(third, 12.0, 12.5),
                    market_with_state(fourth, 9.0, 9.0, false, 20.0, 9.0),
                ],
                vec![
                    candidate(first),
                    candidate(second),
                    candidate(third),
                    candidate(fourth),
                ],
            ),
            scheduled_next_open_broker(FidcRiskControlConfig::default()),
        );

        assert_round_trip_sell_canceled_with_reason(&result, "open at or below lower limit");
    }

    #[test]
    fn next_bar_open_sell_volume_limit_ignores_decision_day_zero_volume() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let third = d(2025, 1, 6);
        let fourth = d(2025, 1, 7);
        let broker = scheduled_next_open_broker(FidcRiskControlConfig::default())
            .with_volume_limit(true)
            .with_volume_percent(0.25);
        let result = run_scheduled_round_trip_next_open_with_dataset_and_broker(
            dataset_from_market_and_candidates(
                vec![
                    market(first, 10.0, 10.5),
                    market(second, 11.0, 11.5),
                    market_with_volume(third, 12.0, 12.5, 0),
                    market(fourth, 12.0, 12.2),
                ],
                vec![
                    candidate(first),
                    candidate(second),
                    candidate(third),
                    candidate(fourth),
                ],
            ),
            broker,
        );

        assert!(result.fills.iter().any(|fill| fill.side == OrderSide::Buy));
        assert!(result.fills.iter().any(|fill| {
            fill.side == OrderSide::Sell && fill.date == fourth && (fill.price - 12.0).abs() < 1e-9
        }));
    }

    #[test]
    fn next_bar_open_sell_volume_limit_rejects_execution_day_zero_volume() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let third = d(2025, 1, 6);
        let fourth = d(2025, 1, 7);
        let broker = scheduled_next_open_broker(FidcRiskControlConfig::default())
            .with_volume_limit(true)
            .with_volume_percent(0.25);
        let result = run_scheduled_round_trip_next_open_with_dataset_and_broker(
            dataset_from_market_and_candidates(
                vec![
                    market(first, 10.0, 10.5),
                    market(second, 11.0, 11.5),
                    market(third, 12.0, 12.5),
                    market_with_volume(fourth, 12.0, 12.2, 0),
                ],
                vec![
                    candidate(first),
                    candidate(second),
                    candidate(third),
                    candidate(fourth),
                ],
            ),
            broker,
        );

        assert_round_trip_sell_canceled_with_reason(&result, "daily volume limit");
    }

    #[test]
    fn next_bar_open_same_day_rebuy_uses_actual_execution_date() {
        let first = d(2025, 1, 2);
        let second = d(2025, 1, 3);
        let third = d(2025, 1, 6);
        let fourth = d(2025, 1, 7);
        let result = run_scheduled_same_day_rebuy_next_open_with_dataset_and_broker(
            dataset_from_market_and_candidates(
                vec![
                    market(first, 10.0, 10.5),
                    market(second, 11.0, 11.5),
                    market(third, 12.0, 12.5),
                    market(fourth, 13.0, 13.5),
                ],
                vec![
                    candidate(first),
                    candidate(second),
                    candidate(third),
                    candidate(fourth),
                ],
            ),
            scheduled_next_open_broker(FidcRiskControlConfig::default()),
        );

        assert!(result.fills.iter().any(|fill| {
            fill.side == OrderSide::Buy
                && fill.date == second
                && fill.reason == "same_day_rebuy_setup_buy"
        }));
        assert!(result.fills.iter().any(|fill| {
            fill.side == OrderSide::Sell
                && fill.date == fourth
                && fill.reason == "same_day_rebuy_sell"
        }));
        assert!(result.order_events.iter().any(|event| {
            event.date == fourth
                && event.side == OrderSide::Buy
                && matches!(event.status, OrderStatus::Canceled | OrderStatus::Rejected)
                && event.reason.contains("same_day_rebuy_forbidden")
        }));
        assert!(!result.order_events.iter().any(|event| {
            event.date == third && event.reason.contains("same_day_rebuy_forbidden")
        }));
    }
}