统一成交量滚动有效样本口径

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<f64>,
     prev_close_prefix: Vec<f64>,
     last_prefix: Vec<f64>,
-    volume_prefix: Vec<f64>,
+    valid_volume_sum_prefix: Vec<f64>,
+    valid_volume_count_prefix: Vec<usize>,
 }
 
 #[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
+        let mut valid_volume_sum_prefix = Vec::with_capacity(volumes.len() + 1);
-            .iter()
+        let mut valid_volume_count_prefix = Vec::with_capacity(volumes.len() + 1);
-            .map(|value| *value as f64)
+        valid_volume_sum_prefix.push(0.0);
-            .collect::<Vec<_>>();
+        valid_volume_count_prefix.push(0);
-        let volume_prefix = prefix_sums(&volume_values);
+        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<f64> {
-        if lookback == 0 {
-            return None;
-        }
         let end = self.previous_completed_end_index(date)?;
-        if end < lookback {
+        self.valid_volume_window(end, lookback)
-            return None;
+            .map(|(start, end)| {
-        }
+                normalize_rolling_factor(
-        let start = end - lookback;
+                    (self.valid_volume_sum_prefix[end] - self.valid_volume_sum_prefix[start])
-        let sum = self.volume_prefix[end] - self.volume_prefix[start];
+                        / lookback as f64,
-        Some(sum / lookback as f64)
+                    12,
+                )
+            })
     }
 
     fn current_volume_moving_average(&self, date: NaiveDate, lookback: usize) -> Option<f64> {
-        if lookback == 0 {
-            return None;
-        }
         let end = self.end_index(date)?;
-        if end < lookback {
+        self.valid_volume_window(end, lookback)
-            return None;
+            .map(|(start, end)| {
-        }
+                normalize_rolling_factor(
-        let start = end - lookback;
+                    (self.valid_volume_sum_prefix[end] - self.valid_volume_sum_prefix[start])
-        let sum = self.volume_prefix[end] - self.volume_prefix[start];
+                        / lookback as f64,
-        if !sum.is_finite() {
+                    12,
-            return None;
+                )
-        }
+            })
-        Some(normalize_rolling_factor(sum / lookback as f64, 6))
     }
 
     fn decision_volume_values(&self, date: NaiveDate, lookback: usize) -> Option<Vec<f64>> {
-        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<Vec<f64>> {
+        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;
+        let valid_count = *self.valid_volume_count_prefix.get(end)?;
-        Some(
+        if valid_count < lookback {
-            self.volumes[start..end]
+            return None;
-                .iter()
+        }
-                .map(|value| *value as f64)
+        let target_count = valid_count - lookback;
-                .collect(),
+        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<Vec<f64>> {
+        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::<Vec<_>>();
+        (values.len() == lookback).then_some(values)
     }
 
     fn end_index(&self, date: NaiveDate) -> Option<usize> {
@@ -1057,8 +1081,6 @@ pub struct DataSet {
     adjusted_close_series_by_symbol: Arc<HashMap<String, Arc<AdjustedCloseSeries>>>,
     benchmark_series_cache: BenchmarkPriceSeries,
     eligible_universe_by_date: Arc<OnceLock<BTreeMap<NaiveDate, Vec<EligibleUniverseSnapshot>>>>,
-    source_daily_volume_contract_symbols: HashSet<String>,
-    source_daily_volume_missing_dates_by_symbol: HashMap<String, Vec<NaiveDate>>,
     benchmark_code: String,
     futures_params_by_symbol: HashMap<String, Vec<FuturesTradingParameter>>,
 }
@@ -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::<String, Vec<NaiveDate>>::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::<Vec<_>>();
         let candidates = candidates.into_iter().map(Arc::new).collect::<Vec<_>>();
 
@@ -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
+        let end_index = if include_now {
-            .iter()
+            series.end_index(date)
-            .any(|trading_day| series.dates.binary_search(trading_day).is_err())
+        } else {
-        {
+            series.previous_completed_end_index(date)
-            return false;
-        }
-        let Some(missing_dates) = self.source_daily_volume_missing_dates_by_symbol.get(symbol)
-        else {
-            return true;
         };
-        !dates
+        end_index
-            .iter()
+            .and_then(|end| series.valid_volume_window(end, lookback))
-            .any(|trading_day| missing_dates.binary_search(trading_day).is_ok())
+            .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),
+                data.market_current_numeric_moving_average(date, "000001.SZ", "volume", 2),
-                Some(133.333333)
+                Some(200.0)
             );
             assert_eq!(
-                data.market_current_numeric_values(date, "000001.SZ", "volume", 3),
+                data.market_current_numeric_values(date, "000001.SZ", "volume", 2),
-                vec![100.0, 0.0, 300.0]
+                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
         );
     }