//! Executes one frozen pool intent against real broker-simulator state.
use super::*;
use crate::holding_policy::HoldingLifecycleEvidence;
use crate::stock_pool_execution as pool;
use rust_decimal::{Decimal, prelude::ToPrimitive};
use chrono::Timelike;

#[derive(Debug)]
pub(super) struct DeferredStockPoolExecution {
    date: NaiveDate,
    contract: Box<pool::FrozenStockPoolIntent>,
    buy_only: bool,
    symbols: BTreeSet<String>,
    initial_holdings: BTreeSet<String>,
}

impl<C, R> BrokerSimulator<C, R> {
    pub(crate) fn pending_stock_pool_symbols(&self) -> BTreeSet<String> {
        self.deferred_stock_pools.borrow().values().flat_map(|pending| pending.symbols.iter().cloned()).collect()
    }

    pub(crate) fn has_pending_stock_pool_execution(&self) -> bool {
        !self.deferred_stock_pools.borrow().is_empty()
    }

    pub(crate) fn finish_stock_pool_session(&self, date: NaiveDate, report: &mut BrokerExecutionReport) {
        self.deferred_stock_pools.borrow_mut().retain(|_, pending| {
            if pending.date <= date {
                report.diagnostics.push(format!("stock_pool_unsubmitted_phase_expired generation={} date={date} no_buy_order_created=true",pending.contract.generation));
                false
            } else { true }
        });
    }
}

fn decimal(value: f64, label: &str) -> Result<Decimal, BacktestError> {
    if !value.is_finite() {
        return Err(BacktestError::Execution(format!(
            "stock_pool_nonfinite_{label}"
        )));
    }
    value
        .to_string()
        .parse()
        .map_err(|_| BacktestError::Execution(format!("stock_pool_decimal_range_{label}")))
}

fn etf_activity(report:&mut BrokerExecutionReport,date:NaiveDate,symbol:&str,side:pool::OrderSide,detail:String) {
    report.process_events.push(ProcessEvent {date,kind:ProcessEventKind::EtfExecutionFallback,order_id:None,
        symbol:Some(symbol.into()),side:Some(if side==pool::OrderSide::Buy {OrderSide::Buy} else {OrderSide::Sell}),detail});
}

fn pool_positions(
    portfolio: &PortfolioState,
    date: NaiveDate,
) -> Result<Vec<pool::Position>, BacktestError> {
    portfolio
        .positions()
        .values()
        .filter(|p| p.quantity > 0)
        .map(|p| {
            Ok(pool::Position {
                symbol: p.symbol.clone(),
                quantity: Decimal::from(p.quantity),
                closable_quantity: Decimal::from(p.sellable_qty(date)),
                average_cost: decimal(p.average_cost, "position_cost")?,
            })
        })
        .collect()
}

impl<C: CostModel, R: EquityRuleHooks> BrokerSimulator<C, R> {
    pub(super) fn resume_stock_pool_executions(&self, date: NaiveDate, portfolio: &mut PortfolioState, data: &DataSet,
        session: &mut BrokerExecutionSession, report: &mut BrokerExecutionReport) -> Result<(), BacktestError> {
        let clock = self.runtime_intraday_start_time.get().or(self.intraday_execution_start_time);
        let mut expired = Vec::new();
        for (id, pending) in self.deferred_stock_pools.borrow().iter() {
            let end = NaiveTime::parse_from_str(&pending.contract.rule.window_end, "%H:%M")
                .map_err(|_| BacktestError::Execution("stock_pool_execution_window_invalid".into()))?;
            if pending.date != date || clock.is_some_and(|clock| clock >= end) { expired.push(id.clone()); }
        }
        for id in expired {
            if let Some(pending) = self.deferred_stock_pools.borrow_mut().remove(&id) {
                report.diagnostics.push(format!("stock_pool_unsubmitted_phase_expired generation={} date={date} no_buy_order_created=true",pending.contract.generation));
            }
        }
        if self.has_open_orders() || clock.is_none() { return Ok(()); }
        let pending = std::mem::take(&mut *self.deferred_stock_pools.borrow_mut());
        for (id, pending) in pending {
            let now = clock.expect("clock checked above");
            let start = NaiveTime::parse_from_str(&pending.contract.rule.window_start, "%H:%M")
                .map_err(|_| BacktestError::Execution("stock_pool_execution_window_invalid".into()))?;
            if now < start || !pool::stock_pool_is_trading_minute(now.hour() * 60 + now.minute()) {
                self.deferred_stock_pools.borrow_mut().insert(id, pending);
                continue;
            }
            let prior_followup = self.runtime_stock_pool_followup.replace(true);
            let prior_decision = self.runtime_decision_date.replace(Some(pending.contract.signal_date));
            let prior_created = self.runtime_order_created_date.replace(Some(date));
            let order_start = report.order_events.len();
            let fill_start = report.fill_events.len();
            report.diagnostics.push(format!("stock_pool_resume_after_order_reports generation={} clock={} cash={}",pending.contract.generation,clock.unwrap(),portfolio.cash()));
            let result = self.process_stock_pool_contract_phase(date, portfolio, data, &pending.contract,
                &mut session.intraday_turnover, &mut session.execution_cursors, &mut session.global_execution_cursor,
                &mut session.commission_state, report, pending.buy_only, Some(&pending.initial_holdings));
            self.runtime_stock_pool_followup.set(prior_followup);
            self.runtime_decision_date.set(prior_decision);
            self.runtime_order_created_date.set(prior_created);
            result?;
            Self::annotate_report_range(report, order_start, fill_start, pending.contract.signal_date, date, date);
        }
        Ok(())
    }

    fn pool_quote_inputs(
        &self,
        date: NaiveDate,
        data: &DataSet,
        symbols: &BTreeSet<String>,
        execution_clock: Option<NaiveDateTime>,
        cumulative_conditions: bool,
    ) -> Result<(Vec<pool::MarketSnapshot>, Vec<String>), BacktestError> {
        let mut unavailable = Vec::new();
        let quotes = symbols
            .iter()
            .map(|symbol| {
                let snapshot = data.market(date, symbol).ok_or_else(|| {
                    BacktestError::Execution(format!(
                        "stock_pool_execution_snapshot_missing:{symbol}:{date}"
                    ))
                })?;
                let instrument = data.instruments().get(symbol).ok_or_else(|| {
                    BacktestError::Execution(format!("stock_pool_instrument_missing:{symbol}"))
                })?;
                let fallback = self.pool_etf_fallback_reference(date, data, symbol, execution_clock)?;
                let (price, prev, volume, amount, bid, ask, buy_price, sell_price) = if let Some(reference) = fallback {
                    let calibration = self.slippage_calibration(data, snapshot)?;
                    (reference.price, snapshot.prev_close, None, None, None, None,
                        self.quote_execution_price(snapshot, OrderSide::Buy, reference.price, None, calibration.as_ref())?,
                        self.quote_execution_price(snapshot, OrderSide::Sell, reference.price, None, calibration.as_ref())?)
                } else if self
                    .matching_type_uses_intraday_quotes()
                {
                    let time = self
                        .runtime_intraday_start_time
                        .get()
                        .or(self.intraday_execution_start_time)
                        .ok_or_else(|| {
                            BacktestError::Execution(
                                "stock_pool_intraday_execution_clock_required".into(),
                            )
                        })?;
                    let clock = execution_clock
                        .unwrap_or(date.and_time(time))
                        .max(date.and_time(time));
                    let quote = data
                        .execution_quotes_on(date, symbol)
                        .iter()
                        .rev()
                        .find(|quote| quote.timestamp <= clock)
                        .ok_or_else(|| {
                            BacktestError::Execution(format!(
                                "stock_pool_execution_quote_missing:{symbol}:{clock}"
                            ))
                        })?;
                    if !quote.last_price.is_finite() || quote.last_price <= 0.0 {
                        return Err(BacktestError::Execution(format!(
                            "stock_pool_execution_quote_invalid:{symbol}:{clock}"
                        )));
                    }
                    let raw_buy = self
                        .select_quote_reference_price(
                            snapshot,
                            quote,
                            OrderSide::Buy,
                            self.matching_type_for_algo_request(None),
                        )
                        .ok_or_else(|| {
                            BacktestError::Execution(format!(
                                "stock_pool_buy_reference_missing:{symbol}:{clock}"
                            ))
                        })?;
                    let raw_sell = self
                        .select_quote_reference_price(
                            snapshot,
                            quote,
                            OrderSide::Sell,
                            self.matching_type_for_algo_request(None),
                        )
                        .ok_or_else(|| {
                            BacktestError::Execution(format!(
                                "stock_pool_sell_reference_missing:{symbol}:{clock}"
                            ))
                        })?;
                    let calibration = self.slippage_calibration(data, snapshot)?;
                    let buy = self.quote_execution_price(
                        snapshot,
                        OrderSide::Buy,
                        raw_buy,
                        None,
                        calibration.as_ref(),
                    )?;
                    let sell = self.quote_execution_price(
                        snapshot,
                        OrderSide::Sell,
                        raw_sell,
                        None,
                        calibration.as_ref(),
                    )?;
                    let totals = if cumulative_conditions {
                        match data.execution_session_totals(symbol, clock) {
                            Ok(totals) => Some(totals),
                            Err(reason) => { unavailable.push(reason); None }
                        }
                    } else { None };
                    (
                        quote.last_price,
                        snapshot.prev_close,
                        totals.map(|total| total.0),
                        totals.map(|total| total.1),
                        Some(quote.bid1),
                        Some(quote.ask1),
                        buy,
                        sell,
                    )
                } else {
                    let price = snapshot.price(self.effective_execution_price_field(date));
                    if !price.is_finite() || price <= 0.0 {
                        return Err(BacktestError::Execution(format!(
                            "stock_pool_execution_price_missing:{symbol}:{date}"
                        )));
                    }
                    // A daily open does not reveal the session's volume/turnover.
                    let completed = self.effective_execution_price_field(date) == PriceField::Close;
                    let totals = if cumulative_conditions && !completed {
                        let at = execution_clock.unwrap_or_else(|| date.and_hms_opt(9,30,0).unwrap());
                        match data.execution_session_totals(symbol, at) {
                            Ok(totals) => Some(totals),
                            Err(reason) => { unavailable.push(reason); None }
                        }
                    } else { None };
                    let amount = if completed && cumulative_conditions {
                        data.factor(date, symbol).and_then(|row| row.extra_factors.get("amount")).copied()
                            .map(|value| decimal(value, "amount")).transpose()?
                    } else { totals.map(|total| total.1) };
                    (
                        price,
                        snapshot.prev_close,
                        if completed { Some(Decimal::from(snapshot.volume)) } else { totals.map(|total| total.0) },
                        amount,
                        None,
                        None,
                        self.snapshot_execution_price(data, snapshot, OrderSide::Buy, None)?,
                        self.snapshot_execution_price(data, snapshot, OrderSide::Sell, None)?,
                    )
                };
                Ok(pool::MarketSnapshot {
                    symbol: symbol.clone(),
                    last_price: decimal(price, "price")?,
                    prev_close: Some(decimal(prev, "prev_close")?),
                    volume,
                    turnover: amount,
                    bid_price_1: bid.map(|v| decimal(v, "bid")).transpose()?,
                    ask_price_1: ask.map(|v| decimal(v, "ask")).transpose()?,
                    is_kcb: Some(instrument.board.eq_ignore_ascii_case("KSH")),
                    instrument_rules: Some(pool::StockPoolInstrumentRules {
                        price_tick: decimal(snapshot.price_tick, "price_tick")?,
                        quantity_step: instrument.order_step_size().into(),
                        minimum_buy_quantity: instrument.minimum_order_quantity().into(),
                    }),
                    buy_sizing_price: Some(decimal(buy_price, "buy_price")?),
                    sell_sizing_price: Some(decimal(sell_price, "sell_price")?),
                })
            })
            .collect::<Result<Vec<_>, BacktestError>>()?;
        Ok((quotes, unavailable))
    }

    fn pool_etf_fallback_reference(&self, date: NaiveDate, data: &DataSet, symbol: &str, clock: Option<NaiveDateTime>) -> Result<Option<crate::etf_execution::EtfFallbackReference>, BacktestError> {
        if !self.matching_type_uses_intraday_quotes() || !self.has_verified_etf_minute_absence(date, symbol) {
            return Ok(None);
        }
        let time = self.runtime_intraday_start_time.get().or(self.intraday_execution_start_time)
            .ok_or_else(|| BacktestError::Execution("etf_daily_open_fallback: execution clock missing".into()))?;
        let at = clock.unwrap_or(date.and_time(time)).max(date.and_time(time));
        crate::etf_execution::reference(data, symbol, at).map(Some)
    }

    pub(super) fn process_stock_pool_contract(
        &self,
        date: NaiveDate,
        portfolio: &mut PortfolioState,
        data: &DataSet,
        contract: &pool::FrozenStockPoolIntent,
        intraday_turnover: &mut BTreeMap<String, u32>,
        execution_cursors: &mut IntradayExecutionLedger,
        global_execution_cursor: &mut Option<NaiveDateTime>,
        commission_state: &mut BTreeMap<u64, f64>,
        report: &mut BrokerExecutionReport,
    ) -> Result<(), BacktestError> {
        self.process_stock_pool_contract_phase(date, portfolio, data, contract, intraday_turnover,
            execution_cursors, global_execution_cursor, commission_state, report, false, None)
    }

    fn process_stock_pool_contract_phase(
        &self, date: NaiveDate, portfolio: &mut PortfolioState, data: &DataSet,
        contract: &pool::FrozenStockPoolIntent, intraday_turnover: &mut BTreeMap<String, u32>,
        execution_cursors: &mut IntradayExecutionLedger, global_execution_cursor: &mut Option<NaiveDateTime>,
        commission_state: &mut BTreeMap<u64, f64>, report: &mut BrokerExecutionReport, buy_only: bool,
        initial_holdings: Option<&BTreeSet<String>>,
    ) -> Result<(), BacktestError> {
        if contract.signal_date > date
            || contract.frozen_equity < Decimal::ZERO
            || contract.generation.is_empty()
            || contract.pool_id.trim().is_empty()
        {
            return Err(BacktestError::Execution(
                "stock_pool_frozen_intent_invalid".into(),
            ));
        }
        if self.matching_type == MatchingType::NextBarOpen && contract.signal_date >= date {
            return Err(BacktestError::Execution(
                "stock_pool_next_open_requires_prior_signal".into(),
            ));
        }
        let mut selection = contract.selection.clone();
        let mut members = contract.members.clone();
        for symbol in &contract.selection.requested_symbols {
            let instrument = data.instruments().get(symbol).ok_or_else(|| {
                BacktestError::Execution(format!("stock_pool_instrument_missing:{symbol}"))
            })?;
            if portfolio.position(symbol).is_none()
                && let Some(reason) = instrument.dated_market_absence_reason(date)
            {
                selection.requested_symbols.retain(|v| v != symbol);
                selection.normal_trading_symbols.retain(|v| v != symbol);
                selection.risk_eligible_symbols.retain(|v| v != symbol);
                selection.final_symbols.retain(|v| v != symbol);
                members.retain(|v| &v.symbol != symbol);
                report.diagnostics.push(format!(
                    "stock_pool_market_absence symbol={symbol} date={date} reason={reason}"
                ));
            }
        }
        let mut scope = selection
            .requested_symbols
            .iter()
            .cloned()
            .collect::<BTreeSet<_>>();
        scope.extend(portfolio.positions().keys().cloned());
        let before_positions = initial_holdings.cloned().unwrap_or_else(|| portfolio.positions().keys().cloned().collect());
        let official_dates = data.calendar().iter().collect::<Vec<_>>();
        let initial_positions = pool_positions(portfolio, date)?;
        let state = portfolio
            .stock_pool_execution_state(&contract.pool_id)
            .observe(
                contract.signal_date,
                date,
                &official_dates,
                &members,
                &initial_positions,
            )
            .map_err(BacktestError::Execution)?;
        portfolio
            .set_stock_pool_execution_state(&contract.pool_id, state)
            .map_err(BacktestError::Execution)?;
        let superseded = self.deferred_etf_targets.borrow_mut().replace_generation(&contract.pool_id, &contract.generation);
        if superseded > 0 { report.diagnostics.push(format!("etf_daily_open_fallback:superseded pool={} generation={} targets={superseded}", contract.pool_id, contract.generation)); }
        if self.has_open_orders() {
            self.deferred_stock_pools.borrow_mut().insert(contract.pool_id.clone(), DeferredStockPoolExecution {
                date, contract: Box::new(contract.clone()), buy_only, symbols: scope, initial_holdings: before_positions,
            });
            report
                .diagnostics
                .push("stock_pool_waiting_for_active_orders no_new_intent=true".into());
            return Ok(());
        }
        let mut constraints = contract.constraints.clone();
        constraints.execution_date = Some(date);
        constraints.frozen_positions.clear();
        let mut quote_scope = scope.clone();
        for symbol in &scope {
            let paused = data.market(date, symbol).is_some_and(|row| row.paused)
                || data
                    .candidate(date, symbol)
                    .is_some_and(|row| row.is_paused);
            if !paused {
                continue;
            }
            quote_scope.remove(symbol);
            if let Some(position) = portfolio
                .position(symbol)
                .filter(|position| position.quantity > 0)
            {
                constraints.frozen_positions.insert(
                    symbol.clone(),
                    pool::FrozenStockPoolPosition {
                        trade_date: date,
                        reason: "paused".into(),
                        valuation_price: decimal(position.last_price, "paused_holding_valuation")?,
                    },
                );
            } else {
                selection
                    .normal_trading_symbols
                    .retain(|item| item != symbol);
                selection
                    .risk_eligible_symbols
                    .retain(|item| item != symbol);
                selection.final_symbols.retain(|item| item != symbol);
                selection
                    .exclusion_reasons
                    .entry(symbol.clone())
                    .or_default()
                    .push("paused".into());
            }
        }
        // All delayed symbols in a generation share immutable configuration.
        // Do not duplicate an N-member pool N times in a large mixed pool.
        let mut deferred_configuration = None;
        for side in [pool::OrderSide::Sell, pool::OrderSide::Buy] {
            if buy_only && side == pool::OrderSide::Sell { continue; }
            if side == pool::OrderSide::Buy && self.has_open_orders()
                && self.effective_rebalance_cash_mode() == RebalanceCashMode::SellThenBuy {
                self.deferred_stock_pools.borrow_mut().insert(contract.pool_id.clone(), DeferredStockPoolExecution {
                    date, contract: Box::new(contract.clone()), buy_only: true, symbols: quote_scope.clone(), initial_holdings: before_positions.clone(),
                });
                report.diagnostics.push(format!("stock_pool_waiting_for_sell_reports generation={} no_buy_order_created=true",contract.generation));
                break;
            }
            let mut fallback_references = BTreeMap::new();
            for symbol in &quote_scope {
                if let Some(reference) = self.pool_etf_fallback_reference(date, data, symbol, *global_execution_cursor)? {
                    let condition = if side == pool::OrderSide::Buy { &contract.rule.buy_condition } else { &contract.rule.sell_condition };
                    if !condition.trim().is_empty() {
                        return Err(BacktestError::Execution(format!("etf_daily_open_fallback: intraday condition evidence unavailable symbol={symbol} side={side:?}; daily reference is not a minute or tick signal")));
                    }
                    fallback_references.insert(symbol.clone(), reference);
                }
            }
            let (quotes, unavailable) = self.pool_quote_inputs(date, data, &quote_scope, *global_execution_cursor,
                crate::stock_pool_quote_facts::requires_session_totals(&contract.rule))?;
            let positions = pool_positions(portfolio, date)?;
            let execution_state = portfolio
                .stock_pool_execution_state(&contract.pool_id)
                .observe(
                    contract.signal_date,
                    date,
                    &official_dates,
                    &members,
                    &positions,
                )
                .map_err(BacktestError::Execution)?;
            constraints.pending_entry_symbols = execution_state.pending_symbols();
            constraints.prior_target_weights = execution_state.last_target_weights.clone();
            constraints.position_action_bases = execution_state.position_action_bases_for(&contract.generation);
            constraints.next_day_outside_exit_symbols = execution_state.next_day_exit_symbols(date);
            let account = pool::AccountSnapshot {
                total_equity: contract.frozen_equity,
                cash: decimal(portfolio.cash(), "cash")?,
                frozen_cash: Decimal::ZERO,
            };
            constraints.automatic_permissions.clear();
            if contract.rule.automatic_trade_protection.enabled() {
                for symbol in &scope {
                    let position = portfolio.position(symbol).filter(|p| p.quantity > 0);
                    let sold = self
                        .same_day_sold_symbols
                        .borrow()
                        .iter()
                        .rev()
                        .find(|(day, symbols)| **day <= date && symbols.contains(symbol))
                        .map(|(day, _)| *day);
                    let evidence = HoldingLifecycleEvidence {
                        has_position: position.is_some(),
                        opened_date: position.and_then(|p| p.opened_date()),
                        last_buy_date: position.and_then(|p| p.last_buy_date()),
                        last_sell_date: sold,
                    };
                    let permission = contract
                        .rule
                        .automatic_trade_protection
                        .evaluate(symbol, date, &evidence, data.calendar())
                        .map_err(BacktestError::Execution)?;
                    constraints
                        .automatic_permissions
                        .insert(symbol.clone(), permission);
                }
            }
            if side == pool::OrderSide::Buy {
                for (symbol, reference) in &fallback_references {
                    if !reference.immediate {
                        // The sell leg was queued, not filled. Keep its real
                        // holdings/slots and do not finance buys with proceeds
                        // from the following session.
                        constraints.automatic_permissions.entry(symbol.clone()).or_default()
                            .sell_denial.get_or_insert("etf_daily_open_deferred");
                    }
                }
            }
            if self
                .risk_config
                .static_rules
                .forbid_same_day_rebuy_after_sell
            {
                constraints.same_day_sold_symbols.extend(
                    self.same_day_sold_symbols
                        .borrow()
                        .get(&date)
                        .into_iter()
                        .flatten()
                        .cloned(),
                );
            }
            constraints.same_day_sold_symbols.extend(
                before_positions
                    .iter()
                    .filter(|symbol| portfolio.position(symbol).is_none_or(|p| p.quantity == 0))
                    .cloned(),
            );
            let fee =
                |symbol: &str, side: pool::OrderSide, gross: Decimal| -> Result<Decimal, String> {
                    let amount = gross
                        .to_f64()
                        .ok_or("stock_pool_cost_amount_out_of_range")?;
                    decimal(
                        self.cost_model
                            .calculate_for_instrument(
                                date,
                                if side == pool::OrderSide::Buy {
                                    OrderSide::Buy
                                } else {
                                    OrderSide::Sell
                                },
                                amount,
                                data.instruments().get(symbol),
                            )
                            .total(),
                        "fee",
                    )
                    .map_err(|e| e.to_string())
                };
            let plan = pool::build_stock_pool_target_plan_with_fee_model(
                &selection,
                &members,
                &contract.rule,
                &account,
                &positions,
                &quotes,
                contract.invest_ratio_bps,
                contract.reserve_cash,
                &contract.out_of_pool_policy,
                "full_rebalance",
                &constraints,
                &contract.generation,
                Decimal::ZERO,
                Decimal::ZERO,
                Decimal::ZERO,
                Some(&fee),
            )
            .map_err(|error| BacktestError::Execution(if !unavailable.is_empty()
                && (error.contains("requires volume") || error.contains("requires amount")) {
                    format!("{error}; {}", unavailable.join("; "))
                } else { error }))?;
            report.diagnostics.extend(unavailable.into_iter().map(|reason| format!("stock_pool_quote_fact_unavailable {reason}")));
            let mut updated = execution_state
                .record_plan(contract.signal_date, &contract.generation, &plan)
                .map_err(BacktestError::Execution)?;
            for (symbol, reference) in &fallback_references {
                if !reference.immediate && let Some(entry) = updated.entries.get_mut(symbol) {
                    // The signal only fixes money, not shares at a stale close.
                    entry.completion_quantity = None;
                }
            }
            portfolio
                .set_stock_pool_execution_state(&contract.pool_id, updated)
                .map_err(BacktestError::Execution)?;
            report.diagnostics.push(format!("stock_pool_plan phase={side:?} generation={} requested_bps={} effective_bps={} budget={}",contract.generation,plan.requested_invest_ratio_bps,plan.effective_invest_ratio_bps,plan.budget));
            let max_positions = constraints
                .target_holding_count
                .unwrap_or(selection.final_symbols.len());
            for row in plan.rows {
                if side == pool::OrderSide::Buy && row.side.is_none() {
                    report.diagnostics.push(format!(
                        "stock_pool_decision symbol={} status={} current={} target={} reason={}",
                        row.symbol,
                        row.status,
                        row.current_quantity,
                        row.target_quantity,
                        row.reason
                    ));
                }
                if row.side != Some(side) {
                    continue;
                }
                if let Some(reference) = fallback_references.get(&row.symbol) {
                    let time = self.runtime_intraday_start_time.get().or(self.intraday_execution_start_time).expect("fallback clock validated");
                    let at = global_execution_cursor.unwrap_or(date.and_time(time)).max(date.and_time(time));
                    if !reference.immediate {
                        report.diagnostics.push(format!("etf_daily_open_fallback:deferred symbol={} signal_at={at} reference_date={} reference_price={} target_value={} execute_on={:?}", row.symbol, reference.reference_date, reference.price, row.target_value, reference.execute_on));
                        let deferred = deferred_configuration.get_or_insert_with(|| (
                            std::sync::Arc::new(contract.rule.clone()), std::sync::Arc::new(members.clone()),
                        ));
                        let opening_date=reference.execute_on.map(|day|day.to_string()).unwrap_or_else(||"回测区间外（后续日历未加载）".into());
                        etf_activity(report,date,&row.symbol,side,format!("ETF 顺延执行：信号 {at}，参考 {} 收盘 {}，目标金额 {}，下一正式开盘日 {opening_date}；未生成成交。",reference.reference_date,reference.price,row.target_value));
                        self.deferred_etf_targets.borrow_mut().upsert(crate::etf_execution::DeferredEtfTarget {
                            pool_id:contract.pool_id.clone(), generation:contract.generation.clone(), symbol:row.symbol.clone(),
                            signal_date:contract.signal_date, signal_at:at, execute_on:reference.execute_on,
                            target_value:row.target_value, target_weight_bps:row.target_weight_bps, side,
                            max_positions, rule:std::sync::Arc::clone(&deferred.0), members:std::sync::Arc::clone(&deferred.1),
                            reason:row.source_intent.clone().unwrap_or_else(||"stock_pool_target".into()),
                        });
                        continue;
                    }
                }
                if side == pool::OrderSide::Buy
                    && portfolio
                        .position(&row.symbol)
                        .is_none_or(|p| p.quantity == 0)
                    && Self::positive_position_count(portfolio) >= max_positions
                {
                    report.diagnostics.push(format!(
                        "stock_pool_buy_deferred symbol={} reason=occupied_position_slots",
                        row.symbol
                    ));
                    continue;
                }
                let target = row.target_quantity.to_i32().ok_or_else(|| {
                    BacktestError::Execution("stock_pool_target_quantity_out_of_range".into())
                })?;
                let fallback_reason = fallback_references.contains_key(&row.symbol).then(|| format!("{}: etf_daily_open_fallback signal_date={} execution_date={date}", row.source_intent.as_deref().unwrap_or("stock_pool_target"), contract.signal_date));
                let reason = fallback_reason.as_deref().unwrap_or_else(|| row.source_intent.as_deref().unwrap_or("stock_pool_target"));
                let first_fill = report.fill_events.len();
                if fallback_references.contains_key(&row.symbol) {
                    report.diagnostics.push(format!("etf_daily_open_fallback:opening symbol={} signal_date={} execution_date={date}", row.symbol, contract.signal_date));
                    etf_activity(report,date,&row.symbol,side,format!("ETF 日线开盘回退：信号日 {}，执行日 {date}，使用正式日线开盘价；不是分钟成交行情。",contract.signal_date));
                }
                let mut execute = || if let Some(price) = row.limit_price {
                    self.process_limit_target_shares(
                        date,
                        portfolio,
                        data,
                        &row.symbol,
                        target,
                        price.to_f64().ok_or_else(|| {
                            BacktestError::Execution("stock_pool_limit_price_out_of_range".into())
                        })?,
                        reason,
                        intraday_turnover,
                        execution_cursors,
                        global_execution_cursor,
                        commission_state,
                        report,
                    )
                } else {
                    self.process_target_shares(
                        date,
                        portfolio,
                        data,
                        &row.symbol,
                        target,
                        reason,
                        intraday_turnover,
                        execution_cursors,
                        global_execution_cursor,
                        commission_state,
                        report,
                    )
                };
                if fallback_references.contains_key(&row.symbol) {
                    self.with_etf_daily_open(execute)?;
                    for fill in &mut report.fill_events[first_fill..] {
                        fill.execution_start_timestamp.get_or_insert(date.and_time(crate::etf_execution::opening_time()));
                        fill.execution_timestamp.get_or_insert(date.and_time(crate::etf_execution::opening_time()));
                    }
                } else { execute()?; }
            }
        }
        Ok(())
    }

    pub(crate) fn pending_etf_target_count(&self) -> usize {
        self.deferred_etf_targets.borrow().len()
    }

    /// Called at the opening clock, after settlement/corporate actions and
    /// auction callbacks. It never sends a stock order or replays a strategy.
    pub(crate) fn execute_deferred_etf_targets(&self, date: NaiveDate, portfolio: &mut PortfolioState, data: &DataSet) -> Result<BrokerExecutionReport, BacktestError> {
        let mut report = BrokerExecutionReport::default();
        if self.has_open_orders() {
            if self.pending_etf_target_count() > 0 {
                report.diagnostics.push("etf_daily_open_fallback:waiting_for_active_orders".into());
            }
            return Ok(report);
        }
        let due = self.deferred_etf_targets.borrow_mut().take_due(date);
        let dates = data.calendar().iter().collect::<Vec<_>>();
        for target in due {
            let instrument = data.instrument(&target.symbol).ok_or_else(|| BacktestError::Execution("etf_daily_open_fallback: instrument identity missing at execution".into()))?;
            if !instrument.is_exchange_traded_fund() { return Err(BacktestError::Execution("etf_daily_open_fallback: instrument identity changed".into())); }
            if let Some(reason) = instrument.dated_market_absence_reason(date) {
                report.diagnostics.push(format!("etf_daily_open_fallback:blocked symbol={} date={date} reason={reason}", target.symbol));
                continue;
            }
            let snapshot = data.market(date, &target.symbol).ok_or_else(|| BacktestError::Execution(format!("etf_daily_open_fallback: daily_open_missing symbol={} date={date}", target.symbol)))?;
            if !snapshot.open.is_finite() || snapshot.open <= 0.0 {
                return Err(BacktestError::Execution(format!("etf_daily_open_fallback: daily_open_invalid symbol={} date={date}", target.symbol)));
            }
            let position = portfolio.position(&target.symbol).filter(|p| p.quantity > 0);
            let before_quantity = position.map_or(0, |p| p.quantity);
            let permission = target.rule.automatic_trade_protection.evaluate(&target.symbol, date, &HoldingLifecycleEvidence {
                has_position:position.is_some(), opened_date:position.and_then(|p| p.opened_date()), last_buy_date:position.and_then(|p| p.last_buy_date()),
                last_sell_date:self.same_day_sold_symbols.borrow().iter().rev().find(|(day, symbols)| **day <= date && symbols.contains(&target.symbol)).map(|(day, _)| *day),
            }, data.calendar()).map_err(BacktestError::Execution)?;
            let denial = if target.side == pool::OrderSide::Buy {
                permission.buy_denial.or(permission.max_holding_exit.then_some("max_holding_exit_pending"))
            } else { permission.sell_denial };
            if let Some(denial) = denial {
                report.diagnostics.push(format!("etf_daily_open_fallback:protected symbol={} date={date} reason={denial}", target.symbol));
                etf_activity(&mut report,date,&target.symbol,target.side,format!("ETF 顺延目标受持有保护限制：{denial}；未提交委托。"));
                continue;
            }
            if target.side == pool::OrderSide::Buy && before_quantity == 0 && Self::positive_position_count(portfolio) >= target.max_positions {
                report.diagnostics.push(format!("etf_daily_open_fallback:blocked symbol={} reason=occupied_position_slots", target.symbol));
                continue;
            }
            let value = target.target_value.to_f64().ok_or_else(|| BacktestError::Execution("etf_daily_open_fallback: target value out of range".into()))?;
            let current_value = snapshot.open * f64::from(before_quantity);
            let satisfied = (target.side == pool::OrderSide::Buy && value <= current_value)
                || (target.side == pool::OrderSide::Sell && value >= current_value);
            let reason = format!("{}: etf_daily_open_fallback signal_at={} execution_at={} target_value={}", target.reason, target.signal_at, date.and_time(crate::etf_execution::opening_time()), target.target_value);
            let mut sub = BrokerExecutionReport::default();
            if !satisfied {
                let (_, limit) = pool::resolve_stock_pool_order_price(&target.rule, &target.symbol, decimal(snapshot.open, "etf_open")?, target.side, decimal(snapshot.price_tick, "etf_tick")?).map_err(BacktestError::Execution)?;
                let intent = match limit {
                    Some(limit) => OrderIntent::LimitTargetValue { symbol:target.symbol.clone(), target_value:value, limit_price:limit.to_f64().ok_or_else(|| BacktestError::Execution("ETF limit out of range".into()))?, reason:reason.clone() },
                    None => OrderIntent::TargetValue { symbol:target.symbol.clone(), target_value:value, reason:reason.clone() },
                };
                let old_time = self.runtime_intraday_start_time.replace(Some(crate::etf_execution::opening_time()));
                let old_origin = self.runtime_resting_order_origin.replace(Some(RestingOrderOrigin { created_date:Some(target.signal_at.date()), submission_time:Some(target.signal_at.time()), accepted_date:date }));
                let outcome = self.with_etf_daily_open(|| self.execute_with_event_dates(date, target.signal_date, target.signal_at.date(), portfolio, data, &StrategyDecision {
                    order_intents:vec![OrderIntent::WithTimeInForce { intent:Box::new(intent), time_in_force:OrderTimeInForce::Day }], ..Default::default()
                }));
                self.runtime_intraday_start_time.set(old_time);
                self.runtime_resting_order_origin.set(old_origin);
                sub = outcome?;
            }
            // The actual open determines the full requested shares. A clipped
            // or rejected execution must not be recorded as completed entry.
            let order = sub.order_events.iter().rev().find(|order| order.symbol == target.symbol);
            let goal_quantity = order.map_or(before_quantity, |order| match order.side {
                OrderSide::Buy => before_quantity.saturating_add(order.requested_quantity),
                OrderSide::Sell => before_quantity.saturating_sub(order.requested_quantity),
            });
            let status = if satisfied || (order.is_none() && !self.has_open_orders()) { "BELOW_MINIMUM_TRADE_UNIT_ALREADY_SATISFIED" } else { "READY" };
            let positions = pool_positions(portfolio, date)?;
            let state = portfolio.stock_pool_execution_state(&target.pool_id)
                .observe(target.signal_date, date, &dates, &target.members, &positions).map_err(BacktestError::Execution)?
                .record_targets(target.signal_date, &target.generation, [crate::stock_pool_state::StockPoolGoalObservation {
                    symbol:&target.symbol, target_weight_bps:target.target_weight_bps, target_value:target.target_value,
                    current_quantity:before_quantity.into(), target_quantity:goal_quantity.into(), status,
                }]).map_err(BacktestError::Execution)?
                .observe(target.signal_date, date, &dates, &target.members, &positions).map_err(BacktestError::Execution)?;
            portfolio.set_stock_pool_execution_state(&target.pool_id, state).map_err(BacktestError::Execution)?;
            for fill in &mut sub.fill_events {
                fill.decision_date.get_or_insert(target.signal_date);
                fill.order_created_date.get_or_insert(target.signal_at.date());
                fill.execution_date.get_or_insert(date);
                fill.execution_start_timestamp.get_or_insert(date.and_time(crate::etf_execution::opening_time()));
                fill.execution_timestamp.get_or_insert(date.and_time(crate::etf_execution::opening_time()));
            }
            for order in &mut sub.order_events {
                order.decision_date.get_or_insert(target.signal_date);
                order.order_created_date.get_or_insert(target.signal_at.date());
                order.execution_date.get_or_insert(date);
            }
            report.diagnostics.push(reason);
            etf_activity(&mut report,date,&target.symbol,target.side,format!("ETF 顺延目标开盘处理：原信号 {}，本次 {date} 09:30，冻结目标金额 {}，持仓 {before_quantity} → {}；按本日开盘价、资金与风控重新定量。",target.signal_at,target.target_value,portfolio.position(&target.symbol).map_or(0,|position|position.quantity)));
            report.order_events.extend(sub.order_events);
            report.fill_events.extend(sub.fill_events);
            report.position_events.extend(sub.position_events);
            report.account_events.extend(sub.account_events);
            report.process_events.extend(sub.process_events);
            report.diagnostics.extend(sub.diagnostics);
        }
        Ok(report)
    }
}