From ac30d86b6ad921241bb863438a17ef95a4e9f84a Mon Sep 17 00:00:00 2001 From: boris Date: Tue, 25 Aug 2026 21:55:18 +0800 Subject: [PATCH] =?UTF-8?q?=E7=BA=BF=E6=80=A7=E6=9E=84=E5=BB=BA=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E9=9B=86=E4=BB=B7=E6=A0=BC=E5=BA=8F=E5=88=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/fidc-core/src/data.rs | 173 +++++++++++++++++++++++++---------- 1 file changed, 127 insertions(+), 46 deletions(-) diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index 22e2731..d8bfb12 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -14,6 +14,8 @@ use crate::futures::FuturesTradingParameter; use crate::instrument::Instrument; use crate::risk_control::{ChinaAShareRiskControl, FidcRiskControlConfig}; +const BACKWARD_ADJUSTMENT_FACTOR_FIELD: &str = "adjustment_factor_backward1"; + mod date_format { use chrono::NaiveDate; use serde::{self, Deserialize, Deserializer, Serializer}; @@ -504,21 +506,35 @@ struct AdjustedCloseSeries { } impl AdjustedCloseSeries { - fn new( - market: &SymbolPriceSeries, - factor_by_date: &BTreeMap>, - ) -> Option { + fn new(market: &SymbolPriceSeries, factor_rows: &[&DailyFactorSnapshot]) -> Option { + debug_assert!( + factor_rows + .windows(2) + .all(|window| window[0].date <= window[1].date) + ); let mut backward_factors = Vec::with_capacity(market.dates.len()); let mut back_adjusted_closes = Vec::with_capacity(market.dates.len()); let mut back_adjusted_close_prefix = Vec::with_capacity(market.dates.len() + 1); let mut missing_back_adjusted_close_prefix = Vec::with_capacity(market.dates.len() + 1); back_adjusted_close_prefix.push(0.0); missing_back_adjusted_close_prefix.push(0); + let mut factor_index = 0usize; for (date, close) in market.dates.iter().zip(&market.closes) { - let factor = factor_by_date - .get(date) - .and_then(|rows| find_by_symbol(rows, &market.symbol, |row| row.symbol.as_str())) - .and_then(|snapshot| factor_numeric_value(snapshot, "adjustment_factor_backward1")) + while factor_rows + .get(factor_index) + .is_some_and(|snapshot| snapshot.date < *date) + { + factor_index += 1; + } + let factor = factor_rows + .get(factor_index) + .filter(|snapshot| snapshot.date == *date) + .and_then(|snapshot| { + snapshot + .extra_factors + .get(BACKWARD_ADJUSTMENT_FACTOR_FIELD) + .copied() + }) .filter(|factor| factor.is_finite() && *factor > 0.0); let back_adjusted_close = factor .filter(|_| close.is_finite() && *close > 0.0) @@ -651,42 +667,64 @@ impl AdjustedCloseSeries { } impl SymbolPriceSeries { + #[cfg(test)] fn new<'a, I>(symbol: String, rows: I) -> Self where I: IntoIterator, { let mut sorted = rows.into_iter().collect::>(); sorted.sort_by_key(|row| row.date); + Self::from_sorted_rows(symbol, sorted) + } - let dates = sorted.iter().map(|row| row.date).collect::>(); - let timestamps = sorted - .iter() - .map(|row| row.timestamp.clone()) - .collect::>(); - let day_opens = sorted.iter().map(|row| row.day_open).collect::>(); - let opens = sorted.iter().map(|row| row.open).collect::>(); - let highs = sorted.iter().map(|row| row.high).collect::>(); - let lows = sorted.iter().map(|row| row.low).collect::>(); - let closes = sorted.iter().map(|row| row.close).collect::>(); - let prev_closes = sorted.iter().map(|row| row.prev_close).collect::>(); - let last_prices = sorted.iter().map(|row| row.last_price).collect::>(); - let bid1s = sorted.iter().map(|row| row.bid1).collect::>(); - let ask1s = sorted.iter().map(|row| row.ask1).collect::>(); - let volumes = sorted.iter().map(|row| row.volume).collect::>(); - let minute_volumes = sorted - .iter() - .map(|row| row.minute_volume) - .collect::>(); - let bid1_volumes = sorted.iter().map(|row| row.bid1_volume).collect::>(); - let ask1_volumes = sorted.iter().map(|row| row.ask1_volume).collect::>(); - let trading_phases = sorted - .iter() - .map(|row| row.trading_phase.clone()) - .collect::>(); - let paused = sorted.iter().map(|row| row.paused).collect::>(); - let upper_limits = sorted.iter().map(|row| row.upper_limit).collect::>(); - let lower_limits = sorted.iter().map(|row| row.lower_limit).collect::>(); - let price_ticks = sorted.iter().map(|row| row.price_tick).collect::>(); + fn from_sorted_rows(symbol: String, rows: Vec<&DailyMarketSnapshot>) -> Self { + debug_assert!( + rows.windows(2) + .all(|window| window[0].date <= window[1].date) + ); + let row_count = rows.len(); + let mut dates = Vec::with_capacity(row_count); + let mut timestamps = Vec::with_capacity(row_count); + let mut day_opens = Vec::with_capacity(row_count); + let mut opens = Vec::with_capacity(row_count); + let mut highs = Vec::with_capacity(row_count); + let mut lows = Vec::with_capacity(row_count); + let mut closes = Vec::with_capacity(row_count); + let mut prev_closes = Vec::with_capacity(row_count); + let mut last_prices = Vec::with_capacity(row_count); + let mut bid1s = Vec::with_capacity(row_count); + let mut ask1s = Vec::with_capacity(row_count); + let mut volumes = Vec::with_capacity(row_count); + let mut minute_volumes = Vec::with_capacity(row_count); + let mut bid1_volumes = Vec::with_capacity(row_count); + let mut ask1_volumes = Vec::with_capacity(row_count); + let mut trading_phases = Vec::with_capacity(row_count); + let mut paused = Vec::with_capacity(row_count); + let mut upper_limits = Vec::with_capacity(row_count); + let mut lower_limits = Vec::with_capacity(row_count); + let mut price_ticks = Vec::with_capacity(row_count); + for row in rows { + dates.push(row.date); + timestamps.push(row.timestamp.clone()); + day_opens.push(row.day_open); + opens.push(row.open); + highs.push(row.high); + lows.push(row.low); + closes.push(row.close); + prev_closes.push(row.prev_close); + last_prices.push(row.last_price); + bid1s.push(row.bid1); + ask1s.push(row.ask1); + volumes.push(row.volume); + minute_volumes.push(row.minute_volume); + bid1_volumes.push(row.bid1_volume); + ask1_volumes.push(row.ask1_volume); + trading_phases.push(row.trading_phase.clone()); + paused.push(row.paused); + upper_limits.push(row.upper_limit); + lower_limits.push(row.lower_limit); + price_ticks.push(row.price_tick); + } let open_prefix = prefix_sums(&opens); let close_prefix = prefix_sums(&closes); let prev_close_prefix = prefix_sums(&prev_closes); @@ -1316,16 +1354,27 @@ impl DataSet { let market_series_by_symbol = market_rows_by_symbol .into_par_iter() .map(|(symbol, rows)| { - let series = Arc::new(SymbolPriceSeries::new(symbol.clone(), rows)); + let series = Arc::new(SymbolPriceSeries::from_sorted_rows(symbol.clone(), rows)); (symbol, series) }) .collect::>() .into_iter() .collect::>(); + let mut factor_rows_by_symbol = AHashMap::<&str, Vec<&DailyFactorSnapshot>>::new(); + for row in factor_by_date.values().flatten() { + factor_rows_by_symbol + .entry(row.symbol.as_str()) + .or_default() + .push(row); + } let adjusted_close_series_by_symbol = market_series_by_symbol .par_iter() .filter_map(|(symbol, market)| { - AdjustedCloseSeries::new(market, &factor_by_date) + 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::>() @@ -3132,8 +3181,8 @@ fn industry_name_factor_aliases(source: &str, level: usize) -> Vec { } fn factor_numeric_value(snapshot: &DailyFactorSnapshot, field: &str) -> Option { - let field = normalize_field(field); - match field.as_str() { + let field = normalized_field(field); + match field.as_ref() { "market_cap" | "market_cap_bn" => Some(snapshot.market_cap_bn), "free_float_cap" | "free_float_market_cap" | "free_float_cap_bn" => { Some(snapshot.free_float_cap_bn) @@ -3148,22 +3197,22 @@ fn factor_numeric_value(snapshot: &DailyFactorSnapshot, field: &str) -> Option snapshot.effective_turnover_ratio, "ths_market_value_stock" | "ths_market_value_stock_bn" => snapshot .extra_factors - .get(field.as_str()) + .get(field.as_ref()) .copied() .or(Some(snapshot.market_cap_bn)), "ths_current_mv_stock" | "ths_current_mv_stock_bn" => snapshot .extra_factors - .get(field.as_str()) + .get(field.as_ref()) .copied() .or(Some(snapshot.free_float_cap_bn)), "ths_turnover_ratio_stock" => snapshot .extra_factors - .get(field.as_str()) + .get(field.as_ref()) .copied() .or(snapshot.turnover_ratio), "ths_vaild_turnover_stock" | "ths_valid_turnover_stock" => snapshot .extra_factors - .get(field.as_str()) + .get(field.as_ref()) .copied() .or(snapshot.effective_turnover_ratio), other => snapshot.extra_factors.get(other).copied(), @@ -3171,7 +3220,7 @@ fn factor_numeric_value(snapshot: &DailyFactorSnapshot, field: &str) -> Option Option { - match normalize_field(field).as_str() { + match normalized_field(field).as_ref() { "last" | "last_price" | "close" | "price" => Some(snapshot.last_price), "bid1" => Some(snapshot.bid1), "ask1" => Some(snapshot.ask1), @@ -3904,6 +3953,38 @@ mod tests { )); } + #[test] + fn factor_numeric_value_normalizes_fields_without_changing_aliases() { + let snapshot = DailyFactorSnapshot { + date: NaiveDate::parse_from_str("2025-01-02", "%Y-%m-%d").unwrap(), + symbol: "000001.SZ".to_string(), + market_cap_bn: 12.5, + free_float_cap_bn: 8.0, + pe_ttm: 10.0, + turnover_ratio: None, + effective_turnover_ratio: None, + extra_factors: BTreeMap::from([("custom_factor".into(), 3.5)]), + }; + + assert_eq!(factor_numeric_value(&snapshot, " MARKET_CAP "), Some(12.5)); + assert_eq!(factor_numeric_value(&snapshot, "CUSTOM_FACTOR"), Some(3.5)); + } + + #[test] + fn symbol_price_series_test_constructor_sorts_unsorted_rows() { + let series = SymbolPriceSeries::new( + "000001.SZ".to_string(), + &[ + market_row("2025-01-06", 12.0, 300), + market_row("2025-01-02", 10.0, 100), + market_row("2025-01-03", 11.0, 200), + ], + ); + + assert!(series.dates.windows(2).all(|window| window[0] < window[1])); + assert_eq!(series.closes, vec![10.0, 11.0, 12.0]); + } + #[test] fn decision_volume_average_uses_previous_completed_days_only() { let series = SymbolPriceSeries::new(