合并分钟行情与策略定时事件时钟

2026-08-28 01:42:58 +08:00
parent 6c47c33cab
commit 1a2e247c8d
2 changed files with 181 additions and 34 deletions
@@ -85,6 +85,8 @@ Source Lake 日线成交量保留原始可用性合同：源 `volume=null` 与

 盘后固定价格不是策略类型，也不是 `matchingType`。自 2026-07-06 起，只有实际同日提交时间落在 15:00–15:30 的普通委托才由 broker 进入盘后固定价格执行阶段；15:00–15:04 的委托等待到 15:05，15:05–15:30 按官方收盘价和真实盘后成交量撮合，不叠加滑点，未成交余量不跨日。窗口外委托继续沿用连续竞价、当前收盘或下一交易日开盘合同；`next_bar_open` 策略即使在 15:00 生成信号，也不得被改写为同日盘后委托。缺失盘后行情时必须明确不成交，禁止回退全天成交量或 15:00 前分钟行情。

+分钟回放使用行情时间戳与策略定时事件的有序合并时钟。显式 `physical_time` 即使该分钟没有预加载行情也必须触发，并由执行层按需查询该时点价格；同一时间戳只形成一个事件，scheduler 回调先于 `on_minute`。多个时间点分别执行，禁止压成最后一个时间或依赖已有 BAR 才触发。
+
 `holdUntilExit=true` 与 `stopTakeReferencePriceMode=signal_day_post_adjusted_close` 组合表示持久模型组合语义：股票进入模型目标后即记录信号日和后复权参考价，不以买单是否成交为前提。涨停、停牌或其他执行风控导致买单未成交时，模型成员仍占用目标槽位、每天累计模型持有日并继续生成目标仓位；达到止盈、止损或最大模型持有期后才从模型组合移除。实际订单仍由成交日风控独立决定，不得用实际持仓集合覆盖模型目标集合。

 ## 内置微盘策略
@@ -1,6 +1,6 @@
 use std::collections::{BTreeMap, BTreeSet};

-use chrono::{Datelike, Duration, NaiveDate, NaiveTime, Timelike};
+use chrono::{Datelike, Duration, NaiveDate, NaiveTime};
 use serde::{Deserialize, Serialize};
 use thiserror::Error;

@@ -2747,20 +2747,41 @@ where
                    ProcessEventKind::Minute,
                    ProcessEventKind::PostMinute,
                ]);
-                let minute_schedule_all_times = schedule_rules
+                let minute_all_time_rules = schedule_rules
                    .iter()
-                    .any(|rule| rule.stage == ScheduleStage::Minute && rule.time_rule.is_none());
-                let minute_schedule_minutes = schedule_rules
+                    .filter(|rule| {
+                        rule.stage == ScheduleStage::Minute && rule.time_rule.is_none()
+                    })
+                    .cloned()
+                    .collect::<Vec<_>>();
+                let minute_schedule_all_times = !minute_all_time_rules.is_empty();
+                let mut minute_schedule_timestamps = schedule_rules
                    .iter()
                    .filter(|rule| rule.stage == ScheduleStage::Minute)
-                    .filter_map(|rule| rule.time_rule.as_ref()?.minute_of_day())
-                    .collect::<BTreeSet<_>>();
+                    .filter(|rule| scheduler.is_due_on(decision_date, rule))
+                    .filter_map(|rule| {
+                        let minute = rule.time_rule.as_ref()?.minute_of_day()?;
+                        NaiveTime::from_hms_opt(minute / 60, minute % 60, 0)
+                    })
+                    .map(|time| execution_date.and_time(time))
+                    .collect::<BTreeSet<_>>()
+                    .into_iter()
+                    .peekable();
                let mut minute_group = Vec::new();
-                while let Some(first_quote) = minute_quotes.next() {
-                    let minute_timestamp = first_quote.timestamp;
+                // Merge the immutable quote stream with clock events. Equal
+                // timestamps form one event; scheduled callbacks run before
+                // `on_minute` below.
+                loop {
+                    let next_quote_timestamp = minute_quotes.peek().map(|quote| quote.timestamp);
+                    let next_schedule_timestamp = minute_schedule_timestamps.peek().copied();
+                    let Some(minute_timestamp) = next_minute_event_timestamp(
+                        next_quote_timestamp,
+                        next_schedule_timestamp,
+                    ) else {
+                        break;
+                    };
                    let minute_time = minute_timestamp.time();
                    minute_group.clear();
-                    minute_group.push(first_quote);
                    while minute_quotes
                        .peek()
                        .is_some_and(|quote| quote.timestamp == minute_timestamp)
@@ -2771,9 +2792,13 @@ where
                                .expect("peeked minute quote must be available"),
                        );
                    }
-                    let schedule_candidate = minute_schedule_all_times
-                        || minute_schedule_minutes
-                            .contains(&(minute_time.hour() * 60 + minute_time.minute()));
+                    let has_specific_schedule =
+                        next_schedule_timestamp == Some(minute_timestamp);
+                    if has_specific_schedule {
+                        minute_schedule_timestamps.next();
+                    }
+                    let schedule_candidate = has_specific_schedule
+                        || (minute_schedule_all_times && !minute_group.is_empty());
                    if !requires_minute_callbacks
                        && !has_minute_process_listeners
                        && !schedule_candidate
@@ -2799,26 +2824,35 @@ where
                        ProcessEventKind::PreMinute,
                        format!("minute:{minute_timestamp}:pre"),
                    )?;
-                    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(),
-                    )?;
+                    let mut minute_decision = if schedule_candidate {
+                        let event_rules = if has_specific_schedule {
+                            schedule_rules.as_slice()
+                        } else {
+                            minute_all_time_rules.as_slice()
+                        };
+                        collect_scheduled_decisions(
+                            &mut self.strategy,
+                            &scheduler,
+                            execution_date,
+                            ScheduleStage::Minute,
+                            event_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(),
+                        )?
+                    } else {
+                        crate::strategy::StrategyDecision::default()
+                    };
                    if requires_minute_callbacks {
                        for &quote in &minute_group {
                            minute_decision.merge_from(self.strategy.on_minute(
@@ -4320,6 +4354,18 @@ fn stage_datetime(
    time.map(|value| date.and_time(value))
 }

+fn next_minute_event_timestamp(
+    quote_timestamp: Option<chrono::NaiveDateTime>,
+    schedule_timestamp: Option<chrono::NaiveDateTime>,
+) -> Option<chrono::NaiveDateTime> {
+    match (quote_timestamp, schedule_timestamp) {
+        (Some(quote), Some(schedule)) => Some(quote.min(schedule)),
+        (Some(quote), None) => Some(quote),
+        (None, Some(schedule)) => Some(schedule),
+        (None, None) => None,
+    }
+}
+
 fn should_run_minute_events(rules: &[ScheduleRule], subscriptions: &BTreeSet<String>) -> bool {
    !subscriptions.is_empty() || rules.iter().any(|rule| rule.stage == ScheduleStage::Minute)
 }
@@ -4624,7 +4670,7 @@ mod tests {
    use std::rc::Rc;
    use std::sync::{Arc, Mutex};

-    use chrono::{NaiveDate, NaiveTime};
+    use chrono::{NaiveDate, NaiveDateTime, NaiveTime};

    use super::{BacktestConfig, BacktestEngine};
    use crate::broker::{BrokerSimulator, MatchingType, SlippageModel};
@@ -4638,7 +4684,7 @@ mod tests {
    use crate::portfolio::PortfolioState;
    use crate::risk_control::{FidcRiskControlConfig, RiskCheckScope};
    use crate::rules::ChinaEquityRuleHooks;
-    use crate::scheduler::{ScheduleRule, ScheduleStage};
+    use crate::scheduler::{ScheduleRule, ScheduleStage, ScheduleTimeRule};
    use crate::strategy::{OrderIntent, Strategy, StrategyContext, StrategyDecision};

    const SYMBOL: &str = "000001.SZ";
@@ -4720,6 +4766,33 @@ mod tests {
        }
    }

+    #[derive(Debug)]
+    struct MinuteScheduleProbeStrategy {
+        rules: Vec<ScheduleRule>,
+        observed: Rc<RefCell<Vec<NaiveDateTime>>>,
+    }
+
+    impl Strategy for MinuteScheduleProbeStrategy {
+        fn name(&self) -> &str {
+            "minute_schedule_probe"
+        }
+
+        fn schedule_rules(&self) -> Vec<ScheduleRule> {
+            self.rules.clone()
+        }
+
+        fn on_scheduled(
+            &mut self,
+            ctx: &StrategyContext<'_>,
+            _rule: &ScheduleRule,
+        ) -> Result<StrategyDecision, super::BacktestError> {
+            self.observed
+                .borrow_mut()
+                .push(ctx.current_datetime().expect("scheduled event time"));
+            Ok(StrategyDecision::default())
+        }
+    }
+
    #[derive(Debug)]
    struct ScheduledBuyOnDecisionDateStrategy {
        rule: ScheduleRule,
@@ -5311,6 +5384,78 @@ mod tests {
        .expect("backtest run")
    }

+    #[test]
+    fn minute_schedules_fire_at_each_declared_time_without_market_rows_at_those_times() {
+        let date = d(2026, 6, 1);
+        let observed = Rc::new(RefCell::new(Vec::new()));
+        let strategy = MinuteScheduleProbeStrategy {
+            rules: vec![
+                ScheduleRule::daily("first", ScheduleStage::Minute)
+                    .with_time_rule(ScheduleTimeRule::physical_time(10, 17)),
+                ScheduleRule::daily("second", ScheduleStage::Minute)
+                    .with_time_rule(ScheduleTimeRule::physical_time(10, 18)),
+            ],
+            observed: Rc::clone(&observed),
+        };
+        let data = dataset_from_market_and_candidates(
+            vec![market(date, 10.0, 10.2)],
+            vec![candidate(date)],
+        );
+        assert!(
+            data.execution_quotes_on(date, SYMBOL).is_empty(),
+            "the test must prove clock-driven scheduling without minute rows"
+        );
+        let broker = BrokerSimulator::new_with_execution_price(
+            ChinaAShareCostModel::default(),
+            ChinaEquityRuleHooks,
+            PriceField::Close,
+        )
+        .with_matching_type(MatchingType::CurrentBarClose)
+        .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(date),
+            end_date: Some(date),
+            decision_lag_trading_days: 0,
+            execution_price_field: PriceField::Close,
+        };
+
+        BacktestEngine::new(data, strategy, broker, config)
+            .run()
+            .expect("clock-driven minute schedules");
+
+        assert_eq!(
+            observed.borrow().as_slice(),
+            &[
+                date.and_time(NaiveTime::from_hms_opt(10, 17, 0).expect("first time")),
+                date.and_time(NaiveTime::from_hms_opt(10, 18, 0).expect("second time")),
+            ]
+        );
+    }
+
+    #[test]
+    fn minute_event_clock_uses_timestamp_order_and_coalesces_equal_heads() {
+        let date = d(2026, 6, 1);
+        let quote = date.and_hms_opt(10, 18, 0).expect("quote time");
+        let earlier_schedule = date.and_hms_opt(10, 17, 0).expect("schedule time");
+
+        assert_eq!(
+            super::next_minute_event_timestamp(Some(quote), Some(earlier_schedule)),
+            Some(earlier_schedule)
+        );
+        assert_eq!(
+            super::next_minute_event_timestamp(Some(quote), Some(quote)),
+            Some(quote)
+        );
+        assert_eq!(
+            super::next_minute_event_timestamp(None, Some(earlier_schedule)),
+            Some(earlier_schedule)
+        );
+    }
+
    #[test]
    fn current_close_order_at_1500_loads_and_uses_post_close_matching_window() {
        let date = d(2026, 7, 6);