Compare commits

...

3 Commits

Author SHA1 Message Date
boris b281045df5 修复事件跨服务序列化生成空窗口 2026-09-09 23:54:22 +08:00
boris 35acb1c7e7 修复分钟事件轮动仅执行最后时点的问题 2026-09-09 23:52:35 +08:00
boris bbbd9cf3e0 统一日线事件上下文并接入完成分钟事件回测 2026-09-09 23:29:54 +08:00
9 changed files with 1043 additions and 56 deletions
@@ -0,0 +1,26 @@
use std::io::Read;
fn main() {
let mut input = String::new();
std::io::stdin().read_to_string(&mut input).unwrap();
let value: serde_json::Value = serde_json::from_str(&input).unwrap();
let spec: fidc_core::daily_patterns::PatternSpec =
serde_json::from_value(value["spec"].clone()).unwrap();
let bars: Vec<fidc_core::session_events::MinuteBar> =
serde_json::from_value(value["bars"].clone()).unwrap();
let result = fidc_core::session_events::evaluate(
&spec.validate().unwrap(),
value["symbol"].as_str().unwrap(),
&bars,
serde_json::from_value(value["decision_at"].clone()).unwrap(),
);
match result {
Ok(row) => println!(
"{}",
serde_json::json!({"contract":fidc_core::session_events::CONTRACT,"row":row,"read_only":true,"source_evidence_verified":false})
),
Err(error) => {
eprintln!("{error}");
std::process::exit(1);
}
}
}
+57 -36
View File
@@ -10,6 +10,7 @@ pub const CONTRACT: &str = "fidc_daily_ohlcv_pattern_v1";
pub fn catalog() -> Value { pub fn catalog() -> Value {
json!({"contract":CONTRACT,"templates":{ json!({"contract":CONTRACT,"templates":{
"expression":{"label":"指标与事件条件","parameters":{"history_window":[300,2,3000]},"stages":["selection","buy","sell","position_management"],"method":"冻结历史窗口与表达式;预热不足或未定义值不产生信号。复用共享指标事件内核,不修改既有任务。"}, "expression":{"label":"指标与事件条件","parameters":{"history_window":[300,2,3000]},"stages":["selection","buy","sell","position_management"],"method":"冻结历史窗口与表达式;预热不足或未定义值不产生信号。复用共享指标事件内核,不修改既有任务。"},
"session_event":{"label":"已完成分钟事件","parameters":{"opening_minutes":[30,1,120],"volume_window":[5,2,120],"volume_multiple":[3.0,1,20]},"stages":["selection","buy","sell"],"method":"仅本交易日完整分钟OHLCVA,信号K线必须早于执行时点;不使用盘口快照伪造K线。"},
"strength":{"label":"趋势强势","parameters":{"momentum_window":[25,5,120],"fast_window":[20,2,60],"slow_window":[60,20,252]},"stages":["selection","buy"],"method":"收盘价>短均线>长均线,按区间动量排序;不是当日金叉。"}, "strength":{"label":"趋势强势","parameters":{"momentum_window":[25,5,120],"fast_window":[20,2,60],"slow_window":[60,20,252]},"stages":["selection","buy"],"method":"收盘价>短均线>长均线,按区间动量排序;不是当日金叉。"},
"breakout":{"label":"前高突破","parameters":{"high_window":[60,5,252],"volume_window":[10,2,60],"volume_multiple":[1.3,1,10],"max_upper_shadow":[0.1,0,1]},"stages":["selection","buy"],"method":"收盘突破此前N日最高价,量达到此前M日均量倍数,上影比例受限;参考窗口不含当日。"}, "breakout":{"label":"前高突破","parameters":{"high_window":[60,5,252],"volume_window":[10,2,60],"volume_multiple":[1.3,1,10],"max_upper_shadow":[0.1,0,1]},"stages":["selection","buy"],"method":"收盘突破此前N日最高价,量达到此前M日均量倍数,上影比例受限;参考窗口不含当日。"},
"volume_spike":{"label":"放量上涨","parameters":{"volume_window":[5,2,60],"volume_multiple":[3.0,1,10]},"stages":["selection","buy"],"method":"当日上涨且量达到此前N日最大量的指定倍数;不等同价格创新高。"}, "volume_spike":{"label":"放量上涨","parameters":{"volume_window":[5,2,60],"volume_multiple":[3.0,1,10]},"stages":["selection","buy"],"method":"当日上涨且量达到此前N日最大量的指定倍数;不等同价格创新高。"},
@@ -31,10 +32,27 @@ pub struct PatternSpec {
pub parameters: BTreeMap<String, Value>, pub parameters: BTreeMap<String, Value>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub expression: Option<crate::factor_events::Expr>, pub expression: Option<crate::factor_events::Expr>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub execution_context: Option<crate::pattern_context::ExecutionContext>,
#[serde(default,skip_serializing_if="Option::is_none")]
pub session_event:Option<String>,
} }
impl PatternSpec { impl PatternSpec {
pub fn validate(self) -> Result<Self, String> { pub fn validate(self) -> Result<Self, String> {
self.validate_with_context(&[]) if self.template=="session_event" {
if !self.session_event.as_deref().is_some_and(|id|crate::session_events::EVENTS.contains(&id)) || self.execution_context.is_some() {return Err("session_event_contract_invalid".into());}
} else if self.session_event.is_some() {return Err("unexpected_session_event_id".into());}
let allowed = if let Some(context) = &self.execution_context {
context.validate(self.expression.as_ref().ok_or("pattern_context_requires_expression")?)?;
crate::pattern_context::CONTEXT_FIELDS
} else { &[] };
let spec = self.validate_with_context(allowed)?;
if let Some(context) = &spec.execution_context {
if context.rank_universe.len().saturating_mul(spec.history_len()) > 2_000_000 {
return Err("pattern_rank_window_budget_exceeded: 完整截面不得截断".into());
}
}
Ok(spec)
} }
fn validate_with_context(mut self, context_fields: &[&str]) -> Result<Self, String> { fn validate_with_context(mut self, context_fields: &[&str]) -> Result<Self, String> {
if (self.template == "expression") != self.expression.is_some() { if (self.template == "expression") != self.expression.is_some() {
@@ -82,7 +100,7 @@ impl PatternSpec {
if number < bounds[1].as_f64().unwrap() || number > bounds[2].as_f64().unwrap() { if number < bounds[1].as_f64().unwrap() || number > bounds[2].as_f64().unwrap() {
return Err(format!("{key}超出允许范围")); return Err(format!("{key}超出允许范围"));
} }
if key.ends_with("window") || key.ends_with("lookback") || key == "anchor_lag" { if key.ends_with("window") || key.ends_with("lookback") || key == "anchor_lag" || key=="opening_minutes" {
if number.fract() != 0.0 { if number.fract() != 0.0 {
return Err(format!("{key}必须是整数")); return Err(format!("{key}必须是整数"));
} }
@@ -104,6 +122,7 @@ impl PatternSpec {
} }
pub fn history_len(&self) -> usize { pub fn history_len(&self) -> usize {
match self.template.as_str() { match self.template.as_str() {
"session_event"=>1,
"expression" => self.n("history_window"), "expression" => self.n("history_window"),
"strength" => self.n("slow_window").max(self.n("momentum_window") + 1), "strength" => self.n("slow_window").max(self.n("momentum_window") + 1),
"breakout" => self.n("high_window").max(self.n("volume_window")) + 1, "breakout" => self.n("high_window").max(self.n("volume_window")) + 1,
@@ -198,13 +217,14 @@ pub fn evaluate(
evaluate_with_context(spec, days, series, &BTreeMap::new(), false) evaluate_with_context(spec, days, series, &BTreeMap::new(), false)
} }
fn evaluate_with_context( pub(crate) fn evaluate_with_context(
spec: &PatternSpec, spec: &PatternSpec,
days: &[NaiveDate], days: &[NaiveDate],
series: &PatternSeries, series: &PatternSeries,
context: &BTreeMap<String, Vec<Option<f64>>>, context: &BTreeMap<String, Vec<Option<f64>>>,
numeric_output: bool, numeric_output: bool,
) -> Result<PatternResult, String> { ) -> Result<PatternResult, String> {
if spec.template=="session_event" {return Err("session_event_requires_completed_minute_endpoint".into());}
if days.len() != spec.history_len() || days.windows(2).any(|w| w[0] >= w[1]) { if days.len() != spec.history_len() || days.windows(2).any(|w| w[0] >= w[1]) {
return Err("pattern_calendar_incomplete: 需要完整、唯一且递增的真实交易日窗口".into()); return Err("pattern_calendar_incomplete: 需要完整、唯一且递增的真实交易日窗口".into());
} }
@@ -616,39 +636,38 @@ pub fn evaluate_dataset(
date: NaiveDate, date: NaiveDate,
symbol: &str, symbol: &str,
) -> Result<PatternResult, String> { ) -> Result<PatternResult, String> {
let context = crate::pattern_context::build_dataset_context(spec, data, date)?;
evaluate_dataset_context(spec, data, date, symbol, &context)
}
pub fn dataset_series(data: &DataSet, days: &[NaiveDate], symbol: &str) -> PatternSeries {
let bars = days.iter().filter_map(|&d| data.market(d, symbol).map(|b| PatternBar {
date:d, open:Some(b.open), high:Some(b.high), low:Some(b.low), close:Some(b.close),
volume:Some(b.volume as f64), prev_close:Some(b.prev_close),
amount:data.factor_numeric_value(d,symbol,"amount"),upper_limit:Some(b.upper_limit),
no_limit:data.factor_numeric_value(d,symbol,"no_limit").map(|v|v==1.0),
adjustment_factor_backward1:data.factor(d,symbol).and_then(|f|f.adjustment_factor_backward1),
paused:Some(b.paused), source_path:None,
})).collect();
PatternSeries{symbol:symbol.into(),name:data.instrument(symbol).map(|i|i.name.clone()),
listed_at:data.instrument(symbol).and_then(|i|i.listed_at),bars}
}
pub fn evaluate_dataset_context(
spec: &PatternSpec, data: &DataSet, date: NaiveDate, symbol: &str, context: &ResearchContext,
) -> Result<PatternResult,String> {
let days = data.calendar().trailing_days(date, spec.history_len()); let days = data.calendar().trailing_days(date, spec.history_len());
let bars = days let mut fields = context.common.clone();
.iter() fields.extend(context.by_symbol.get(symbol).cloned().unwrap_or_default());
.filter_map(|&d| { let outside = spec.execution_context.as_ref().is_some_and(|c| c.rank_expression.is_some() && !c.rank_universe.iter().any(|s|s==symbol));
data.market(d, symbol).map(|b| PatternBar { if outside {
date: d, for name in ["scope_rank","scope_percentile"] {fields.insert(name.into(),vec![None;days.len()]);}
open: Some(b.open), fields.insert("scope_size".into(),vec![Some(spec.execution_context.as_ref().unwrap().rank_universe.len() as f64);days.len()]);
high: Some(b.high), }
low: Some(b.low), let mut result = evaluate_with_context(spec,&days,&dataset_series(data,&days,symbol),&fields,false)?;
close: Some(b.close), if outside && result.score.is_none() { result.exclusion=Some(json!({"reason":"outside_frozen_rank_universe","symbol":symbol,"signal_date":date})); }
volume: Some(b.volume as f64), result.values["execution_context_latest"]=json!(fields.iter().map(|(k,v)|(k,v.last().copied().flatten())).collect::<BTreeMap<_,_>>());
prev_close: data.factor_numeric_value(d, symbol, "pre_close"), Ok(result)
amount: data.factor_numeric_value(d, symbol, "amount"),
upper_limit: Some(b.upper_limit),
no_limit: data.factor_numeric_value(d,symbol,"no_limit").map(|v|v==1.0),
adjustment_factor_backward1: data
.factor(d, symbol)
.and_then(|f| f.adjustment_factor_backward1),
paused: Some(b.paused),
source_path: None,
})
})
.collect();
evaluate(
spec,
&days,
&PatternSeries {
symbol: symbol.into(),
name: None,
listed_at: data.instrument(symbol).and_then(|i| i.listed_at),
bars,
},
)
} }
pub fn evaluate_batch( pub fn evaluate_batch(
@@ -677,7 +696,7 @@ pub fn evaluate_batch(
) )
} }
/// Research transport only. Strategy PatternSpec validation still rejects these fields. /// Values are supplied only by the verified research transport or dataset context builder.
#[derive(Debug, Clone, Default, Deserialize)] #[derive(Debug, Clone, Default, Deserialize)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
pub struct ResearchContext { pub struct ResearchContext {
@@ -868,6 +887,8 @@ mod tests {
template: template.into(), template: template.into(),
parameters: BTreeMap::new(), parameters: BTreeMap::new(),
expression: None, expression: None,
execution_context: None,
session_event: None,
} }
.validate() .validate()
.unwrap(); .unwrap();
+15
View File
@@ -491,6 +491,7 @@ pub struct DataSetSnapshotComponents {
pub benchmarks: Vec<BenchmarkSnapshot>, pub benchmarks: Vec<BenchmarkSnapshot>,
pub corporate_actions: Vec<CorporateAction>, pub corporate_actions: Vec<CorporateAction>,
pub execution_quotes: Vec<IntradayExecutionQuote>, pub execution_quotes: Vec<IntradayExecutionQuote>,
pub completed_minute_bars: Vec<crate::session_events::MinuteBar>,
} }
#[derive(Debug, Clone, Serialize)] #[derive(Debug, Clone, Serialize)]
@@ -1418,6 +1419,7 @@ pub struct DataSet {
eligible_universe_by_date: Arc<OnceLock<BTreeMap<NaiveDate, Vec<EligibleUniverseSnapshot>>>>, eligible_universe_by_date: Arc<OnceLock<BTreeMap<NaiveDate, Vec<EligibleUniverseSnapshot>>>>,
benchmark_code: String, benchmark_code: String,
futures_params_by_symbol: Arc<HashMap<String, Vec<FuturesTradingParameter>>>, futures_params_by_symbol: Arc<HashMap<String, Vec<FuturesTradingParameter>>>,
completed_minute_bars: Arc<BTreeMap<(NaiveDate,String),Vec<crate::session_events::MinuteBar>>>,
} }
struct DailySymbolRows<'a, T> { struct DailySymbolRows<'a, T> {
@@ -1954,6 +1956,7 @@ impl DataSet {
eligible_universe_by_date: Arc::new(OnceLock::new()), eligible_universe_by_date: Arc::new(OnceLock::new()),
benchmark_code, benchmark_code,
futures_params_by_symbol: Arc::new(futures_params_by_symbol), futures_params_by_symbol: Arc::new(futures_params_by_symbol),
completed_minute_bars: Arc::new(BTreeMap::new()),
}) })
} }
@@ -2627,9 +2630,21 @@ impl DataSet {
benchmarks, benchmarks,
corporate_actions, corporate_actions,
execution_quotes, execution_quotes,
completed_minute_bars:self.completed_minute_bars.values().flatten().cloned().collect(),
} }
} }
pub fn with_completed_minute_bars(mut self,bars:Vec<crate::session_events::MinuteBar>)->Result<Self,String> {
self.completed_minute_bars=crate::session_events::bar_store(bars)?;Ok(self)
}
pub fn with_shared_completed_minute_bars(mut self,bars:crate::session_events::BarStore)->Self {self.completed_minute_bars=bars;self}
pub fn completed_minute_bar_count(&self)->usize {self.completed_minute_bars.values().map(Vec::len).sum()}
pub fn completed_minute_bars_on(&self,date:NaiveDate,symbol:&str)->&[crate::session_events::MinuteBar] {
self.completed_minute_bars.get(&(date,symbol.into())).map(Vec::as_slice).unwrap_or(&[])
}
pub fn benchmark_series(&self) -> Vec<BenchmarkSnapshot> { pub fn benchmark_series(&self) -> Vec<BenchmarkSnapshot> {
self.benchmark_by_date.values().cloned().collect() self.benchmark_by_date.values().cloned().collect()
} }
+4 -1
View File
@@ -53,7 +53,7 @@ pub enum Expr {
Operator { Operator {
name: String, name: String,
args: Vec<Expr>, args: Vec<Expr>,
#[serde(default)] #[serde(default, skip_serializing_if = "Option::is_none")]
window: Option<usize>, window: Option<usize>,
}, },
} }
@@ -177,6 +177,9 @@ pub fn catalog() -> Value {
"unstable_period":format!("{:?}",f.unst_id), "production_eligible":false, "unstable_period":format!("{:?}",f.unst_id), "production_eligible":false,
})).collect(); })).collect();
json!({"contract":CONTRACT,"library":{"name":"TA-Lib native Rust","revision":TA_REV,"license":"BSD-3-Clause"}, json!({"contract":CONTRACT,"library":{"name":"TA-Lib native Rust","revision":TA_REV,"license":"BSD-3-Clause"},
"execution_context_contract":crate::pattern_context::CONTRACT,
"execution_context_fields":crate::pattern_context::CONTEXT_FIELDS,
"session_events":crate::session_events::EVENTS,"session_event_contract":crate::session_events::CONTRACT,
"indicators":indicators,"operators":OPERATORS,"cross_section_operators":crate::factor_cross_section::OPERATORS,"read_only":true,"live_routing":false, "indicators":indicators,"operators":OPERATORS,"cross_section_operators":crate::factor_cross_section::OPERATORS,"read_only":true,"live_routing":false,
"policies":{"null":"unknown_not_false","warmup":"null_until_full_history","recursive_seed":"frozen_input_start", "policies":{"null":"unknown_not_false","warmup":"null_until_full_history","recursive_seed":"frozen_input_start",
"breakout":"previous_window_excludes_current","boolean":"three_valued_logic","daily_execution":"next_completed_session", "breakout":"previous_window_excludes_current","boolean":"three_valued_logic","daily_execution":"next_completed_session",
+2
View File
@@ -3,6 +3,8 @@ pub mod calendar;
pub mod cost; pub mod cost;
pub mod data; pub mod data;
pub mod daily_patterns; pub mod daily_patterns;
pub mod pattern_context;
pub mod session_events;
pub mod factor_events; pub mod factor_events;
pub mod factor_cross_section; pub mod factor_cross_section;
pub mod engine; pub mod engine;
+427
View File
@@ -0,0 +1,427 @@
//! Explicit reference identities and frozen rank universes shared by all daily runtimes.
use crate::{
daily_patterns::{dataset_series, evaluate_with_context, PatternSpec, ResearchContext},
factor_events::{field_dependencies, Expr},
DataSet,
};
use chrono::NaiveDate;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::{BTreeMap, BTreeSet};
pub const CONTRACT: &str = "fidc_pattern_execution_context_v1";
pub const CONTEXT_FIELDS: &[&str] = &[
"index_open",
"index_high",
"index_low",
"index_close",
"scope_rank",
"scope_percentile",
"scope_size",
];
const STOCK_FIELDS: &[&str] = &[
"open",
"high",
"low",
"close",
"volume",
"raw_open",
"raw_high",
"raw_low",
"raw_close",
"prev_close",
"amount",
];
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ExecutionContext {
pub contract: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub benchmark: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub rank_expression: Option<Expr>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub rank_universe: Vec<String>,
}
fn valid_symbol(s: &str) -> bool {
let Some((code, market)) = s.split_once('.') else {
return false;
};
code.len() == 6
&& code.bytes().all(|c| c.is_ascii_digit())
&& matches!(market, "SH" | "SZ" | "BJ" | "CSI")
}
impl ExecutionContext {
pub fn fields(&self, expression: &Expr) -> BTreeSet<String> {
let mut fields = field_dependencies(expression);
if let Some(rank) = &self.rank_expression {
fields.extend(field_dependencies(rank));
}
fields
}
pub fn validate(&self, expression: &Expr) -> Result<(), String> {
if self.contract != CONTRACT {
return Err("pattern_context_contract_invalid".into());
}
let needed = field_dependencies(expression);
let ranked = needed.iter().any(|f| f.starts_with("scope_"));
if ranked != self.rank_expression.is_some() || !ranked && !self.rank_universe.is_empty() {
return Err("pattern_rank_expression_and_universe_required".into());
}
if ranked
&& (self.rank_universe.len() < 2
|| self.rank_universe.len() > 20_000
|| self.rank_universe.iter().any(|s| !valid_symbol(s))
|| self.rank_universe.iter().collect::<BTreeSet<_>>().len()
!= self.rank_universe.len())
{
return Err("pattern_rank_universe_invalid".into());
}
if let Some(rank) = &self.rank_expression {
let fields = field_dependencies(rank);
if fields
.iter()
.any(|f| !STOCK_FIELDS.contains(&f.as_str()) && !f.starts_with("index_"))
{
return Err("pattern_rank_expression_invalid_or_recursive".into());
}
}
let fields = self.fields(expression);
if fields
.iter()
.any(|f| !STOCK_FIELDS.contains(&f.as_str()) && !CONTEXT_FIELDS.contains(&f.as_str()))
{
return Err("pattern_context_unmapped_field".into());
}
let index = fields.iter().any(|f| f.starts_with("index_"));
if index != self.benchmark.is_some()
|| self
.benchmark
.as_ref()
.is_some_and(|s| !valid_symbol(s) || s.ends_with(".BJ"))
{
return Err("pattern_reference_index_required".into());
}
if !index && !ranked {
return Err("pattern_unused_context".into());
}
Ok(())
}
}
pub fn build_dataset_context(
spec: &PatternSpec,
data: &DataSet,
date: NaiveDate,
) -> Result<ResearchContext, String> {
let Some(config) = &spec.execution_context else {
return Ok(ResearchContext::default());
};
config.validate(
spec.expression
.as_ref()
.ok_or("pattern_context_requires_expression")?,
)?;
let days = data.calendar().trailing_days(date, spec.history_len());
if days.len() != spec.history_len() || days.last() != Some(&date) {
return Err("pattern_context_calendar_incomplete".into());
}
let needed = config.fields(spec.expression.as_ref().unwrap());
let mut context = ResearchContext::default();
if let Some(symbol) = &config.benchmark {
for name in needed.iter().filter(|f| f.starts_with("index_")) {
let values = days
.iter()
.map(|d| {
let value = if let Some(b) = data.market(*d, symbol) {
match name.as_str() {
"index_open" => Some(b.open),
"index_high" => Some(b.high),
"index_low" => Some(b.low),
"index_close" => Some(b.close),
_ => None,
}
} else if let Some(b) = data.benchmark(*d).filter(|b| &b.benchmark == symbol) {
match name.as_str() {
"index_open" => Some(b.open),
"index_close" => Some(b.close),
_ => None,
}
} else {
None
};
value
.filter(|v| v.is_finite() && *v > 0.0)
.map(Some)
.ok_or_else(|| format!("pattern_reference_missing: {symbol} {d} {name}"))
})
.collect::<Result<Vec<_>, _>>()?;
context.common.insert(name.clone(), values);
}
}
if let Some(expression) = &config.rank_expression {
let mut input = spec.clone();
input.execution_context = None;
input.expression = Some(expression.clone());
let mut values = BTreeMap::new();
for symbol in &config.rank_universe {
let row = evaluate_with_context(
&input,
&days,
&dataset_series(data, &days, symbol),
&context.common,
true,
)?;
if let Some(reason) = row.exclusion {
return Err(format!("pattern_rank_member_incomplete: {symbol} {reason}"));
}
values.insert(
symbol.clone(),
serde_json::from_value::<Vec<Option<f64>>>(
row.values["expression"]["values"].clone(),
)
.map_err(|e| e.to_string())?,
);
}
let ranks =
crate::factor_cross_section::rank_history(&days, &config.rank_universe, &values)?;
for symbol in &config.rank_universe {
let decode = |value: &Value| {
serde_json::from_value::<Vec<Option<f64>>>(value.clone()).map_err(|e| e.to_string())
};
context.by_symbol.insert(
symbol.clone(),
BTreeMap::from([
("scope_rank".into(), decode(&ranks["rank"][symbol])?),
(
"scope_percentile".into(),
decode(&ranks["percentile"][symbol])?,
),
(
"scope_size".into(),
vec![Some(config.rank_universe.len() as f64); days.len()],
),
]),
);
}
}
Ok(context)
}
pub fn specs_in_value(value: &Value) -> Result<Vec<PatternSpec>, String> {
let mut specs = Vec::new();
match value {
Value::String(text) => specs.extend(crate::daily_patterns::expression_specs(text)?),
Value::Array(items) => {
for v in items {
specs.extend(specs_in_value(v)?);
}
}
Value::Object(items) => {
for v in items.values() {
specs.extend(specs_in_value(v)?);
}
}
_ => {}
}
Ok(specs)
}
pub fn required_symbols(value: &Value) -> Result<(BTreeSet<String>, BTreeSet<String>), String> {
let (mut indices, mut stocks) = (BTreeSet::new(), BTreeSet::new());
for spec in specs_in_value(value)? {
if let Some(context) = spec.execution_context {
if let Some(index) = context.benchmark {
indices.insert(index);
}
stocks.extend(context.rank_universe);
}
}
Ok((indices, stocks))
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{BenchmarkSnapshot, DailyFactorSnapshot, DailyMarketSnapshot, Instrument};
use serde_json::json;
#[test]
fn normalized_rule_does_not_turn_an_omitted_window_into_explicit_null() {
let expression:Expr=serde_json::from_value(json!({"kind":"operator","name":"GT","args":[{"kind":"field","name":"close"},{"kind":"number","value":1}]})).unwrap();
assert!(serde_json::to_value(expression).unwrap().get("window").is_none());
}
fn data(future: bool, reference: bool) -> DataSet {
let mut days = vec![
NaiveDate::from_ymd_opt(2026, 9, 4).unwrap(),
NaiveDate::from_ymd_opt(2026, 9, 7).unwrap(),
NaiveDate::from_ymd_opt(2026, 9, 8).unwrap(),
];
if future {
days.push(NaiveDate::from_ymd_opt(2026, 9, 9).unwrap());
}
let symbols = vec!["000001.SZ", "000002.SZ", "000003.SZ"];
let mut instruments = symbols
.iter()
.map(|s| Instrument {
symbol: s.to_string(),
name: s.to_string(),
board: "SZ_MAIN".into(),
round_lot: 100,
listed_at: None,
delisted_at: None,
status: "active".into(),
})
.collect::<Vec<_>>();
if reference {
instruments.push(Instrument {
symbol: "399006.SZ".into(),
name: "reference".into(),
board: "INDEX".into(),
round_lot: 1,
listed_at: None,
delisted_at: None,
status: "active".into(),
});
}
let mut market = vec![];
let mut factors = vec![];
let mut benchmark = vec![];
for (i, d) in days.iter().enumerate() {
for (n, s) in symbols.iter().enumerate() {
let c = [
[10., 12., 11., 1000.],
[10., 11., 12., 1.],
[10., 10., 13., 1.],
][n][i];
market.push(DailyMarketSnapshot {
date: *d,
symbol: s.to_string(),
timestamp: None,
day_open: c,
open: c,
high: c,
low: c,
close: c,
last_price: c,
bid1: c,
ask1: c,
prev_close: 10.,
volume: 100000,
minute_volume: 0,
bid1_volume: 10000,
ask1_volume: 10000,
trading_phase: None,
paused: false,
upper_limit: c * 2.,
lower_limit: c / 2.,
price_tick: 0.01,
});
factors.push(DailyFactorSnapshot {
date: *d,
symbol: s.to_string(),
market_cap_bn: 1.,
free_float_cap_bn: 1.,
pe_ttm: 10.,
turnover_ratio: None,
effective_turnover_ratio: None,
adjustment_factor_backward1: Some(1.),
extra_factors: Default::default(),
});
}
if reference {
let mut row = market.last().unwrap().clone();
row.symbol = "399006.SZ".into();
row.open = 30.;
row.high = 30.;
row.low = 30.;
row.close = 30.;
market.push(row);
}
benchmark.push(BenchmarkSnapshot {
date: *d,
benchmark: "000300.SH".into(),
open: 4000.,
close: 4000.,
prev_close: 4000.,
volume: 1000,
});
}
DataSet::from_components(instruments, market, factors, vec![], benchmark).unwrap()
}
fn spec(rank: bool) -> PatternSpec {
let expression = if rank {
json!({"kind":"operator","name":"GT","args":[{"kind":"field","name":"scope_rank"},{"kind":"number","value":2}]})
} else {
json!({"kind":"operator","name":"LT","args":[{"kind":"field","name":"index_close"},{"kind":"number","value":100}]})
};
let context = if rank {
json!({"contract":CONTRACT,"rank_expression":{"kind":"operator","name":"PCT_CHANGE","window":1,"args":[{"kind":"field","name":"close"}]},"rank_universe":["000001.SZ","000002.SZ","000003.SZ"]})
} else {
json!({"contract":CONTRACT,"benchmark":"399006.SZ"})
};
serde_json::from_value::<PatternSpec>(json!({"template":"expression","parameters":{"history_window":3},"expression":expression,"execution_context":context})).unwrap().validate().unwrap()
}
#[test]
fn dataset_rank_is_full_scope_causal_and_equal_to_pure_cross_section() {
let spec = spec(true);
let date = NaiveDate::from_ymd_opt(2026, 9, 8).unwrap();
let original = build_dataset_context(&spec, &data(false, true), date).unwrap();
let future = build_dataset_context(&spec, &data(true, true), date).unwrap();
assert_eq!(original.by_symbol, future.by_symbol);
assert_eq!(original.by_symbol["000001.SZ"]["scope_rank"][2], Some(3.));
assert_eq!(original.by_symbol["000002.SZ"]["scope_rank"][2], Some(2.));
assert_eq!(original.by_symbol["000003.SZ"]["scope_rank"][2], Some(1.));
assert!(
crate::daily_patterns::evaluate_dataset(&spec, &data(false, true), date, "000001.SZ")
.unwrap()
.matched
);
let mut incomplete = data(false, true).snapshot_components();
incomplete.market.retain(|r| r.symbol != "000003.SZ");
let broken = DataSet::from_components(
incomplete.instruments,
incomplete.market,
incomplete.factors,
incomplete.candidates,
incomplete.benchmarks,
)
.unwrap();
assert!(build_dataset_context(&spec, &broken, date).is_err());
}
#[test]
fn reference_index_never_defaults_to_performance_benchmark() {
let spec = spec(false);
let date = NaiveDate::from_ymd_opt(2026, 9, 8).unwrap();
assert!(
crate::daily_patterns::evaluate_dataset(&spec, &data(false, true), date, "000001.SZ")
.unwrap()
.matched
);
assert!(build_dataset_context(&spec, &data(false, false), date)
.unwrap_err()
.contains("399006.SZ"));
}
#[test]
fn runtime_contract_rejects_missing_range_and_recursive_ranks() {
let mut missing = spec(true);
missing
.execution_context
.as_mut()
.unwrap()
.rank_universe
.clear();
assert!(missing.validate().is_err());
let mut recursive = spec(true);
recursive
.execution_context
.as_mut()
.unwrap()
.rank_expression = Some(Expr::Field {
name: "scope_rank".into(),
});
assert!(recursive.validate().is_err());
}
}
+96 -19
View File
@@ -672,6 +672,7 @@ pub struct PlatformExprStrategyConfig {
pub completed_session_factor_fields: BTreeSet<String>, pub completed_session_factor_fields: BTreeSet<String>,
pub candidate_symbols_by_date: BTreeMap<NaiveDate, BTreeSet<String>>, pub candidate_symbols_by_date: BTreeMap<NaiveDate, BTreeSet<String>>,
pub intraday_execution_time: Option<NaiveTime>, pub intraday_execution_time: Option<NaiveTime>,
pub session_event_times: Vec<NaiveTime>,
pub explicit_action_times: Vec<NaiveTime>, pub explicit_action_times: Vec<NaiveTime>,
pub delayed_limit_open_exit_enabled: bool, pub delayed_limit_open_exit_enabled: bool,
pub delayed_limit_open_exit_time: Option<NaiveTime>, pub delayed_limit_open_exit_time: Option<NaiveTime>,
@@ -753,6 +754,7 @@ impl PlatformExprStrategyConfig {
completed_session_factor_fields: BTreeSet::new(), completed_session_factor_fields: BTreeSet::new(),
candidate_symbols_by_date: BTreeMap::new(), candidate_symbols_by_date: BTreeMap::new(),
intraday_execution_time: None, intraday_execution_time: None,
session_event_times: Vec::new(),
explicit_action_times: Vec::new(), explicit_action_times: Vec::new(),
delayed_limit_open_exit_enabled: false, delayed_limit_open_exit_enabled: false,
delayed_limit_open_exit_time: None, delayed_limit_open_exit_time: None,
@@ -1287,7 +1289,7 @@ struct RuntimeHelperBinding {
#[derive(Clone)] #[derive(Clone)]
enum CompiledRuntimeHelperArgs { enum CompiledRuntimeHelperArgs {
DailyPattern { spec: crate::daily_patterns::PatternSpec }, DailyPattern { spec: crate::daily_patterns::PatternSpec, identity: String },
RollingMean { RollingMean {
field: String, field: String,
lookback: usize, lookback: usize,
@@ -1347,6 +1349,9 @@ enum RuntimeHelperResolution {
pub struct PlatformExprStrategy { pub struct PlatformExprStrategy {
pattern_results_date: RefCell<Option<NaiveDate>>, pattern_results_date: RefCell<Option<NaiveDate>>,
pattern_results: RefCell<BTreeMap<(NaiveDate, String, String), crate::daily_patterns::PatternResult>>, pattern_results: RefCell<BTreeMap<(NaiveDate, String, String), crate::daily_patterns::PatternResult>>,
pattern_contexts: RefCell<BTreeMap<String,crate::daily_patterns::ResearchContext>>,
pattern_specs: RefCell<BTreeMap<String,String>>,
pattern_frame_at:RefCell<Option<NaiveDateTime>>,
config: PlatformExprStrategyConfig, config: PlatformExprStrategyConfig,
engine: Engine, engine: Engine,
rebalance_day_counter: usize, rebalance_day_counter: usize,
@@ -1796,6 +1801,9 @@ impl PlatformExprStrategy {
stock_extra_factor_map_required, stock_extra_factor_map_required,
stock_text_factors_required, stock_text_factors_required,
pattern_results: RefCell::new(BTreeMap::new()), pattern_results: RefCell::new(BTreeMap::new()),
pattern_contexts: RefCell::new(BTreeMap::new()),
pattern_specs: RefCell::new(BTreeMap::new()),
pattern_frame_at:RefCell::new(None),
pattern_results_date: RefCell::new(None), pattern_results_date: RefCell::new(None),
stock_state_cache_date: RefCell::new(None), stock_state_cache_date: RefCell::new(None),
stock_state_cache_calendar_index: RefCell::new(None), stock_state_cache_calendar_index: RefCell::new(None),
@@ -5705,23 +5713,48 @@ impl PlatformExprStrategy {
args: &CompiledRuntimeHelperArgs, args: &CompiledRuntimeHelperArgs,
) -> Result<RuntimeHelperResolution, BacktestError> { ) -> Result<RuntimeHelperResolution, BacktestError> {
match args { match args {
CompiledRuntimeHelperArgs::DailyPattern { spec } => { CompiledRuntimeHelperArgs::DailyPattern { spec, identity } => {
if self.config.matching_type != MatchingType::NextBarOpen { if *self.pattern_results_date.borrow()!=Some(ctx.execution_date) {
self.pattern_results.borrow_mut().clear();self.pattern_contexts.borrow_mut().clear();self.pattern_specs.borrow_mut().clear();
*self.pattern_results_date.borrow_mut()=Some(ctx.execution_date);
}
if *self.pattern_frame_at.borrow()!=ctx.active_datetime {
self.pattern_results.borrow_mut().clear();*self.pattern_frame_at.borrow_mut()=ctx.active_datetime;
}
if spec.template=="session_event" {
if self.config.matching_type!=MatchingType::MinuteLast {return Err(BacktestError::Execution("session_event_requires_minute_last".into()));}
let active=ctx.active_datetime.ok_or_else(||BacktestError::Execution("session_event_requires_explicit_clock".into()))?;
let clock=active.time();
if !((NaiveTime::from_hms_opt(9,31,0).unwrap()<=clock&&clock<=NaiveTime::from_hms_opt(11,30,0).unwrap())||(NaiveTime::from_hms_opt(13,1,0).unwrap()<=clock&&clock<=NaiveTime::from_hms_opt(15,0,0).unwrap())) {
return Ok(if helper=="pattern_signal"{RuntimeHelperResolution::Boolean(false)}else{RuntimeHelperResolution::Number(0.)});
}
let stock=stock.ok_or_else(||BacktestError::Execution("session_event_requires_stock".into()))?;
let key=(active.date(),stock.symbol.to_string(),identity.clone());
if !self.pattern_results.borrow().contains_key(&key) {
let result=crate::session_events::evaluate(spec,&stock.symbol,ctx.data.completed_minute_bars_on(active.date(),&stock.symbol),active).map_err(BacktestError::Execution)?;
self.pattern_results.borrow_mut().insert(key.clone(),result);
self.pattern_specs.borrow_mut().insert(identity.clone(),serde_json::to_string(spec).unwrap());
}
let rows=self.pattern_results.borrow();let result=&rows[&key];
return if helper=="pattern_signal" {Ok(RuntimeHelperResolution::Boolean(result.matched))}else{result.score.map(RuntimeHelperResolution::Number).ok_or_else(||BacktestError::Execution("session_score_unknown".into()))};
}
if !matches!(self.config.matching_type,MatchingType::NextBarOpen|MatchingType::MinuteLast) {
return Err(BacktestError::Execution("daily_pattern_requires_next_bar_open: 完整日线形态只能在下一交易日执行".into())); return Err(BacktestError::Execution("daily_pattern_requires_next_bar_open: 完整日线形态只能在下一交易日执行".into()));
} }
let date = day.date.min(ctx.decision_date); let date = if self.config.matching_type==MatchingType::MinuteLast {ctx.data.previous_trading_date(ctx.execution_date,1).ok_or_else(||BacktestError::Execution("daily_pattern_previous_completed_date_missing".into()))?} else {day.date.min(ctx.decision_date)};
// Lagged replay retains the decision day's schedule label; execution is on a later session. // Lagged replay retains the decision day's schedule label; execution is on a later session.
if !ctx.is_lagged_execution() && ctx.active_datetime.is_some_and(|t| t.date() == date && t.time() < NaiveTime::from_hms_opt(16, 0, 0).unwrap()) { if !ctx.is_lagged_execution() && ctx.active_datetime.is_some_and(|t| t.date() == date && t.time() < NaiveTime::from_hms_opt(16, 0, 0).unwrap()) {
return Err(BacktestError::Execution(format!("daily_pattern_not_yet_visible: 不允许使用未完成的当日日线; decision_date={date}, execution_date={}, active_datetime={:?}",ctx.execution_date,ctx.active_datetime))); return Err(BacktestError::Execution(format!("daily_pattern_not_yet_visible: 不允许使用未完成的当日日线; decision_date={date}, execution_date={}, active_datetime={:?}",ctx.execution_date,ctx.active_datetime)));
} }
let stock = stock.ok_or_else(|| BacktestError::Execution("pattern_signal requires stock context".into()))?; let stock = stock.ok_or_else(|| BacktestError::Execution("pattern_signal requires stock context".into()))?;
let key = (date, stock.symbol.to_string(), serde_json::to_string(spec).unwrap()); let key = (date, stock.symbol.to_string(), identity.clone());
if *self.pattern_results_date.borrow() != Some(date) {
self.pattern_results.borrow_mut().clear();
*self.pattern_results_date.borrow_mut() = Some(date);
}
if !self.pattern_results.borrow().contains_key(&key) { if !self.pattern_results.borrow().contains_key(&key) {
let result = crate::daily_patterns::evaluate_dataset(spec,ctx.data,date,&stock.symbol).map_err(BacktestError::Execution)?; if !self.pattern_contexts.borrow().contains_key(&key.2) {
let context=crate::pattern_context::build_dataset_context(spec,ctx.data,date).map_err(BacktestError::Execution)?;
self.pattern_contexts.borrow_mut().insert(key.2.clone(),context);
self.pattern_specs.borrow_mut().insert(key.2.clone(),serde_json::to_string(spec).unwrap());
}
let result = crate::daily_patterns::evaluate_dataset_context(spec,ctx.data,date,&stock.symbol,&self.pattern_contexts.borrow()[&key.2]).map_err(BacktestError::Execution)?;
self.pattern_results.borrow_mut().insert(key.clone(),result); self.pattern_results.borrow_mut().insert(key.clone(),result);
} }
let results = self.pattern_results.borrow(); let results = self.pattern_results.borrow();
@@ -6983,7 +7016,10 @@ impl PlatformExprStrategy {
"pattern_signal" | "pattern_score" if args.len() == 1 => { "pattern_signal" | "pattern_score" if args.len() == 1 => {
let text: String = serde_json::from_str(&args[0]).ok()?; let text: String = serde_json::from_str(&args[0]).ok()?;
let spec: crate::daily_patterns::PatternSpec = serde_json::from_str(&text).ok()?; let spec: crate::daily_patterns::PatternSpec = serde_json::from_str(&text).ok()?;
Some(CompiledRuntimeHelperArgs::DailyPattern { spec: spec.validate().ok()? }) let spec=spec.validate().ok()?;
use sha2::Digest;
let identity=format!("{:x}",sha2::Sha256::digest(serde_json::to_vec(&spec).ok()?));
Some(CompiledRuntimeHelperArgs::DailyPattern { spec, identity })
} }
"rolling_mean" | "sma" | "ma" => { "rolling_mean" | "sma" | "ma" => {
let (field, lookback) = field_lookback()?; let (field, lookback) = field_lookback()?;
@@ -12260,13 +12296,15 @@ impl Strategy for PlatformExprStrategy {
.is_some() .is_some()
&& self.config.rotation_enabled && self.config.rotation_enabled
{ {
rules.push( let schedule=self.config.rebalance_schedule.as_ref().expect("checked timed rebalance schedule");
self.config if self.config.session_event_times.is_empty() {
.rebalance_schedule rules.push(schedule.as_schedule_rule(ScheduleStage::OnDay));
.as_ref() } else {
.expect("checked timed rebalance schedule") for time in &self.config.session_event_times {
.as_schedule_rule(ScheduleStage::OnDay), let mut timed=schedule.clone();timed.time_rule=Some(ScheduleTimeRule::physical_time(time.hour(),time.minute()));
); rules.push(timed.as_schedule_rule(ScheduleStage::OnDay));
}
}
} }
if self.config.explicit_actions.is_empty() { if self.config.explicit_actions.is_empty() {
return rules; return rules;
@@ -12329,6 +12367,7 @@ impl Strategy for PlatformExprStrategy {
fn decision_quote_times(&self) -> Vec<NaiveTime> { fn decision_quote_times(&self) -> Vec<NaiveTime> {
let mut times = BTreeSet::new(); let mut times = BTreeSet::new();
times.extend(self.config.session_event_times.iter().copied());
if self.uses_intraday_execution_quotes() { if self.uses_intraday_execution_quotes() {
if self.config.explicit_action_times.is_empty() { if self.config.explicit_action_times.is_empty() {
times.insert(self.intraday_execution_start_time()); times.insert(self.intraday_execution_start_time());
@@ -12423,12 +12462,28 @@ impl PlatformExprStrategy {
} }
fn append_pattern_diagnostics(&self, decision: &mut StrategyDecision) { fn append_pattern_diagnostics(&self, decision: &mut StrategyDecision) {
let mut contexts=BTreeMap::<String,serde_json::Value>::new();
for ((date, symbol, spec), result) in self.pattern_results.borrow().iter() { for ((date, symbol, spec), result) in self.pattern_results.borrow().iter() {
if result.values["session_contract"]==crate::session_events::CONTRACT {
let record=serde_json::json!({"event":"session_event_decision","date":date,"symbol":symbol,"spec_sha256":spec,"matched":result.matched,"signal_bar_end":result.values["signal_bar_end"],"decision_at":result.values["decision_at"],"exclusion":result.exclusion}).to_string();
if !decision.diagnostics.contains(&record){decision.diagnostics.push(record)}
continue;
}
if result.values["execution_context_latest"].as_object().is_some_and(|v|!v.is_empty()) {
let group=contexts.entry(spec.clone()).or_insert_with(||serde_json::json!({"event":"daily_pattern_context_decisions","date":date,"spec_sha256":spec,"evaluated":0,"matched":0,"excluded":0,"sample_limit":20,"samples":[]}));
group["evaluated"]=serde_json::json!(group["evaluated"].as_u64().unwrap()+1);
group["matched"]=serde_json::json!(group["matched"].as_u64().unwrap()+u64::from(result.matched));
group["excluded"]=serde_json::json!(group["excluded"].as_u64().unwrap()+u64::from(result.exclusion.is_some()));
let samples=group["samples"].as_array_mut().unwrap();
if samples.len()<20 {samples.push(serde_json::json!({"symbol":symbol,"matched":result.matched,"context":result.values["execution_context_latest"],"exclusion":result.exclusion}));}
continue;
}
if let Some(evidence) = &result.exclusion { if let Some(evidence) = &result.exclusion {
let record = serde_json::json!({"event":"daily_pattern_excluded","date":date,"symbol":symbol,"spec":spec,"evidence":evidence}).to_string(); let record = serde_json::json!({"event":"daily_pattern_excluded","date":date,"symbol":symbol,"spec":self.pattern_specs.borrow().get(spec),"spec_sha256":spec,"evidence":evidence}).to_string();
if !decision.diagnostics.contains(&record) { decision.diagnostics.push(record); } if !decision.diagnostics.contains(&record) { decision.diagnostics.push(record); }
} }
} }
for value in contexts.values() {let record=value.to_string();if !decision.diagnostics.contains(&record){decision.diagnostics.push(record);}}
} }
fn compute_day_decision(&mut self, ctx: &StrategyContext<'_>) -> Result<StrategyDecision, BacktestError> { fn compute_day_decision(&mut self, ctx: &StrategyContext<'_>) -> Result<StrategyDecision, BacktestError> {
@@ -14299,6 +14354,28 @@ mod tests {
assert_eq!(strategy.selection_quote_usage, StockFilterQuoteUsage::DailyOnly); assert_eq!(strategy.selection_quote_usage, StockFilterQuoteUsage::DailyOnly);
} }
#[test]
fn session_event_uses_previous_bar_and_recomputes_at_each_minute_without_becoming_a_quote() {
let date = d(2026,9,8); let symbol = "000001.SZ";
let bars = (0..=33).map(|i| {
let timestamp = date.and_hms_opt(9,30,0).unwrap() + chrono::Duration::minutes(i);
let close = if i==31 {11.} else {10.};
crate::session_events::MinuteBar {symbol:symbol.into(),timestamp,available_at:timestamp,open:close,high:close,low:close,close,volume:100.,amount:close*100.}
}).collect();
let data = single_symbol_platform_data(&[date],symbol).with_completed_minute_bars(bars).unwrap();
let portfolio=PortfolioState::new(100_000.); let subscriptions=BTreeSet::new();
let mut ctx=StrategyContext {execution_date:date,decision_date:date,decision_index:0,data:&data,portfolio:&portfolio,futures_account:None,open_orders:&[],dynamic_universe:None,subscriptions:&subscriptions,process_events:&[],active_process_event:None,active_datetime:None,order_events:&[],fills:&[]};
let mut config=PlatformExprStrategyConfig::generic();config.signal_symbol=symbol.into();config.matching_type=MatchingType::MinuteLast;
let strategy=PlatformExprStrategy::new(config);
let expression=r#"pattern_signal("{\"template\":\"session_event\",\"session_event\":\"OPENING_RANGE_BREAKOUT_UP\",\"parameters\":{}}")"#;
let day=strategy.day_state(&ctx,date).unwrap();let stock=strategy.stock_state(&ctx,date,symbol).unwrap();
for (minute,expected) in [(1,false),(2,true),(3,false)] {
ctx.active_datetime=date.and_hms_opt(10,minute,0);
assert_eq!(strategy.eval_bool(&ctx,expression,&day,Some(&stock),None).unwrap(),expected);
}
assert!(data.snapshot_components().execution_quotes.is_empty());
}
#[test] #[test]
fn buy_filter_uses_active_schedule_time_instead_of_first_configured_time() { fn buy_filter_uses_active_schedule_time_instead_of_first_configured_time() {
let date = d(2025, 1, 2); let date = d(2025, 1, 2);
@@ -2538,6 +2538,10 @@ pub fn platform_expr_config_from_spec(
cfg.benchmark_symbol = normalize_symbol(&cfg.benchmark_symbol, None); cfg.benchmark_symbol = normalize_symbol(&cfg.benchmark_symbol, None);
} }
let trade_times = spec_trade_times(spec); let trade_times = spec_trade_times(spec);
if crate::pattern_context::specs_in_value(&serde_json::to_value(spec).map_err(|e|e.to_string())?)?.iter().any(|p|p.template=="session_event") {
if trade_times.is_empty() {return Err("session_event_requires_explicit_trade_times".into());}
cfg.session_event_times=trade_times.clone();
}
let explicit_trading_schedule = spec let explicit_trading_schedule = spec
.runtime_expressions .runtime_expressions
.as_ref() .as_ref()
@@ -4458,6 +4462,20 @@ mod tests {
assert_eq!(cfg.delayed_limit_open_exit_time, None); assert_eq!(cfg.delayed_limit_open_exit_time, None);
} }
#[test]
fn session_rotation_keeps_every_declared_clock_not_only_the_last_one() {
use crate::Strategy;
let literal=serde_json::to_string(&serde_json::json!({"template":"session_event","session_event":"INTRADAY_VOLUME_SPIKE","parameters":{}}).to_string()).unwrap();
let mut spec=serde_json::json!({"rebalance":{"tradeTimes":["09:35","10:40","14:59"]},"runtimeExpressions":{"schedule":{"frequency":"daily","time":"14:59"},"trading":{"rotationEnabled":true,"buyFilterExpr":format!("pattern_signal({literal})")}},"execution":{"matchingType":"minute_last"}});
let config=platform_expr_config_from_value("session","000300.SH",&spec).unwrap();
assert_eq!(config.session_event_times.len(),3);
let strategy=crate::PlatformExprStrategy::new(config);
assert_eq!(strategy.schedule_rules().len(),3);
assert_eq!(strategy.decision_quote_times().len(),3);
spec["rebalance"]["tradeTimes"]=serde_json::json!([]);
assert!(platform_expr_config_from_value("session","000300.SH",&spec).unwrap_err().to_string().contains("explicit_trade_times"));
}
#[test] #[test]
fn explicit_trading_schedule_overrides_rebalance_trade_times() { fn explicit_trading_schedule_overrides_rebalance_trade_times() {
let spec = serde_json::json!({ let spec = serde_json::json!({
+398
View File
@@ -0,0 +1,398 @@
//! Completed, same-session minute events. These bars never become execution quotes.
use crate::{
daily_patterns::{PatternResult, PatternSpec},
factor_events::{Expr, Frame},
};
use chrono::{FixedOffset, NaiveDateTime, NaiveTime, TimeZone, Timelike};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::collections::BTreeMap;
use std::sync::Arc;
pub const CONTRACT: &str = "fidc_completed_session_events_v1";
pub const EVENTS: &[&str] = &[
"PRICE_CROSS_VWAP_UP",
"PRICE_CROSS_VWAP_DOWN",
"INTRADAY_HIGH_BREAKOUT",
"INTRADAY_LOW_BREAKDOWN",
"OPENING_RANGE_BREAKOUT_UP",
"OPENING_RANGE_BREAKOUT_DOWN",
"INTRADAY_VOLUME_SPIKE",
"MORNING_HIGH_BREAKOUT",
"MORNING_LOW_BREAKDOWN",
"AFTERNOON_MOMENTUM_UP",
"AFTERNOON_MOMENTUM_DOWN",
"LATE_SESSION_STRENGTH",
"LATE_SESSION_WEAKNESS",
];
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MinuteBar {
pub symbol: String,
pub timestamp: NaiveDateTime,
pub available_at: NaiveDateTime,
pub open: f64,
pub high: f64,
pub low: f64,
pub close: f64,
pub volume: f64,
pub amount: f64,
}
pub type BarStore = Arc<BTreeMap<(chrono::NaiveDate, String), Vec<MinuteBar>>>;
pub fn bar_store(bars: Vec<MinuteBar>) -> Result<BarStore, String> {
let mut groups = BTreeMap::<(chrono::NaiveDate, String), Vec<MinuteBar>>::new();
for bar in bars {
groups
.entry((bar.timestamp.date(), bar.symbol.clone()))
.or_default()
.push(bar);
}
for rows in groups.values_mut() {
rows.sort_by_key(|r| r.timestamp);
if rows
.windows(2)
.any(|pair| pair[0].timestamp == pair[1].timestamp)
{
return Err("duplicate_completed_minute_bar".into());
}
}
Ok(Arc::new(groups))
}
fn f(name: &str) -> Expr {
Expr::Field { name: name.into() }
}
fn n(value: f64) -> Expr {
Expr::Number { value }
}
fn op(name: &str, args: Vec<Expr>, window: Option<usize>) -> Expr {
Expr::Operator {
name: name.into(),
args,
window,
}
}
fn time(minutes: u32) -> NaiveTime {
NaiveTime::from_hms_opt(minutes / 60, minutes % 60, 0).unwrap()
}
pub fn is_regular_label(t: NaiveTime) -> bool {
t.second() == 0 && (time(570) <= t && t <= time(690) || time(780) < t && t <= time(900))
}
pub fn expression(event: &str, p: &BTreeMap<String, Value>) -> Result<Expr, String> {
let cross = |up: bool, a: Expr, b: Expr| {
op(
if up { "CROSS_ABOVE" } else { "CROSS_BELOW" },
vec![a, b],
None,
)
};
Ok(match event {
"PRICE_CROSS_VWAP_UP" => cross(true, f("close"), f("session_vwap")),
"PRICE_CROSS_VWAP_DOWN" => cross(false, f("close"), f("session_vwap")),
"INTRADAY_HIGH_BREAKOUT" => op(
"GT",
vec![
f("close"),
op("LAG", vec![op("CUMMAX", vec![f("high")], None)], Some(1)),
],
None,
),
"INTRADAY_LOW_BREAKDOWN" => op(
"LT",
vec![
f("close"),
op("LAG", vec![op("CUMMIN", vec![f("low")], None)], Some(1)),
],
None,
),
"OPENING_RANGE_BREAKOUT_UP" => cross(true, f("close"), f("opening_high")),
"OPENING_RANGE_BREAKOUT_DOWN" => cross(false, f("close"), f("opening_low")),
"MORNING_HIGH_BREAKOUT" => cross(true, f("close"), f("morning_high")),
"MORNING_LOW_BREAKDOWN" => cross(false, f("close"), f("morning_low")),
"AFTERNOON_MOMENTUM_UP" => cross(true, f("afternoon_return"), n(0.)),
"AFTERNOON_MOMENTUM_DOWN" => cross(false, f("afternoon_return"), n(0.)),
"LATE_SESSION_STRENGTH" => cross(true, f("late_return"), n(0.)),
"LATE_SESSION_WEAKNESS" => cross(false, f("late_return"), n(0.)),
"INTRADAY_VOLUME_SPIKE" => op(
"GTE",
vec![
f("volume"),
op(
"MUL",
vec![
op(
"LAG",
vec![op(
"ROLLING_MEAN",
vec![f("volume")],
Some(p["volume_window"].as_u64().unwrap() as usize),
)],
Some(1),
),
n(p["volume_multiple"].as_f64().unwrap()),
],
None,
),
],
None,
),
_ => return Err("session_event_not_registered".into()),
})
}
pub fn evaluate(
spec: &PatternSpec,
symbol: &str,
bars: &[MinuteBar],
decision: NaiveDateTime,
) -> Result<PatternResult, String> {
let mut result = PatternResult {
symbol: symbol.into(),
name: None,
matched: false,
score: None,
checks: vec![],
values: json!({}),
anchor: Value::Null,
exclusion: None,
};
if bars.is_empty() {
return Err(format!(
"session_source_missing: {symbol} {}",
decision.date()
));
}
let visible = bars
.iter()
.filter(|b| {
b.timestamp.date() == decision.date()
&& b.timestamp < decision
&& b.available_at <= decision
})
.collect::<Vec<_>>();
if visible.is_empty() {
result.exclusion = Some(json!({"reason":"session_before_first_completed_bar"}));
return Ok(result);
}
let last = visible.last().unwrap().timestamp;
let expected = (570..=690)
.chain(781..=900)
.map(|m| decision.date().and_time(time(m)))
.filter(|t| *t < decision)
.last();
if expected != Some(last) {
return Err(format!(
"session_latest_bar_missing: {symbol} expected={expected:?} actual={last}"
));
}
let mut indexed = BTreeMap::new();
for b in &visible {
if b.symbol != symbol
|| !is_regular_label(b.timestamp.time())
|| b.available_at < b.timestamp
|| [b.open, b.high, b.low, b.close, b.volume, b.amount]
.iter()
.any(|v| !v.is_finite())
|| b.low <= 0.
|| b.open <= 0.
|| b.close <= 0.
|| b.high < b.open.max(b.close)
|| b.low > b.open.min(b.close)
|| b.volume < 0.
|| b.amount < 0.
|| indexed.insert(b.timestamp, b).is_some()
{
return Err(format!("session_bar_invalid: {symbol} {}", b.timestamp));
}
}
for minute in (571..=690).chain(781..=900) {
let stamp = decision.date().and_time(time(minute));
if stamp <= last && !indexed.contains_key(&stamp) {
return Err(format!(
"session_bar_gap: {symbol} {stamp}; no filling or calendar compression"
));
}
}
let opening_end = time(570 + spec.n("opening_minutes") as u32);
let (mut volume, mut amount) = (0., 0.);
let (mut opening_high, mut opening_low) = (f64::NEG_INFINITY, f64::INFINITY);
let (mut morning_high, mut morning_low) = (f64::NEG_INFINITY, f64::INFINITY);
let (mut morning_close, mut late_close) = (None, None);
let mut fields: BTreeMap<String, Vec<Option<f64>>> = [
"open",
"high",
"low",
"close",
"volume",
"amount",
"session_vwap",
"opening_high",
"opening_low",
"morning_high",
"morning_low",
"afternoon_return",
"late_return",
]
.into_iter()
.map(|s| (s.into(), vec![]))
.collect();
let mut timestamps = vec![];
let mut available_at = vec![];
let zone = FixedOffset::east_opt(8 * 3600).unwrap();
for b in indexed.values() {
let t = b.timestamp.time();
volume += b.volume;
amount += b.amount;
if t <= opening_end {
opening_high = opening_high.max(b.high);
opening_low = opening_low.min(b.low);
}
if t <= time(690) {
morning_high = morning_high.max(b.high);
morning_low = morning_low.min(b.low);
}
if t == time(690) {
morning_close = Some(b.close);
}
if t == time(870) {
late_close = Some(b.close);
}
for (name, value) in [
("open", Some(b.open)),
("high", Some(b.high)),
("low", Some(b.low)),
("close", Some(b.close)),
("volume", Some(b.volume)),
("amount", Some(b.amount)),
("session_vwap", (volume > 0.).then_some(amount / volume)),
("opening_high", (t >= opening_end).then_some(opening_high)),
("opening_low", (t >= opening_end).then_some(opening_low)),
("morning_high", (t >= time(690)).then_some(morning_high)),
("morning_low", (t >= time(690)).then_some(morning_low)),
("afternoon_return", morning_close.map(|v| b.close / v - 1.)),
("late_return", late_close.map(|v| b.close / v - 1.)),
] {
fields.get_mut(name).unwrap().push(value);
}
timestamps.push(zone.from_local_datetime(&b.timestamp).single().unwrap());
available_at.push(zone.from_local_datetime(&b.available_at).single().unwrap());
}
let frame = Frame {
symbol: symbol.into(),
frequency: "1m".into(),
decision_at: zone.from_local_datetime(&decision).single().unwrap(),
timestamps,
available_at,
fields,
};
let event = spec
.session_event
.as_deref()
.ok_or("session_event_id_required")?;
let values = crate::factor_events::evaluate(&expression(event, &spec.parameters)?, &frame)?;
let latest = values.values.last().copied().flatten();
result.score = latest;
result.matched = latest == Some(1.);
result.values = json!({"session_event":event,"session_contract":CONTRACT,"expression":values,"signal_bar_end":last,"decision_at":decision,"bars":visible.len(),"bar_times":frame.timestamps.iter().map(|t|t.format("%Y-%m-%dT%H:%M:%S").to_string()).collect::<Vec<_>>(),"close":visible.last().unwrap().close,"session_return":visible.last().unwrap().close/visible.first().unwrap().open-1.,"price_policy":"same_session_raw_ohlcv"});
if latest.is_none() {
result.exclusion = Some(json!({"reason":"session_warmup_or_undefined"}));
} else {
result.checks.push(json!({"label":"分钟事件","actual":latest,"operator":"==","threshold":1,"passed":result.matched}));
}
Ok(result)
}
#[cfg(test)]
mod tests {
use super::*;
fn spec(event: &str) -> PatternSpec {
serde_json::from_value::<PatternSpec>(
json!({"template":"session_event","session_event":event,"parameters":{}}),
)
.unwrap()
.validate()
.unwrap()
}
fn bars() -> Vec<MinuteBar> {
let date = chrono::NaiveDate::from_ymd_opt(2026, 9, 8).unwrap();
(570..=690)
.chain(781..=900)
.enumerate()
.map(|(i, m)| {
let timestamp = date.and_time(time(m));
let price = 100. + (i % 17) as f64 / 10.;
let volume = if i % 39 == 0 { 1000. } else { 100. };
MinuteBar {
symbol: "300395.SZ".into(),
timestamp,
available_at: timestamp,
open: price,
high: price + 0.1,
low: price - 0.1,
close: price,
volume,
amount: volume * price,
}
})
.collect()
}
#[test]
fn all_thirteen_events_return_native_boolean_series() {
let bars = bars();
let decision = "2026-09-08T15:00:01".parse().unwrap();
for event in EVENTS {
let value = evaluate(&spec(event), "300395.SZ", &bars, decision).unwrap();
assert!(value.score.is_some(), "{event}");
assert_eq!(value.values["expression"]["value_type"], "boolean");
}
}
#[test]
fn decision_uses_the_previous_completed_label_and_future_prices_do_not_rewrite() {
let mut bars = bars();
let decision = "2026-09-08T10:02:00".parse().unwrap();
for event in EVENTS {
let before = evaluate(&spec(event), "300395.SZ", &bars, decision).unwrap();
for bar in &mut bars {
if bar.timestamp >= decision {
bar.open = 1000.;
bar.close = 1000.;
bar.high = 1001.;
bar.low = 999.;
}
}
let after = evaluate(&spec(event), "300395.SZ", &bars, decision).unwrap();
assert_eq!(before.values, after.values);
assert_eq!(after.values["signal_bar_end"], "2026-09-08T10:01:00");
}
}
#[test]
fn gaps_and_stale_last_bars_do_not_become_false_or_repeated_signals() {
let mut values = bars();
let decision = "2026-09-08T10:02:00".parse().unwrap();
values.retain(|r| r.timestamp.time() != time(600));
assert!(evaluate(&spec(EVENTS[0]), "300395.SZ", &values, decision)
.unwrap_err()
.contains("session_bar_gap"));
let stale = bars()
.into_iter()
.filter(|r| r.timestamp.time() < time(601))
.collect::<Vec<_>>();
assert!(evaluate(&spec(EVENTS[0]), "300395.SZ", &stale, decision)
.unwrap_err()
.contains("latest_bar_missing"));
}
#[test]
fn opening_range_is_unavailable_before_the_range_has_completed() {
let value = evaluate(
&spec("OPENING_RANGE_BREAKOUT_UP"),
"300395.SZ",
&bars(),
"2026-09-08T09:59:01".parse().unwrap(),
)
.unwrap();
assert_eq!(value.score, None);
assert!(!value.matched);
}
}