//! 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>>; pub fn bar_store(bars: Vec) -> Result { let mut groups = BTreeMap::<(chrono::NaiveDate, String), Vec>::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, window: Option) -> 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) -> Result { 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 { 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::>(); 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>> = [ "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::>(),"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::( json!({"template":"session_event","session_event":event,"parameters":{}}), ) .unwrap() .validate() .unwrap() } fn bars() -> Vec { 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::>(); 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); } }