diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs
index c76847e..4248e87 100644
--- a/crates/fidc-core/src/data.rs
+++ b/crates/fidc-core/src/data.rs
@@ -983,8 +983,8 @@ impl BenchmarkPriceSeries {
 pub struct DataSet {
     instruments: HashMap<String, Instrument>,
     calendar: TradingCalendar,
-    market_by_date: BTreeMap<NaiveDate, Vec<DailyMarketSnapshot>>,
-    market_index: HashMap<NaiveDate, HashMap<String, DailyMarketSnapshot>>,
+    market_by_date: BTreeMap<NaiveDate, Vec<Arc<DailyMarketSnapshot>>>,
+    market_index: HashMap<NaiveDate, HashMap<String, Arc<DailyMarketSnapshot>>>,
     factor_by_date: BTreeMap<NaiveDate, Vec<Arc<DailyFactorSnapshot>>>,
     factor_index: HashMap<NaiveDate, HashMap<String, Arc<DailyFactorSnapshot>>>,
     factor_text_by_date: BTreeMap<NaiveDate, Vec<FactorTextValue>>,
@@ -1246,8 +1246,10 @@ impl DataSet {
             .map(|instrument| (instrument.symbol.clone(), instrument))
             .collect::<HashMap<_, _>>();
 
-        let market_by_date = group_by_date(market.clone(), |item| item.date);
-        let mut market_index = HashMap::<NaiveDate, HashMap<String, DailyMarketSnapshot>>::new();
+        let market = market.into_iter().map(Arc::new).collect::<Vec<_>>();
+        let market_by_date = group_arc_by_date(&market, |item| item.date);
+        let mut market_index =
+            HashMap::<NaiveDate, HashMap<String, Arc<DailyMarketSnapshot>>>::new();
         for item in market {
             market_index
                 .entry(item.date)
@@ -1373,6 +1375,7 @@ impl DataSet {
         self.market_index
             .get(&date)
             .and_then(|rows| rows.get(symbol))
+            .map(Arc::as_ref)
     }
 
     pub fn factor(&self, date: NaiveDate, symbol: &str) -> Option<&DailyFactorSnapshot> {
@@ -1500,7 +1503,7 @@ impl DataSet {
         let market = self
             .market_by_date
             .values()
-            .flat_map(|rows| rows.iter().cloned())
+            .flat_map(|rows| rows.iter().map(|row| row.as_ref().clone()))
             .collect::<Vec<_>>();
         let factors = self
             .factor_by_date
@@ -2142,6 +2145,7 @@ impl DataSet {
                 .range(start..=end)
                 .flat_map(|(_, rows)| rows.iter())
                 .filter(|row| row.symbol == symbol)
+                .map(Arc::as_ref)
                 .map(daily_market_price_bar)
                 .collect(),
             Some("1m") | Some("tick") => {
@@ -2190,6 +2194,7 @@ impl DataSet {
         self.market_index
             .get(&previous_date)
             .and_then(|rows| rows.get(symbol))
+            .map(Arc::as_ref)
     }
 
     pub fn factor_snapshots_on(&self, date: NaiveDate) -> Vec<&DailyFactorSnapshot> {
@@ -2209,7 +2214,7 @@ impl DataSet {
     pub fn market_snapshots_on(&self, date: NaiveDate) -> Vec<&DailyMarketSnapshot> {
         self.market_by_date
             .get(&date)
-            .map(|rows| rows.iter().collect())
+            .map(|rows| rows.iter().map(Arc::as_ref).collect())
             .unwrap_or_default()
     }
 
@@ -2228,7 +2233,11 @@ impl DataSet {
         Ok(DailySnapshotBundle {
             date,
             benchmark,
-            market: self.market_by_date.get(&date).cloned().unwrap_or_default(),
+            market: self
+                .market_by_date
+                .get(&date)
+                .map(|rows| rows.iter().map(|row| row.as_ref().clone()).collect())
+                .unwrap_or_default(),
             factors: self
                 .factor_by_date
                 .get(&date)
@@ -2314,7 +2323,10 @@ impl DataSet {
                         .get(day)
                         .and_then(|rows| rows.get(symbol))
                         .map(Arc::as_ref),
-                    self.market_index.get(day).and_then(|rows| rows.get(symbol)),
+                    self.market_index
+                        .get(day)
+                        .and_then(|rows| rows.get(symbol))
+                        .map(Arc::as_ref),
                 )
             })
             .collect()
@@ -3578,12 +3590,12 @@ mod optional_date_format {
 }
 
 fn build_market_series(
-    market_by_date: &BTreeMap<NaiveDate, Vec<DailyMarketSnapshot>>,
+    market_by_date: &BTreeMap<NaiveDate, Vec<Arc<DailyMarketSnapshot>>>,
 ) -> HashMap<String, SymbolPriceSeries> {
     let mut grouped = HashMap::<String, Vec<&DailyMarketSnapshot>>::new();
     for rows in market_by_date.values() {
         for row in rows {
-            grouped.entry(row.symbol.clone()).or_default().push(row);
+            grouped.entry(row.symbol.clone()).or_default().push(row.as_ref());
         }
     }
 
@@ -3656,7 +3668,7 @@ fn build_order_book_depth_index(
 fn build_eligible_universe(
     factor_by_date: &BTreeMap<NaiveDate, Vec<Arc<DailyFactorSnapshot>>>,
     candidate_index: &HashMap<NaiveDate, HashMap<String, Arc<CandidateEligibility>>>,
-    market_index: &HashMap<NaiveDate, HashMap<String, DailyMarketSnapshot>>,
+    market_index: &HashMap<NaiveDate, HashMap<String, Arc<DailyMarketSnapshot>>>,
     instruments: &HashMap<String, Instrument>,
 ) -> BTreeMap<NaiveDate, Vec<EligibleUniverseSnapshot>> {
     let mut per_date = BTreeMap::<NaiveDate, Vec<EligibleUniverseSnapshot>>::new();
@@ -3679,6 +3691,7 @@ fn build_eligible_universe(
             else {
                 continue;
             };
+            let market = market.as_ref();
             if ChinaAShareRiskControl::selection_rejection_reason(
                 *date,
                 candidate,