保留成交量空值滚动语义

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、性能验证和策略迁移基线：

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<RwLock<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>>,
 }
@@ -1186,10 +1190,30 @@ impl DataSet {
     ) -> Result<Self, DataSetError> {
         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 factors = normalize_factor_snapshots(factors);
-            .into_iter()
+        let mut source_daily_volume_contract_symbols = HashSet::new();
-            .map(Arc::new)
+        let mut source_daily_volume_missing_dates_by_symbol =
-            .collect::<Vec<_>>();
+            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<_>>();
 
         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
+            "volume" | "stock_volume" => {
-                .market_series(symbol)
+                if !self.source_daily_volume_window_available(date, symbol, lookback, false) {
-                .and_then(|series| series.decision_volume_moving_average(date, lookback)),
+                    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
+            "volume" | "stock_volume" => {
-                .market_series(symbol)
+                if !self.source_daily_volume_window_available(date, symbol, lookback, true) {
-                .and_then(|series| series.current_volume_moving_average(date, lookback))
+                    None
-                .or_else(|| self.factor_moving_average(date, symbol, "daily_volume", lookback)),
+                } 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
+            "volume" | "stock_volume" => {
-                .market_series(symbol)
+                if !self.source_daily_volume_window_available(date, symbol, lookback, false) {
-                .and_then(|series| series.decision_volume_values(date, lookback))
+                    Vec::new()
-                .unwrap_or_default(),
+                } 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<f64> {
+        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);