diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index 6aabc22..a6e957a 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -1,5 +1,6 @@ use std::borrow::Cow; -use std::collections::{BTreeMap, HashMap, HashSet}; +use std::cmp::Reverse; +use std::collections::{BTreeMap, BinaryHeap, HashMap, HashSet}; use std::sync::{Arc, OnceLock}; use ahash::AHashMap; @@ -1618,19 +1619,38 @@ impl DataSet { } pub fn execution_quotes_on_date(&self, date: NaiveDate) -> Vec { - let mut quotes = self - .execution_quotes_by_date - .get(&date) - .into_iter() - .flat_map(|rows_by_symbol| rows_by_symbol.values()) - .flat_map(|rows| rows.iter().cloned()) + let Some(rows_by_symbol) = self.execution_quotes_by_date.get(&date) else { + return Vec::new(); + }; + let mut streams = rows_by_symbol + .iter() + .map(|(symbol, rows)| (symbol.as_str(), rows.as_slice())) .collect::>(); - quotes.sort_by(|left, right| { - left.timestamp - .cmp(&right.timestamp) - .then_with(|| left.symbol.cmp(&right.symbol)) - }); - quotes + streams.sort_by_key(|(symbol, _)| *symbol); + let total_rows = streams.iter().map(|(_, rows)| rows.len()).sum(); + let mut heap = BinaryHeap::>::new(); + for (stream_index, (_, rows)) in streams.iter().enumerate() { + if let Some(first) = rows.first() { + heap.push(Reverse((first.timestamp, stream_index, 0))); + } + } + let mut merged = Vec::with_capacity(total_rows); + while let Some(Reverse((_timestamp, stream_index, row_index))) = heap.pop() { + let (_, rows) = streams[stream_index]; + merged.push(rows[row_index].clone()); + let next_index = row_index + 1; + if let Some(next) = rows.get(next_index) { + heap.push(Reverse((next.timestamp, stream_index, next_index))); + } + } + merged + } + + pub fn remove_execution_quotes_on_date(&mut self, date: NaiveDate) -> usize { + self.execution_quotes_by_date + .remove(&date) + .map(|rows_by_symbol| rows_by_symbol.into_values().map(|rows| rows.len()).sum()) + .unwrap_or_default() } pub fn snapshot_components(&self) -> DataSetSnapshotComponents { @@ -3728,6 +3748,68 @@ mod tests { assert_eq!(run_data.execution_quote_count(), 1); } + #[test] + fn execution_quotes_use_stable_k_way_merge_and_release_by_date() { + let date = NaiveDate::parse_from_str("2025-01-02", "%Y-%m-%d").unwrap(); + let data = DataSet::from_components( + vec![Instrument { + symbol: "000001.SZ".to_string(), + name: "平安银行".to_string(), + board: "SZ".to_string(), + round_lot: 100, + listed_at: None, + delisted_at: None, + status: "active".to_string(), + }], + vec![market_row("2025-01-02", 10.0, 1_000_000)], + Vec::new(), + Vec::new(), + vec![benchmark_row("2025-01-02", 12.0)], + ) + .unwrap(); + let quote = |symbol: &str, time: &str| IntradayExecutionQuote { + date, + timestamp: NaiveDateTime::parse_from_str( + &format!("2025-01-02 {time}"), + "%Y-%m-%d %H:%M:%S", + ) + .unwrap(), + symbol: symbol.to_string(), + last_price: 10.0, + bid1: 0.0, + ask1: 0.0, + bid1_volume: 0, + ask1_volume: 0, + volume_delta: 100, + amount_delta: 1_000.0, + trading_phase: Some("continuous".to_string()), + }; + let mut run_data = data.clone(); + run_data.add_execution_quotes(vec![ + quote("000002.SZ", "09:31:00"), + quote("000001.SZ", "09:31:00"), + quote("000002.SZ", "09:30:00"), + quote("000001.SZ", "09:30:00"), + ]); + + let merged = run_data.execution_quotes_on_date(date); + let keys = merged + .iter() + .map(|row| (row.timestamp.time().to_string(), row.symbol.clone())) + .collect::>(); + assert_eq!( + keys, + vec![ + ("09:30:00".to_string(), "000001.SZ".to_string()), + ("09:30:00".to_string(), "000002.SZ".to_string()), + ("09:31:00".to_string(), "000001.SZ".to_string()), + ("09:31:00".to_string(), "000002.SZ".to_string()), + ] + ); + assert_eq!(run_data.remove_execution_quotes_on_date(date), 4); + assert_eq!(run_data.execution_quote_count(), 0); + } + #[test] fn baseline_selection_uses_structured_instrument_dates_and_status_only() { let date = NaiveDate::parse_from_str("2025-01-02", "%Y-%m-%d").unwrap(); diff --git a/crates/fidc-core/src/engine.rs b/crates/fidc-core/src/engine.rs index 1b7af7d..877572a 100644 --- a/crates/fidc-core/src/engine.rs +++ b/crates/fidc-core/src/engine.rs @@ -578,6 +578,9 @@ where if self.execution_quote_request_cache.contains(&request_key) { return false; } + if start_time.is_none() && end_time.is_none() { + return true; + } if start_time.is_some() && end_time.is_none() { return !has_execution_quote_near_start_time( &self.data, @@ -1666,6 +1669,7 @@ where F: FnMut(&BacktestDayProgress), { let mut portfolio = PortfolioState::new(self.config.initial_cash); + self.subscriptions = self.strategy.initial_subscriptions(); let scheduler_calendar = self.data.calendar().clone(); let scheduler = Scheduler::new(&scheduler_calendar); let calendar_dates = self @@ -2410,6 +2414,15 @@ where )?; if should_run_minute_events(&schedule_rules, &self.subscriptions) { + if self.execution_quote_loader.is_some() && !self.subscriptions.is_empty() { + let mut minute_symbols = self.subscriptions.clone(); + self.load_missing_execution_quotes( + execution_date, + None, + None, + &mut minute_symbols, + )?; + } let filter_by_subscription = !self.subscriptions.is_empty(); let minute_quotes = self .data @@ -2419,8 +2432,17 @@ where !filter_by_subscription || self.subscriptions.contains("e.symbol) }) .collect::>(); - for quote in minute_quotes { - let minute_time = quote.timestamp.time(); + let mut minute_cursor = 0usize; + while minute_cursor < minute_quotes.len() { + let minute_timestamp = minute_quotes[minute_cursor].timestamp; + let minute_time = minute_timestamp.time(); + let mut minute_end = minute_cursor + 1; + while minute_end < minute_quotes.len() + && minute_quotes[minute_end].timestamp == minute_timestamp + { + minute_end += 1; + } + let minute_group = &minute_quotes[minute_cursor..minute_end]; let minute_open_orders = self.open_order_views(); publish_phase_event( &mut self.strategy, @@ -2437,7 +2459,7 @@ where &mut process_events, execution_date, ProcessEventKind::PreMinute, - format!("minute:{}:{}:pre", quote.symbol, quote.timestamp), + format!("minute:{minute_timestamp}:pre"), )?; let mut minute_decision = collect_scheduled_decisions( &mut self.strategy, @@ -2459,25 +2481,27 @@ where 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(), - }, - "e, - )?); + for quote in minute_group { + 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(minute_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, @@ -2493,7 +2517,7 @@ where &mut process_events, execution_date, ProcessEventKind::Minute, - format!("minute:{}:{}", quote.symbol, quote.timestamp), + format!("minute:{minute_timestamp}"), )?; self.apply_strategy_directives( execution_date, @@ -2559,9 +2583,11 @@ where &mut process_events, execution_date, ProcessEventKind::PostMinute, - format!("minute:{}:{}:post", quote.symbol, quote.timestamp), + format!("minute:{minute_timestamp}:post"), )?; + minute_cursor = minute_end; } + self.data.remove_execution_quotes_on_date(execution_date); } portfolio.update_prices_with_options( diff --git a/crates/fidc-core/src/platform_expr_strategy.rs b/crates/fidc-core/src/platform_expr_strategy.rs index 1085e49..315e19b 100644 --- a/crates/fidc-core/src/platform_expr_strategy.rs +++ b/crates/fidc-core/src/platform_expr_strategy.rs @@ -30,6 +30,7 @@ use crate::strategy::{ #[derive(Debug, Clone, PartialEq, Eq)] pub enum PlatformScheduleFrequency { + Daily, Weekly { weekday: Option, tradingday: Option, @@ -198,6 +199,9 @@ impl SelectionRiskDeferral { impl PlatformRebalanceSchedule { fn as_schedule_rule(&self, stage: ScheduleStage) -> ScheduleRule { let rule = match self.frequency { + PlatformScheduleFrequency::Daily => { + ScheduleRule::daily("platform_periodic_rebalance", stage) + } PlatformScheduleFrequency::Weekly { weekday: Some(weekday), .. @@ -328,6 +332,7 @@ pub enum PlatformTradeAction { pub enum PlatformExplicitActionStage { OpenAuction, OnDay, + Minute, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -402,6 +407,7 @@ pub struct PlatformExprStrategyConfig { pub explicit_action_stage: PlatformExplicitActionStage, pub explicit_action_schedule: Option, pub subscription_guard_required: bool, + pub initial_subscriptions: BTreeSet, pub explicit_actions: Vec, } @@ -472,6 +478,7 @@ impl PlatformExprStrategyConfig { explicit_action_stage: PlatformExplicitActionStage::OnDay, explicit_action_schedule: None, subscription_guard_required: false, + initial_subscriptions: BTreeSet::new(), explicit_actions: Vec::new(), } } @@ -8436,6 +8443,7 @@ impl PlatformExprStrategy { match self.config.explicit_action_stage { PlatformExplicitActionStage::OpenAuction => "open_auction", PlatformExplicitActionStage::OnDay => "on_day", + PlatformExplicitActionStage::Minute => "minute", } )]; diagnostics.extend(action_diagnostics); @@ -8458,6 +8466,7 @@ impl PlatformExprStrategy { let stage = match self.config.explicit_action_stage { PlatformExplicitActionStage::OpenAuction => ScheduleStage::OpenAuction, PlatformExplicitActionStage::OnDay => ScheduleStage::OnDay, + PlatformExplicitActionStage::Minute => ScheduleStage::Minute, }; self.config .explicit_action_schedule @@ -10044,6 +10053,35 @@ impl Strategy for PlatformExprStrategy { self.config.strategy_name.as_str() } + fn initial_subscriptions(&self) -> BTreeSet { + self.config.initial_subscriptions.clone() + } + + fn schedule_rules(&self) -> Vec { + if self.config.explicit_action_stage != PlatformExplicitActionStage::Minute { + return Vec::new(); + } + self.config + .explicit_action_schedule + .as_ref() + .map(|schedule| schedule.as_schedule_rule(ScheduleStage::Minute)) + .into_iter() + .collect() + } + + fn on_scheduled( + &mut self, + ctx: &StrategyContext<'_>, + _rule: &ScheduleRule, + ) -> Result { + if self.config.explicit_action_stage == PlatformExplicitActionStage::Minute + && !self.config.explicit_actions.is_empty() + { + return self.explicit_action_decision(ctx); + } + Ok(StrategyDecision::default()) + } + fn decision_quote_times(&self) -> Vec { let mut times = BTreeSet::new(); if self.uses_intraday_execution_quotes() { diff --git a/crates/fidc-core/src/platform_strategy_spec.rs b/crates/fidc-core/src/platform_strategy_spec.rs index 426ca71..40eded8 100644 --- a/crates/fidc-core/src/platform_strategy_spec.rs +++ b/crates/fidc-core/src/platform_strategy_spec.rs @@ -756,6 +756,8 @@ pub struct StrategyExpressionTradingConfig { #[serde(default)] pub subscription_guard_required: Option, #[serde(default)] + pub subscriptions: Vec, + #[serde(default)] pub actions: Vec, } @@ -1855,9 +1857,16 @@ pub fn platform_expr_config_from_spec( if let Some(required) = trading.subscription_guard_required { cfg.subscription_guard_required = required; } + cfg.initial_subscriptions = trading + .subscriptions + .iter() + .map(|symbol| symbol.trim().to_ascii_uppercase()) + .filter(|symbol| !symbol.is_empty()) + .collect(); if let Some(stage) = trading.stage.as_deref().map(str::trim) { cfg.explicit_action_stage = match stage.to_ascii_lowercase().as_str() { "open_auction" | "open-auction" => PlatformExplicitActionStage::OpenAuction, + "minute" | "on_minute" | "on-minute" => PlatformExplicitActionStage::Minute, _ => PlatformExplicitActionStage::OnDay, }; } @@ -2017,6 +2026,10 @@ fn parse_platform_rebalance_schedule( let frequency = schedule.frequency.as_deref()?.trim().to_ascii_lowercase(); let time_rule = parse_schedule_time_rule(schedule); match frequency.as_str() { + "daily" => Some(PlatformRebalanceSchedule { + frequency: PlatformScheduleFrequency::Daily, + time_rule, + }), "weekly" => Some(PlatformRebalanceSchedule { frequency: PlatformScheduleFrequency::Weekly { weekday: schedule.weekday, @@ -2502,6 +2515,48 @@ mod tests { ); } + #[test] + fn parses_minute_stage_schedule_and_initial_subscriptions() { + let spec = serde_json::json!({ + "strategyId": "minute_runtime_strategy", + "signalSymbol": "000300.SH", + "benchmark": {"instrumentId": "000300.SH"}, + "runtimeExpressions": { + "selection": {"limitExpr": "1"}, + "trading": { + "stage": "minute", + "subscriptions": ["000001.sz", "000002.SZ"], + "schedule": {"frequency": "daily", "time": "10:18"}, + "actions": [ + { + "kind": "target_percent", + "symbol": "000001.SZ", + "amountExpr": "0.5", + "reason": "minute_target" + } + ] + } + } + }); + + let cfg = platform_expr_config_from_value("", "", &spec).expect("minute config"); + assert_eq!( + cfg.explicit_action_stage, + PlatformExplicitActionStage::Minute + ); + assert_eq!( + cfg.initial_subscriptions, + BTreeSet::from(["000001.SZ".to_string(), "000002.SZ".to_string()]) + ); + let schedule = cfg.explicit_action_schedule.expect("minute schedule"); + assert_eq!(schedule.frequency, PlatformScheduleFrequency::Daily); + assert_eq!( + schedule.time_rule, + Some(ScheduleTimeRule::physical_time(10, 18)) + ); + assert_eq!(cfg.explicit_actions.len(), 1); + } + #[test] fn runtime_expression_parser_does_not_inherit_microcap_template_defaults() { let spec = serde_json::json!({ @@ -3200,7 +3255,13 @@ mod tests { let cfg = platform_expr_config_from_value("", "", &spec).expect("config"); - assert_eq!(cfg.rebalance_schedule, None); + assert_eq!( + cfg.rebalance_schedule, + Some(PlatformRebalanceSchedule { + frequency: PlatformScheduleFrequency::Daily, + time_rule: Some(ScheduleTimeRule::MinuteOfDay(9 * 60 + 33)), + }) + ); assert_eq!( cfg.intraday_execution_time, Some(NaiveTime::from_hms_opt(9, 33, 0).unwrap()) diff --git a/crates/fidc-core/src/strategy.rs b/crates/fidc-core/src/strategy.rs index 0369015..c8d0021 100644 --- a/crates/fidc-core/src/strategy.rs +++ b/crates/fidc-core/src/strategy.rs @@ -19,6 +19,9 @@ use crate::universe::{DynamicMarketCapBandSelector, SelectionContext, UniverseSe pub trait Strategy { fn name(&self) -> &str; + fn initial_subscriptions(&self) -> BTreeSet { + BTreeSet::new() + } fn management_fee( &mut self, _ctx: &StrategyContext<'_>, diff --git a/crates/fidc-core/tests/engine_hooks.rs b/crates/fidc-core/tests/engine_hooks.rs index d720913..6161346 100644 --- a/crates/fidc-core/tests/engine_hooks.rs +++ b/crates/fidc-core/tests/engine_hooks.rs @@ -1,18 +1,19 @@ use std::cell::RefCell; use std::collections::{BTreeMap, BTreeSet}; use std::rc::Rc; +use std::sync::{Arc, Mutex}; use chrono::{NaiveDate, NaiveDateTime}; use fidc_core::{ BacktestConfig, BacktestEngine, BacktestProcessMod, BacktestProcessModLoader, BenchmarkSnapshot, BrokerSimulator, CandidateEligibility, ChinaAShareCostModel, - ChinaEquityRuleHooks, DailyFactorSnapshot, DailyMarketSnapshot, DataSet, FuturesAccountState, - FuturesCommissionType, FuturesContractSpec, FuturesDirection, FuturesOrderIntent, - FuturesTradingParameter, FuturesValidationConfig, Instrument, IntradayExecutionQuote, - IntradayOrderBookDepthLevel, MatchingType, OpenOrderView, OrderIntent, OrderSide, OrderStatus, - PlatformExprStrategy, PlatformExprStrategyConfig, PortfolioState, PriceField, ProcessEvent, - ProcessEventBus, ProcessEventKind, ScheduleRule, ScheduleStage, ScheduleTimeRule, Strategy, - StrategyContext, StrategyDecision, + ChinaEquityRuleHooks, DailyFactorSnapshot, DailyMarketSnapshot, DataSet, ExecutionQuoteRequest, + FuturesAccountState, FuturesCommissionType, FuturesContractSpec, FuturesDirection, + FuturesOrderIntent, FuturesTradingParameter, FuturesValidationConfig, Instrument, + IntradayExecutionQuote, IntradayOrderBookDepthLevel, MatchingType, OpenOrderView, OrderIntent, + OrderSide, OrderStatus, PlatformExprStrategy, PlatformExprStrategyConfig, PortfolioState, + PriceField, ProcessEvent, ProcessEventBus, ProcessEventKind, ScheduleRule, ScheduleStage, + ScheduleTimeRule, Strategy, StrategyContext, StrategyDecision, }; fn d(year: i32, month: u32, day: u32) -> NaiveDate { @@ -634,6 +635,8 @@ struct UniverseDirectiveStrategy { struct MinuteProbeStrategy { seen_ticks: Rc>>, + scheduled_count: Rc>, + subscribe_symbols: BTreeSet, ordered: bool, } @@ -809,6 +812,22 @@ impl Strategy for MinuteProbeStrategy { "minute-probe" } + fn schedule_rules(&self) -> Vec { + vec![ + ScheduleRule::daily("minute_barrier", ScheduleStage::Minute) + .with_time_rule(ScheduleTimeRule::physical_time(10, 18)), + ] + } + + fn on_scheduled( + &mut self, + _ctx: &StrategyContext<'_>, + _rule: &ScheduleRule, + ) -> Result { + *self.scheduled_count.borrow_mut() += 1; + Ok(StrategyDecision::default()) + } + fn on_day( &mut self, _ctx: &StrategyContext<'_>, @@ -818,7 +837,7 @@ impl Strategy for MinuteProbeStrategy { target_weights: BTreeMap::new(), exit_symbols: BTreeSet::new(), order_intents: vec![OrderIntent::Subscribe { - symbols: BTreeSet::from(["000001.SZ".to_string()]), + symbols: self.subscribe_symbols.clone(), reason: "subscribe_minute_probe".to_string(), }], notes: Vec::new(), @@ -2011,6 +2030,19 @@ fn engine_runs_minute_hooks_and_executes_minute_orders() { amount_delta: 10_200.0, trading_phase: Some("continuous".to_string()), }, + IntradayExecutionQuote { + date, + symbol: "000002.SZ".to_string(), + timestamp: dt(2025, 1, 2, 10, 18, 0), + last_price: 20.4, + bid1: 20.3, + ask1: 20.4, + bid1_volume: 1_000, + ask1_volume: 1_000, + volume_delta: 1_000, + amount_delta: 20_400.0, + trading_phase: Some("continuous".to_string()), + }, IntradayExecutionQuote { date, symbol: "000001.SZ".to_string(), @@ -2029,8 +2061,11 @@ fn engine_runs_minute_hooks_and_executes_minute_orders() { .expect("dataset"); let seen_ticks = Rc::new(RefCell::new(Vec::new())); + let scheduled_count = Rc::new(RefCell::new(0usize)); let strategy = MinuteProbeStrategy { seen_ticks: seen_ticks.clone(), + scheduled_count: scheduled_count.clone(), + subscribe_symbols: BTreeSet::from(["000001.SZ".to_string(), "000002.SZ".to_string()]), ordered: false, }; let broker = BrokerSimulator::new_with_execution_price( @@ -2038,6 +2073,8 @@ fn engine_runs_minute_hooks_and_executes_minute_orders() { ChinaEquityRuleHooks::default(), PriceField::Last, ); + let loader_requests = Arc::new(Mutex::new(Vec::::new())); + let loader_requests_for_callback = Arc::clone(&loader_requests); let mut engine = BacktestEngine::new( data, strategy, @@ -2050,7 +2087,11 @@ fn engine_runs_minute_hooks_and_executes_minute_orders() { decision_lag_trading_days: 0, execution_price_field: PriceField::Last, }, - ); + ) + .with_execution_quote_loader(move |request| { + loader_requests_for_callback.lock().unwrap().push(request); + Ok(Vec::new()) + }); let result = engine.run().expect("backtest run"); @@ -2058,9 +2099,19 @@ fn engine_runs_minute_hooks_and_executes_minute_orders() { seen_ticks.borrow().as_slice(), [ "000001.SZ:10:18:00:true:visible=10.20:previous=", + "000002.SZ:10:18:00:true:visible=20.40:previous=", "000001.SZ:10:19:00:true:visible=10.20,10.30:previous=10.20" ] ); + assert_eq!(*scheduled_count.borrow(), 1); + let loader_requests = loader_requests.lock().unwrap(); + assert_eq!(loader_requests.len(), 1); + assert_eq!(loader_requests[0].start_time, None); + assert_eq!(loader_requests[0].end_time, None); + assert_eq!( + loader_requests[0].symbols, + BTreeSet::from(["000001.SZ".to_string(), "000002.SZ".to_string()]) + ); assert_eq!(result.fills.len(), 1); assert_eq!(result.fills[0].reason, "minute_buy"); assert_eq!(result.fills[0].quantity, 100); @@ -2082,6 +2133,14 @@ fn engine_runs_minute_hooks_and_executes_minute_orders() { .iter() .any(|event| event.kind == ProcessEventKind::PostMinute) ); + assert_eq!( + result + .process_events + .iter() + .filter(|event| event.kind == ProcessEventKind::PreMinute) + .count(), + 2 + ); } #[test]