diff --git a/README.md b/README.md index ff931ef..99562cc 100644 --- a/README.md +++ b/README.md @@ -80,6 +80,8 @@ 任意数据库指标和自定义因子通过 `factor("field")`、`factor_value("field", lookback)`、`rolling_mean("field", n)`、`sma("close", n)` 等函数读取。未预计算的均线窗口可在回测中按已有历史数据实时计算。 +Source Lake 日线成交量保留原始可用性合同：源 `volume=null` 与真实 `volume=0` 含义不同。依赖成交量的 rolling 窗口只要包含源空值就返回缺失，不得把空值补成 0；停牌日明确提供的 0 成交量仍是合法观测。该合同随 runner 快照版本冻结，旧快照不能跨版本复用。 + ## 内置微盘策略 `OmniMicroCapStrategy` 是平台内置的微盘轮动策略，用于 demo、性能验证和策略迁移基线： diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index b13de4e..80c6a1a 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -10,6 +10,8 @@ use crate::futures::FuturesTradingParameter; use crate::instrument::Instrument; use crate::risk_control::{ChinaAShareRiskControl, FidcRiskControlConfig}; +const SOURCE_DAILY_VOLUME_AVAILABLE_FIELD: &str = "source_daily_volume_available"; + mod date_format { use chrono::NaiveDate; use serde::{self, Deserialize, Deserializer, Serializer}; @@ -1039,6 +1041,8 @@ pub struct DataSet { adjusted_close_series_by_symbol: Arc>>>, benchmark_series_cache: BenchmarkPriceSeries, eligible_universe_by_date: Arc>>>, + source_daily_volume_contract_symbols: HashSet, + source_daily_volume_missing_dates_by_symbol: HashMap>, benchmark_code: String, futures_params_by_symbol: HashMap>, } @@ -1186,10 +1190,30 @@ impl DataSet { ) -> Result { let benchmark_code = collect_benchmark_code(&benchmarks)?; let calendar = TradingCalendar::new(benchmarks.iter().map(|item| item.date).collect()); - let factors = normalize_factor_snapshots(factors) - .into_iter() - .map(Arc::new) - .collect::>(); + let factors = normalize_factor_snapshots(factors); + let mut source_daily_volume_contract_symbols = HashSet::new(); + let mut source_daily_volume_missing_dates_by_symbol = + HashMap::>::new(); + for snapshot in &factors { + let Some(available) = snapshot + .extra_factors + .get(SOURCE_DAILY_VOLUME_AVAILABLE_FIELD) + else { + continue; + }; + source_daily_volume_contract_symbols.insert(snapshot.symbol.clone()); + if *available < 0.5 { + source_daily_volume_missing_dates_by_symbol + .entry(snapshot.symbol.clone()) + .or_default() + .push(snapshot.date); + } + } + for dates in source_daily_volume_missing_dates_by_symbol.values_mut() { + dates.sort_unstable(); + dates.dedup(); + } + let factors = factors.into_iter().map(Arc::new).collect::>(); let candidates = candidates.into_iter().map(Arc::new).collect::>(); let instruments = instruments @@ -1250,6 +1274,8 @@ impl DataSet { adjusted_close_series_by_symbol: Arc::new(RwLock::new(HashMap::new())), benchmark_series_cache, eligible_universe_by_date: Arc::new(OnceLock::new()), + source_daily_volume_contract_symbols, + source_daily_volume_missing_dates_by_symbol, benchmark_code, futures_params_by_symbol, }) @@ -2400,9 +2426,14 @@ impl DataSet { "close" | "prev_close" | "stock_close" | "price" => self .market_series(symbol) .and_then(|series| series.decision_close_moving_average(date, lookback)), - "volume" | "stock_volume" => self - .market_series(symbol) - .and_then(|series| series.decision_volume_moving_average(date, lookback)), + "volume" | "stock_volume" => { + if !self.source_daily_volume_window_available(date, symbol, lookback, false) { + None + } else { + self.market_series(symbol) + .and_then(|series| series.decision_volume_moving_average(date, lookback)) + } + } "day_open" | "dayopen" => { self.market_moving_average(date, symbol, lookback, PriceField::DayOpen) } @@ -2426,10 +2457,17 @@ impl DataSet { "close" | "prev_close" | "stock_close" | "price" => self .adjusted_close_series(symbol) .and_then(|series| series.current_moving_average(date, lookback)), - "volume" | "stock_volume" => self - .market_series(symbol) - .and_then(|series| series.current_volume_moving_average(date, lookback)) - .or_else(|| self.factor_moving_average(date, symbol, "daily_volume", lookback)), + "volume" | "stock_volume" => { + if !self.source_daily_volume_window_available(date, symbol, lookback, true) { + None + } else { + self.market_series(symbol) + .and_then(|series| series.current_volume_moving_average(date, lookback)) + .or_else(|| { + self.factor_moving_average(date, symbol, "daily_volume", lookback) + }) + } + } "day_open" | "dayopen" => { self.market_moving_average(date, symbol, lookback, PriceField::DayOpen) } @@ -2462,10 +2500,15 @@ impl DataSet { .market_series(symbol) .and_then(|series| series.decision_prev_close_values(date, lookback)) .unwrap_or_default(), - "volume" | "stock_volume" => self - .market_series(symbol) - .and_then(|series| series.decision_volume_values(date, lookback)) - .unwrap_or_default(), + "volume" | "stock_volume" => { + if !self.source_daily_volume_window_available(date, symbol, lookback, false) { + Vec::new() + } else { + self.market_series(symbol) + .and_then(|series| series.decision_volume_values(date, lookback)) + .unwrap_or_default() + } + } "day_open" | "dayopen" => self .market_series(symbol) .map(|series| series.trailing_values(date, lookback, PriceField::DayOpen)) @@ -2489,11 +2532,60 @@ impl DataSet { field: &str, lookback: usize, ) -> Vec { + let field = normalize_field(field); + if matches!(field.as_str(), "volume" | "stock_volume") + && !self.source_daily_volume_window_available(date, symbol, lookback, true) + { + return Vec::new(); + } self.market_series(symbol) - .map(|series| series.trailing_numeric_values(date, lookback, field, true)) + .map(|series| series.trailing_numeric_values(date, lookback, &field, true)) .unwrap_or_default() } + fn source_daily_volume_window_available( + &self, + date: NaiveDate, + symbol: &str, + lookback: usize, + include_now: bool, + ) -> bool { + if !self.source_daily_volume_contract_symbols.contains(symbol) { + return true; + } + if lookback == 0 { + return false; + } + let end = if include_now && self.calendar.index_of(date).is_some() { + date + } else { + let Some(previous) = self.calendar.previous_trading_date(date, 1) else { + return false; + }; + previous + }; + let dates = self.calendar.trailing_days(end, lookback); + if dates.len() != lookback { + return false; + } + let Some(series) = self.market_series(symbol) else { + return false; + }; + if dates + .iter() + .any(|trading_day| series.dates.binary_search(trading_day).is_err()) + { + return false; + } + let Some(missing_dates) = self.source_daily_volume_missing_dates_by_symbol.get(symbol) + else { + return true; + }; + !dates + .iter() + .any(|trading_day| missing_dates.binary_search(trading_day).is_ok()) + } + pub fn factor_numeric_values( &self, date: NaiveDate, @@ -3407,6 +3499,113 @@ mod tests { ); } + fn volume_contract_data(availability: Option<[f64; 3]>) -> DataSet { + let dates = [ + NaiveDate::parse_from_str("2025-01-02", "%Y-%m-%d").unwrap(), + NaiveDate::parse_from_str("2025-01-03", "%Y-%m-%d").unwrap(), + NaiveDate::parse_from_str("2025-01-06", "%Y-%m-%d").unwrap(), + ]; + let volumes = [100_u64, 0, 300]; + DataSet::from_components( + vec![Instrument { + symbol: "000001.SZ".to_string(), + name: "000001.SZ".to_string(), + board: "SZ".to_string(), + round_lot: 100, + listed_at: Some(dates[0]), + delisted_at: None, + status: "active".to_string(), + }], + dates + .iter() + .zip(volumes) + .map(|(date, volume)| { + market_row(&date.format("%Y-%m-%d").to_string(), 10.0, volume) + }) + .collect(), + dates + .iter() + .enumerate() + .map(|(index, date)| { + let mut extra_factors = BTreeMap::new(); + if let Some(values) = availability { + extra_factors.insert( + SOURCE_DAILY_VOLUME_AVAILABLE_FIELD.to_string(), + values[index], + ); + if values[index] >= 0.5 { + extra_factors.insert("daily_volume".to_string(), volumes[index] as f64); + } + } + DailyFactorSnapshot { + date: *date, + symbol: "000001.SZ".to_string(), + market_cap_bn: 10.0, + free_float_cap_bn: 8.0, + pe_ttm: 10.0, + turnover_ratio: None, + effective_turnover_ratio: None, + extra_factors, + } + }) + .collect(), + Vec::new(), + dates + .iter() + .map(|date| BenchmarkSnapshot { + date: *date, + benchmark: "000852.SH".to_string(), + open: 100.0, + close: 100.0, + prev_close: 100.0, + volume: 1_000_000, + }) + .collect(), + ) + .expect("volume contract dataset") + } + + #[test] + fn source_volume_contract_rejects_windows_containing_missing_values() { + let data = volume_contract_data(Some([1.0, 0.0, 1.0])); + let date = NaiveDate::parse_from_str("2025-01-06", "%Y-%m-%d").unwrap(); + + assert_eq!( + data.market_current_numeric_moving_average(date, "000001.SZ", "volume", 3), + None + ); + assert!( + data.market_current_numeric_values(date, "000001.SZ", "volume", 3) + .is_empty() + ); + assert_eq!( + data.market_decision_numeric_moving_average(date, "000001.SZ", "volume", 2), + None + ); + assert!( + data.market_decision_numeric_values(date, "000001.SZ", "volume", 2) + .is_empty() + ); + } + + #[test] + fn source_volume_contract_keeps_valid_zero_volume_and_legacy_data() { + let date = NaiveDate::parse_from_str("2025-01-06", "%Y-%m-%d").unwrap(); + for data in [ + volume_contract_data(Some([1.0, 1.0, 1.0])), + volume_contract_data(None), + ] { + assert_eq!( + data.market_current_numeric_moving_average(date, "000001.SZ", "volume", 3), + Some(133.333333) + ); + assert_eq!( + data.market_current_numeric_values(date, "000001.SZ", "volume", 3), + vec![100.0, 0.0, 300.0] + ); + } + } + #[test] fn decision_close_average_ignores_current_day_close() { let mut current = market_row("2025-01-06", 12.0, 10_000);