fix(stock-pool): execute verified ETF daily fallbacks with frozen next-open targets

crates/fidc-core/src/broker_stock_pool.rs

@@ -16,6 +16,11 @@ fn decimal(value: f64, label: &str) -> Result<Decimal, BacktestError> {
         .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,
@@ -54,7 +59,13 @@ impl<C: CostModel, R: EquityRuleHooks> BrokerSimulator<C, R> {
                 let instrument = data.instruments().get(symbol).ok_or_else(|| {
                     BacktestError::Execution(format!("stock_pool_instrument_missing:{symbol}"))
                 })?;
-                let (price, prev, volume, amount, bid, ask, buy_price, sell_price) = if self
+                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
@@ -174,6 +185,16 @@ impl<C: CostModel, R: EquityRuleHooks> BrokerSimulator<C, R> {
             .collect()
     }
 
+    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,
@@ -240,6 +261,8 @@ impl<C: CostModel, R: EquityRuleHooks> BrokerSimulator<C, R> {
         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() {
             report
                 .diagnostics
@@ -291,7 +314,20 @@ impl<C: CostModel, R: EquityRuleHooks> BrokerSimulator<C, R> {
             .keys()
             .cloned()
             .collect::<BTreeSet<_>>();
+        // 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] {
+            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 =
                 self.pool_quote_inputs(date, data, &quote_scope, *global_execution_cursor)?;
             let positions = pool_positions(portfolio, date)?;
@@ -340,6 +376,17 @@ impl<C: CostModel, R: EquityRuleHooks> BrokerSimulator<C, R> {
                         .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
@@ -401,9 +448,15 @@ impl<C: CostModel, R: EquityRuleHooks> BrokerSimulator<C, R> {
                 Some(&fee),
             )
             .map_err(BacktestError::Execution)?;
-            let updated = execution_state
+            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)?;
@@ -425,6 +478,26 @@ impl<C: CostModel, R: EquityRuleHooks> BrokerSimulator<C, R> {
                 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)
@@ -440,8 +513,14 @@ impl<C: CostModel, R: EquityRuleHooks> BrokerSimulator<C, R> {
                 let target = row.target_quantity.to_i32().ok_or_else(|| {
                     BacktestError::Execution("stock_pool_target_quantity_out_of_range".into())
                 })?;
-                let reason = row.source_intent.as_deref().unwrap_or("stock_pool_target");
-                if let Some(price) = row.limit_price {
+                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,
@@ -457,7 +536,7 @@ impl<C: CostModel, R: EquityRuleHooks> BrokerSimulator<C, R> {
                         global_execution_cursor,
                         commission_state,
                         report,
-                    )?;
+                    )
                 } else {
                     self.process_target_shares(
                         date,
@@ -471,10 +550,124 @@ impl<C: CostModel, R: EquityRuleHooks> BrokerSimulator<C, R> {
                         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)
+    }
 }