revert: reject neutral symbol-id series storage

2026-09-06 06:29:51 +08:00
parent a7f96c030f
commit c5767ca272
3 changed files with 111 additions and 174 deletions
@@ -1318,6 +1318,8 @@ pub struct DataSet {
    execution_quote_dates: Arc<Vec<NaiveDate>>,
    order_book_depth_index: Arc<HashMap<(NaiveDate, String), Vec<IntradayOrderBookDepthLevel>>>,
    benchmark_by_date: Arc<BTreeMap<NaiveDate, BenchmarkSnapshot>>,
+    market_series_by_symbol: Arc<AHashMap<String, Arc<SymbolPriceSeries>>>,
+    adjusted_close_series_by_symbol: Arc<AHashMap<String, Arc<AdjustedCloseSeries>>>,
    market_series_by_symbol_id: Arc<Vec<Option<Arc<SymbolPriceSeries>>>>,
    adjusted_close_series_by_symbol_id: Arc<Vec<Option<Arc<AdjustedCloseSeries>>>>,
    market_series_end_positions_by_calendar_index: Arc<Option<CalendarSeriesEndPositions>>,
@@ -1707,68 +1709,44 @@ impl DataSet {
            .into_iter()
            .map(|instrument| (instrument.symbol.clone(), instrument))
            .collect::<HashMap<_, _>>();
-        let symbol_id_by_code = build_symbol_id_index(
-            &instruments,
-            &market_by_date,
-            &factor_by_date,
-            &candidate_by_date,
-        );
-        let symbol_count = symbol_id_by_code.len();
-        let mut symbol_by_id = vec![Arc::<str>::from(""); symbol_count];
-        for (symbol, symbol_id) in &symbol_id_by_code {
-            symbol_by_id[*symbol_id as usize] = Arc::<str>::from(symbol.as_str());
-        }
-        let mut instruments_by_symbol_id = vec![None; symbol_count];
-        for (symbol, instrument) in &instruments {
-            if let Some(symbol_id) = symbol_id_by_code.get(symbol).copied() {
-                instruments_by_symbol_id[symbol_id as usize] = Some(instrument.clone());
-            }
-        }
-
-        let mut market_rows_by_symbol_id = (0..symbol_count)
-            .map(|_| Vec::<&DailyMarketSnapshot>::new())
-            .collect::<Vec<_>>();
+        let mut market_rows_by_symbol = AHashMap::<String, Vec<&DailyMarketSnapshot>>::new();
        for row in market_by_date.values().flatten() {
-            let symbol_id = *symbol_id_by_code
-                .get(row.symbol.as_str())
-                .expect("market symbol missing from FIDC symbol index");
-            market_rows_by_symbol_id[symbol_id as usize].push(row);
+            if let Some(rows) = market_rows_by_symbol.get_mut(row.symbol.as_str()) {
+                rows.push(row);
+                continue;
+            }
+            market_rows_by_symbol.insert(row.symbol.clone(), vec![row]);
        }
-        let market_series_by_symbol_id = market_rows_by_symbol_id
+        let market_rows_by_symbol = market_rows_by_symbol.into_iter().collect::<Vec<_>>();
+        let market_series_by_symbol = market_rows_by_symbol
            .into_par_iter()
-            .enumerate()
-            .map(|(symbol_id, rows)| {
-                if rows.is_empty() {
-                    return None;
-                }
-                Some(Arc::new(SymbolPriceSeries::from_sorted_rows(
-                    symbol_by_id[symbol_id].to_string(),
-                    rows,
-                )))
+            .map(|(symbol, rows)| {
+                let series = Arc::new(SymbolPriceSeries::from_sorted_rows(symbol.clone(), rows));
+                (symbol, series)
            })
-            .collect::<Vec<_>>();
-        let mut factor_rows_by_symbol_id = (0..symbol_count)
-            .map(|_| Vec::<&DailyFactorSnapshot>::new())
-            .collect::<Vec<_>>();
+            .collect::<Vec<_>>()
+            .into_iter()
+            .collect::<AHashMap<_, _>>();
+        let mut factor_rows_by_symbol = AHashMap::<&str, Vec<&DailyFactorSnapshot>>::new();
        for row in factor_by_date.values().flatten() {
-            let symbol_id = *symbol_id_by_code
-                .get(row.symbol.as_str())
-                .expect("factor symbol missing from FIDC symbol index");
-            factor_rows_by_symbol_id[symbol_id as usize].push(row);
+            factor_rows_by_symbol
+                .entry(row.symbol.as_str())
+                .or_default()
+                .push(row);
        }
-        let adjusted_close_series_by_symbol_id = market_series_by_symbol_id
+        let adjusted_close_series_by_symbol = market_series_by_symbol
            .par_iter()
-            .enumerate()
-            .map(|(symbol_id, market)| {
-                market.as_ref().and_then(|market| {
-                    AdjustedCloseSeries::new(
-                        market,
-                        factor_rows_by_symbol_id[symbol_id].as_slice(),
-                    )
-                    .map(Arc::new)
-                })
+            .filter_map(|(symbol, market)| {
+                let factor_rows = factor_rows_by_symbol
+                    .get(symbol.as_str())
+                    .map(Vec::as_slice)
+                    .unwrap_or_default();
+                AdjustedCloseSeries::new(market, factor_rows)
+                    .map(|series| (symbol.clone(), Arc::new(series)))
            })
-            .collect::<Vec<_>>();
+            .collect::<Vec<_>>()
+            .into_iter()
+            .collect::<AHashMap<_, _>>();
        let factor_texts = factor_texts
            .into_iter()
            .filter_map(|mut item| {
@@ -1786,6 +1764,22 @@ impl DataSet {
            .map(|item| ((item.date, item.symbol.clone(), item.field.clone()), item))
            .collect::<HashMap<_, _>>();

+        let symbol_id_by_code = build_symbol_id_index(
+            &instruments,
+            &market_by_date,
+            &factor_by_date,
+            &candidate_by_date,
+        );
+        let mut symbol_by_id = vec![Arc::<str>::from(""); symbol_id_by_code.len()];
+        for (symbol, symbol_id) in &symbol_id_by_code {
+            symbol_by_id[*symbol_id as usize] = Arc::<str>::from(symbol.as_str());
+        }
+        let mut instruments_by_symbol_id = vec![None; symbol_id_by_code.len()];
+        for (symbol, instrument) in &instruments {
+            if let Some(symbol_id) = symbol_id_by_code.get(symbol).copied() {
+                instruments_by_symbol_id[symbol_id as usize] = Some(instrument.clone());
+            }
+        }
        let market_symbol_ids_by_date =
            build_group_symbol_ids(&market_by_date, &symbol_id_by_code, |item| {
                item.symbol.as_str()
@@ -1815,6 +1809,18 @@ impl DataSet {
            &candidate_symbol_ids_by_date,
            symbol_id_by_code.len(),
        );
+        let mut market_series_by_symbol_id = vec![None; symbol_id_by_code.len()];
+        for (symbol, series) in &market_series_by_symbol {
+            if let Some(symbol_id) = symbol_id_by_code.get(symbol).copied() {
+                market_series_by_symbol_id[symbol_id as usize] = Some(Arc::clone(series));
+            }
+        }
+        let mut adjusted_close_series_by_symbol_id = vec![None; symbol_id_by_code.len()];
+        for (symbol, series) in &adjusted_close_series_by_symbol {
+            if let Some(symbol_id) = symbol_id_by_code.get(symbol).copied() {
+                adjusted_close_series_by_symbol_id[symbol_id as usize] = Some(Arc::clone(series));
+            }
+        }
        let market_series_end_positions_by_calendar_index =
            build_calendar_series_end_positions(&market_series_by_symbol_id, &calendar);
        let execution_quotes_by_date = build_execution_quote_index(execution_quotes);
@@ -1847,6 +1853,8 @@ impl DataSet {
            execution_quote_dates: Arc::new(execution_quote_dates),
            order_book_depth_index: Arc::new(order_book_depth_index),
            benchmark_by_date: Arc::new(benchmark_by_date),
+            market_series_by_symbol: Arc::new(market_series_by_symbol),
+            adjusted_close_series_by_symbol: Arc::new(adjusted_close_series_by_symbol),
            market_series_by_symbol_id: Arc::new(market_series_by_symbol_id),
            adjusted_close_series_by_symbol_id: Arc::new(adjusted_close_series_by_symbol_id),
            market_series_end_positions_by_calendar_index: Arc::new(
@@ -1985,7 +1993,7 @@ impl DataSet {
    }

    fn market_series(&self, symbol: &str) -> Option<&SymbolPriceSeries> {
-        self.market_series_by_symbol_id(self.symbol_id(symbol)?)
+        self.market_series_by_symbol.get(symbol).map(Arc::as_ref)
    }

    fn market_series_by_symbol_id(&self, symbol_id: u32) -> Option<&SymbolPriceSeries> {
@@ -1995,7 +2003,9 @@ impl DataSet {
    }

    fn adjusted_close_series(&self, symbol: &str) -> Option<&AdjustedCloseSeries> {
-        self.adjusted_close_series_by_symbol_id(self.symbol_id(symbol)?)
+        self.adjusted_close_series_by_symbol
+            .get(symbol)
+            .map(Arc::as_ref)
    }

    fn adjusted_close_series_by_symbol_id(&self, symbol_id: u32) -> Option<&AdjustedCloseSeries> {