From 279d6a100fdb17a6f4049129f348ec0be3acfe70 Mon Sep 17 00:00:00 2001 From: boris Date: Sun, 23 Aug 2026 13:09:01 +0800 Subject: [PATCH] =?UTF-8?q?=E7=BB=9F=E4=B8=80=E6=88=90=E4=BA=A4=E9=87=8F?= =?UTF-8?q?=E6=BB=9A=E5=8A=A8=E6=9C=89=E6=95=88=E6=A0=B7=E6=9C=AC=E5=8F=A3?= =?UTF-8?q?=E5=BE=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/fidc-core/src/data.rs | 200 +++++++++++++++++------------------ 1 file changed, 98 insertions(+), 102 deletions(-) diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index 138def1..6ec03a2 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -11,7 +11,6 @@ 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; @@ -486,7 +485,8 @@ struct SymbolPriceSeries { close_prefix: Vec, prev_close_prefix: Vec, last_prefix: Vec, - volume_prefix: Vec, + valid_volume_sum_prefix: Vec, + valid_volume_count_prefix: Vec, } #[derive(Debug, Clone)] @@ -630,11 +630,21 @@ impl SymbolPriceSeries { let close_prefix = prefix_sums(&closes); let prev_close_prefix = prefix_sums(&prev_closes); let last_prefix = prefix_sums(&last_prices); - let volume_values = volumes - .iter() - .map(|value| *value as f64) - .collect::>(); - let volume_prefix = prefix_sums(&volume_values); + let mut valid_volume_sum_prefix = Vec::with_capacity(volumes.len() + 1); + let mut valid_volume_count_prefix = Vec::with_capacity(volumes.len() + 1); + valid_volume_sum_prefix.push(0.0); + valid_volume_count_prefix.push(0); + for volume in &volumes { + let valid = *volume > 0; + valid_volume_sum_prefix.push( + valid_volume_sum_prefix.last().copied().unwrap_or_default() + + if valid { *volume as f64 } else { 0.0 }, + ); + valid_volume_count_prefix.push( + valid_volume_count_prefix.last().copied().unwrap_or_default() + + usize::from(valid), + ); + } Self { symbol, @@ -662,7 +672,8 @@ impl SymbolPriceSeries { close_prefix, prev_close_prefix, last_prefix, - volume_prefix, + valid_volume_sum_prefix, + valid_volume_count_prefix, } } @@ -783,49 +794,62 @@ impl SymbolPriceSeries { } fn decision_volume_moving_average(&self, date: NaiveDate, lookback: usize) -> Option { - if lookback == 0 { - return None; - } let end = self.previous_completed_end_index(date)?; - if end < lookback { - return None; - } - let start = end - lookback; - let sum = self.volume_prefix[end] - self.volume_prefix[start]; - Some(sum / lookback as f64) + self.valid_volume_window(end, lookback) + .map(|(start, end)| { + normalize_rolling_factor( + (self.valid_volume_sum_prefix[end] - self.valid_volume_sum_prefix[start]) + / lookback as f64, + 12, + ) + }) } fn current_volume_moving_average(&self, date: NaiveDate, lookback: usize) -> Option { - if lookback == 0 { - return None; - } let end = self.end_index(date)?; - if end < lookback { - return None; - } - let start = end - lookback; - let sum = self.volume_prefix[end] - self.volume_prefix[start]; - if !sum.is_finite() { - return None; - } - Some(normalize_rolling_factor(sum / lookback as f64, 6)) + self.valid_volume_window(end, lookback) + .map(|(start, end)| { + normalize_rolling_factor( + (self.valid_volume_sum_prefix[end] - self.valid_volume_sum_prefix[start]) + / lookback as f64, + 12, + ) + }) } fn decision_volume_values(&self, date: NaiveDate, lookback: usize) -> Option> { - if lookback == 0 { - return None; - } let end = self.previous_completed_end_index(date)?; - if end < lookback { + self.valid_volume_values(end, lookback) + } + + fn current_volume_values(&self, date: NaiveDate, lookback: usize) -> Option> { + let end = self.end_index(date)?; + self.valid_volume_values(end, lookback) + } + + fn valid_volume_window(&self, end: usize, lookback: usize) -> Option<(usize, usize)> { + if lookback == 0 || end > self.volumes.len() { return None; } - let start = end - lookback; - Some( - self.volumes[start..end] - .iter() - .map(|value| *value as f64) - .collect(), - ) + let valid_count = *self.valid_volume_count_prefix.get(end)?; + if valid_count < lookback { + return None; + } + let target_count = valid_count - lookback; + let start = self.valid_volume_count_prefix[..=end] + .partition_point(|count| *count <= target_count) + .saturating_sub(1); + Some((start, end)) + } + + fn valid_volume_values(&self, end: usize, lookback: usize) -> Option> { + let (start, end) = self.valid_volume_window(end, lookback)?; + let values = self.volumes[start..end] + .iter() + .filter(|value| **value > 0) + .map(|value| *value as f64) + .collect::>(); + (values.len() == lookback).then_some(values) } fn end_index(&self, date: NaiveDate) -> Option { @@ -1057,8 +1081,6 @@ 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>, } @@ -1207,28 +1229,6 @@ impl DataSet { let benchmark_code = collect_benchmark_code(&benchmarks)?; let calendar = TradingCalendar::new(benchmarks.iter().map(|item| item.date).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::>(); @@ -1311,8 +1311,6 @@ impl DataSet { adjusted_close_series_by_symbol: Arc::new(adjusted_close_series_by_symbol), 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, }) @@ -2525,6 +2523,12 @@ impl DataSet { { return Vec::new(); } + if matches!(field.as_str(), "volume" | "stock_volume") { + return self + .market_series(symbol) + .and_then(|series| series.current_volume_values(date, lookback)) + .unwrap_or_default(); + } self.market_series(symbol) .map(|series| series.trailing_numeric_values(date, lookback, &field, true)) .unwrap_or_default() @@ -2537,40 +2541,20 @@ impl DataSet { 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; + let end_index = if include_now { + series.end_index(date) + } else { + series.previous_completed_end_index(date) }; - !dates - .iter() - .any(|trading_day| missing_dates.binary_search(trading_day).is_ok()) + end_index + .and_then(|end| series.valid_volume_window(end, lookback)) + .is_some() } pub fn factor_numeric_values( @@ -3517,7 +3501,7 @@ mod tests { let mut extra_factors = BTreeMap::new(); if let Some(values) = availability { extra_factors.insert( - SOURCE_DAILY_VOLUME_AVAILABLE_FIELD.to_string(), + "source_daily_volume_available".to_string(), values[index], ); if values[index] >= 0.5 { @@ -3576,19 +3560,31 @@ mod tests { } #[test] - fn source_volume_contract_keeps_valid_zero_volume_and_legacy_data() { + fn volume_rolling_ignores_zero_volume_rows_for_source_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) + data.market_current_numeric_moving_average(date, "000001.SZ", "volume", 2), + Some(200.0) ); assert_eq!( - data.market_current_numeric_values(date, "000001.SZ", "volume", 3), - vec![100.0, 0.0, 300.0] + data.market_current_numeric_values(date, "000001.SZ", "volume", 2), + vec![100.0, 300.0] + ); + assert_eq!( + data.market_current_numeric_moving_average(date, "000001.SZ", "volume", 3), + None + ); + assert_eq!( + data.market_decision_numeric_moving_average(date, "000001.SZ", "volume", 1), + Some(100.0) + ); + assert_eq!( + data.market_decision_numeric_values(date, "000001.SZ", "volume", 1), + vec![100.0] ); } } @@ -3782,7 +3778,7 @@ mod tests { } #[test] - fn decision_volume_average_includes_paused_zero_volume_days() { + fn decision_volume_average_ignores_paused_zero_volume_days() { let mut paused = market_row("2025-01-03", 11.0, 0); paused.paused = true; let series = SymbolPriceSeries::new( @@ -3800,14 +3796,14 @@ mod tests { NaiveDate::parse_from_str("2025-01-07", "%Y-%m-%d").unwrap(), 2 ), - Some(150.0) + Some(200.0) ); assert_eq!( series.decision_volume_moving_average( NaiveDate::parse_from_str("2025-01-07", "%Y-%m-%d").unwrap(), 3 ), - Some((100.0 + 0.0 + 300.0) / 3.0) + None ); }