构建无锁证券序列索引

crates/fidc-core/src/data.rs

@@ -1,5 +1,5 @@
 use std::collections::{BTreeMap, HashMap, HashSet};
-use std::sync::{Arc, OnceLock, RwLock};
+use std::sync::{Arc, OnceLock};
 
 use chrono::{NaiveDate, NaiveDateTime};
 use serde::{Deserialize, Serialize};
@@ -1052,8 +1052,8 @@ pub struct DataSet {
     execution_quotes_by_date: HashMap<NaiveDate, HashMap<String, Vec<IntradayExecutionQuote>>>,
     order_book_depth_index: HashMap<(NaiveDate, String), Vec<IntradayOrderBookDepthLevel>>,
     benchmark_by_date: BTreeMap<NaiveDate, BenchmarkSnapshot>,
-    market_series_by_symbol: Arc<RwLock<HashMap<String, Arc<SymbolPriceSeries>>>>,
+    market_series_by_symbol: Arc<HashMap<String, Arc<SymbolPriceSeries>>>,
-    adjusted_close_series_by_symbol: Arc<RwLock<HashMap<String, Arc<AdjustedCloseSeries>>>>,
+    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>,
@@ -1242,6 +1242,27 @@ impl DataSet {
 
         let mut factor_by_date = group_arc_by_date(&factors, |item| item.date);
         sort_arc_groups_by_symbol(&mut factor_by_date, |item| item.symbol.as_str());
+        let mut market_rows_by_symbol = HashMap::<String, Vec<&DailyMarketSnapshot>>::new();
+        for row in &market {
+            market_rows_by_symbol
+                .entry(row.symbol.clone())
+                .or_default()
+                .push(row.as_ref());
+        }
+        let market_series_by_symbol = market_rows_by_symbol
+            .into_iter()
+            .map(|(symbol, rows)| {
+                let series = Arc::new(SymbolPriceSeries::new(symbol.clone(), rows));
+                (symbol, series)
+            })
+            .collect::<HashMap<_, _>>();
+        let adjusted_close_series_by_symbol = market_series_by_symbol
+            .iter()
+            .filter_map(|(symbol, market)| {
+                AdjustedCloseSeries::new(market, &factor_by_date)
+                    .map(|series| (symbol.clone(), Arc::new(series)))
+            })
+            .collect::<HashMap<_, _>>();
         let factor_texts = factor_texts
             .into_iter()
             .filter_map(|mut item| {
@@ -1285,8 +1306,8 @@ impl DataSet {
             execution_quotes_by_date,
             order_book_depth_index,
             benchmark_by_date,
-            market_series_by_symbol: Arc::new(RwLock::new(HashMap::new())),
+            market_series_by_symbol: Arc::new(market_series_by_symbol),
-            adjusted_close_series_by_symbol: Arc::new(RwLock::new(HashMap::new())),
+            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,
@@ -1340,61 +1361,11 @@ impl DataSet {
     }
 
     fn market_series(&self, symbol: &str) -> Option<Arc<SymbolPriceSeries>> {
-        if let Some(series) = self
+        self.market_series_by_symbol.get(symbol).cloned()
-            .market_series_by_symbol
-            .read()
-            .expect("market series cache lock poisoned")
-            .get(symbol)
-            .cloned()
-        {
-            return Some(series);
-        }
-
-        let rows = self
-            .market_by_date
-            .values()
-            .filter_map(|day_rows| find_arc_by_symbol(day_rows, symbol, |row| row.symbol.as_str()))
-            .collect::<Vec<_>>();
-        if rows.is_empty() {
-            return None;
-        }
-
-        let series = Arc::new(SymbolPriceSeries::new(symbol.to_string(), rows));
-        let mut cache = self
-            .market_series_by_symbol
-            .write()
-            .expect("market series cache lock poisoned");
-        Some(
-            cache
-                .entry(symbol.to_string())
-                .or_insert_with(|| Arc::clone(&series))
-                .clone(),
-        )
     }
 
     fn adjusted_close_series(&self, symbol: &str) -> Option<Arc<AdjustedCloseSeries>> {
-        if let Some(series) = self
+        self.adjusted_close_series_by_symbol.get(symbol).cloned()
-            .adjusted_close_series_by_symbol
-            .read()
-            .expect("adjusted close series cache lock poisoned")
-            .get(symbol)
-            .cloned()
-        {
-            return Some(series);
-        }
-
-        let market = self.market_series(symbol)?;
-        let series = Arc::new(AdjustedCloseSeries::new(&market, &self.factor_by_date)?);
-        let mut cache = self
-            .adjusted_close_series_by_symbol
-            .write()
-            .expect("adjusted close series cache lock poisoned");
-        Some(
-            cache
-                .entry(symbol.to_string())
-                .or_insert_with(|| Arc::clone(&series))
-                .clone(),
-        )
     }
 
     pub fn factor(&self, date: NaiveDate, symbol: &str) -> Option<&DailyFactorSnapshot> {