共享日线行情索引存储

This commit is contained in:
boris
2026-06-21 02:11:44 +08:00
parent c409d500b3
commit 9d41971d3f
+24 -11
View File
@@ -983,8 +983,8 @@ impl BenchmarkPriceSeries {
pub struct DataSet { pub struct DataSet {
instruments: HashMap<String, Instrument>, instruments: HashMap<String, Instrument>,
calendar: TradingCalendar, calendar: TradingCalendar,
market_by_date: BTreeMap<NaiveDate, Vec<DailyMarketSnapshot>>, market_by_date: BTreeMap<NaiveDate, Vec<Arc<DailyMarketSnapshot>>>,
market_index: HashMap<NaiveDate, HashMap<String, DailyMarketSnapshot>>, market_index: HashMap<NaiveDate, HashMap<String, Arc<DailyMarketSnapshot>>>,
factor_by_date: BTreeMap<NaiveDate, Vec<Arc<DailyFactorSnapshot>>>, factor_by_date: BTreeMap<NaiveDate, Vec<Arc<DailyFactorSnapshot>>>,
factor_index: HashMap<NaiveDate, HashMap<String, Arc<DailyFactorSnapshot>>>, factor_index: HashMap<NaiveDate, HashMap<String, Arc<DailyFactorSnapshot>>>,
factor_text_by_date: BTreeMap<NaiveDate, Vec<FactorTextValue>>, factor_text_by_date: BTreeMap<NaiveDate, Vec<FactorTextValue>>,
@@ -1246,8 +1246,10 @@ impl DataSet {
.map(|instrument| (instrument.symbol.clone(), instrument)) .map(|instrument| (instrument.symbol.clone(), instrument))
.collect::<HashMap<_, _>>(); .collect::<HashMap<_, _>>();
let market_by_date = group_by_date(market.clone(), |item| item.date); let market = market.into_iter().map(Arc::new).collect::<Vec<_>>();
let mut market_index = HashMap::<NaiveDate, HashMap<String, DailyMarketSnapshot>>::new(); 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 { for item in market {
market_index market_index
.entry(item.date) .entry(item.date)
@@ -1373,6 +1375,7 @@ impl DataSet {
self.market_index self.market_index
.get(&date) .get(&date)
.and_then(|rows| rows.get(symbol)) .and_then(|rows| rows.get(symbol))
.map(Arc::as_ref)
} }
pub fn factor(&self, date: NaiveDate, symbol: &str) -> Option<&DailyFactorSnapshot> { pub fn factor(&self, date: NaiveDate, symbol: &str) -> Option<&DailyFactorSnapshot> {
@@ -1500,7 +1503,7 @@ impl DataSet {
let market = self let market = self
.market_by_date .market_by_date
.values() .values()
.flat_map(|rows| rows.iter().cloned()) .flat_map(|rows| rows.iter().map(|row| row.as_ref().clone()))
.collect::<Vec<_>>(); .collect::<Vec<_>>();
let factors = self let factors = self
.factor_by_date .factor_by_date
@@ -2142,6 +2145,7 @@ impl DataSet {
.range(start..=end) .range(start..=end)
.flat_map(|(_, rows)| rows.iter()) .flat_map(|(_, rows)| rows.iter())
.filter(|row| row.symbol == symbol) .filter(|row| row.symbol == symbol)
.map(Arc::as_ref)
.map(daily_market_price_bar) .map(daily_market_price_bar)
.collect(), .collect(),
Some("1m") | Some("tick") => { Some("1m") | Some("tick") => {
@@ -2190,6 +2194,7 @@ impl DataSet {
self.market_index self.market_index
.get(&previous_date) .get(&previous_date)
.and_then(|rows| rows.get(symbol)) .and_then(|rows| rows.get(symbol))
.map(Arc::as_ref)
} }
pub fn factor_snapshots_on(&self, date: NaiveDate) -> Vec<&DailyFactorSnapshot> { 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> { pub fn market_snapshots_on(&self, date: NaiveDate) -> Vec<&DailyMarketSnapshot> {
self.market_by_date self.market_by_date
.get(&date) .get(&date)
.map(|rows| rows.iter().collect()) .map(|rows| rows.iter().map(Arc::as_ref).collect())
.unwrap_or_default() .unwrap_or_default()
} }
@@ -2228,7 +2233,11 @@ impl DataSet {
Ok(DailySnapshotBundle { Ok(DailySnapshotBundle {
date, date,
benchmark, 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 factors: self
.factor_by_date .factor_by_date
.get(&date) .get(&date)
@@ -2314,7 +2323,10 @@ impl DataSet {
.get(day) .get(day)
.and_then(|rows| rows.get(symbol)) .and_then(|rows| rows.get(symbol))
.map(Arc::as_ref), .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() .collect()
@@ -3578,12 +3590,12 @@ mod optional_date_format {
} }
fn build_market_series( fn build_market_series(
market_by_date: &BTreeMap<NaiveDate, Vec<DailyMarketSnapshot>>, market_by_date: &BTreeMap<NaiveDate, Vec<Arc<DailyMarketSnapshot>>>,
) -> HashMap<String, SymbolPriceSeries> { ) -> HashMap<String, SymbolPriceSeries> {
let mut grouped = HashMap::<String, Vec<&DailyMarketSnapshot>>::new(); let mut grouped = HashMap::<String, Vec<&DailyMarketSnapshot>>::new();
for rows in market_by_date.values() { for rows in market_by_date.values() {
for row in rows { 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( fn build_eligible_universe(
factor_by_date: &BTreeMap<NaiveDate, Vec<Arc<DailyFactorSnapshot>>>, factor_by_date: &BTreeMap<NaiveDate, Vec<Arc<DailyFactorSnapshot>>>,
candidate_index: &HashMap<NaiveDate, HashMap<String, Arc<CandidateEligibility>>>, 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>, instruments: &HashMap<String, Instrument>,
) -> BTreeMap<NaiveDate, Vec<EligibleUniverseSnapshot>> { ) -> BTreeMap<NaiveDate, Vec<EligibleUniverseSnapshot>> {
let mut per_date = BTreeMap::<NaiveDate, Vec<EligibleUniverseSnapshot>>::new(); let mut per_date = BTreeMap::<NaiveDate, Vec<EligibleUniverseSnapshot>>::new();
@@ -3679,6 +3691,7 @@ fn build_eligible_universe(
else { else {
continue; continue;
}; };
let market = market.as_ref();
if ChinaAShareRiskControl::selection_rejection_reason( if ChinaAShareRiskControl::selection_rejection_reason(
*date, *date,
candidate, candidate,