diff --git a/README.md b/README.md index 0f684cd..a67a6bb 100644 --- a/README.md +++ b/README.md @@ -105,6 +105,8 @@ Source Lake 日线成交量保留原始可用性合同：源 `volume=null` 与 `fidc-backtest-engine` 不再维护本地 CSV demo、partitioned snapshot 目录或导出融合表作为运行入口。生产和集成回测由 `fidc-backtest-service` runner 创建 `DataSet`，数据来自 Strategy Factory Source Lake 的 Arrow/Parquet、manifest/data_epoch 缓存和运行时逻辑视图。 +日循环 CPU profiling 使用 `scripts/profile-daily-loop.sh` 只读附着到已运行的 runner，生成 `perf` 报告、分配器采样摘要和可选 SVG 火焰图。该脚本不会启动、停止或重启服务；`malloc/cfree` 仅是 CPU 采样代理，不能替代隔离环境中的精确 heap allocation 统计。当前 177 基线证据位于 `/Users/boris/WorkSpace/docs/fidc/evidence/engine-profile-20260829/`。 + 本仓库只保留核心库构建和测试入口： ## 测试与构建 diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index bba85e9..153e97a 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -1314,6 +1314,8 @@ pub struct DataSet { factor_symbol_ids_by_date: Arc>>, factor_row_positions_by_date: Arc>, factor_text_by_date: Arc>>, + factor_text_symbol_indices_by_date: + Arc>>>, factor_text_index: Arc>, candidate_by_date: Arc>>, candidate_symbol_ids_by_date: Arc>>, @@ -1711,6 +1713,19 @@ impl DataSet { }) .collect::>(); let factor_text_by_date = group_by_date(factor_texts.clone(), |item| item.date); + let mut factor_text_symbol_indices_by_date = + BTreeMap::>>::new(); + for (date, rows) in &factor_text_by_date { + let by_symbol = factor_text_symbol_indices_by_date + .entry(*date) + .or_default(); + for (index, row) in rows.iter().enumerate() { + by_symbol + .entry(row.symbol.clone()) + .or_default() + .push(index); + } + } let factor_text_index = factor_texts .into_iter() .map(|item| ((item.date, item.symbol.clone(), item.field.clone()), item)) @@ -1781,6 +1796,7 @@ impl DataSet { factor_symbol_ids_by_date: Arc::new(factor_symbol_ids_by_date), factor_row_positions_by_date: Arc::new(factor_row_positions_by_date), factor_text_by_date: Arc::new(factor_text_by_date), + factor_text_symbol_indices_by_date: Arc::new(factor_text_symbol_indices_by_date), factor_text_index: Arc::new(factor_text_index), candidate_by_date: Arc::new(candidate_by_date), candidate_symbol_ids_by_date: Arc::new(candidate_symbol_ids_by_date), @@ -2897,6 +2913,31 @@ impl DataSet { .unwrap_or(&[]) } + pub fn for_each_factor_text_row_for_symbol_on( + &self, + date: NaiveDate, + symbol: &str, + mut visit: F, + ) where + F: FnMut(&FactorTextValue), + { + let Some(rows) = self.factor_text_by_date.get(&date) else { + return; + }; + let Some(indices) = self + .factor_text_symbol_indices_by_date + .get(&date) + .and_then(|by_symbol| by_symbol.get(symbol)) + else { + return; + }; + for index in indices { + if let Some(row) = rows.get(*index) { + visit(row); + } + } + } + pub fn market_snapshots_on(&self, date: NaiveDate) -> Vec<&DailyMarketSnapshot> { self.market_by_date .get(&date) @@ -4282,7 +4323,7 @@ fn build_fundamental_universe_for_date( free_float_cap_bn: decision_free_float_cap_bn(factor), }); } - rows.sort_by(|left, right| { + rows.sort_unstable_by(|left, right| { left.market_cap_bn .partial_cmp(&right.market_cap_bn) .unwrap_or(std::cmp::Ordering::Equal) @@ -4365,7 +4406,7 @@ fn build_eligible_universe_for_date_from_factors( free_float_cap_bn, }); } - rows.sort_by(|left, right| { + rows.sort_unstable_by(|left, right| { left.market_cap_bn .partial_cmp(&right.market_cap_bn) .unwrap_or(std::cmp::Ordering::Equal) @@ -5909,4 +5950,50 @@ mod tests { Some((200.0 + 9_999.0) / 2.0) ); } + + #[test] + fn factor_text_symbol_index_matches_scan_semantics() { + let date = NaiveDate::parse_from_str("2025-01-02", "%Y-%m-%d").unwrap(); + let data = DataSet::from_components_with_factor_texts( + vec![Instrument { + symbol: "000001.SZ".to_string(), + name: "平安银行".to_string(), + board: "SZ".to_string(), + round_lot: 100, + listed_at: None, + delisted_at: None, + status: "active".to_string(), + }], + vec![market_row("2025-01-02", 10.0, 1_000_000)], + Vec::new(), + Vec::new(), + vec![benchmark_row("2025-01-02", 12.0)], + vec![ + FactorTextValue { + date, + symbol: "000001.SZ".to_string(), + field: "signal".to_string(), + value: "buy".to_string(), + }, + FactorTextValue { + date, + symbol: "000002.SZ".to_string(), + field: "signal".to_string(), + value: "hold".to_string(), + }, + ], + ) + .unwrap(); + + let mut values = Vec::new(); + data.for_each_factor_text_row_for_symbol_on(date, "000001.SZ", |row| { + values.push((row.field.clone(), row.value.clone())); + }); + assert_eq!(values, vec![("signal".to_string(), "buy".to_string())]); + let mut missing = Vec::new(); + data.for_each_factor_text_row_for_symbol_on(date, "000003.SZ", |row| { + missing.push(row.field.clone()); + }); + assert!(missing.is_empty()); + } } diff --git a/crates/fidc-core/src/platform_expr_strategy.rs b/crates/fidc-core/src/platform_expr_strategy.rs index 9ca70f8..21f6da6 100644 --- a/crates/fidc-core/src/platform_expr_strategy.rs +++ b/crates/fidc-core/src/platform_expr_strategy.rs @@ -996,6 +996,7 @@ pub struct PlatformExprStrategy { stock_state_cache: RefCell< AHashMap<(NaiveDate, NaiveDate, u32, Option, bool), Arc>, >, + normalized_universe_exclude: AHashSet, } #[derive(Debug, Clone, PartialEq)] @@ -1274,6 +1275,12 @@ impl PlatformExprStrategy { &normalized_stock_filter_expr, &prelude_declared_identifiers, ); + let normalized_universe_exclude = config + .universe_exclude + .iter() + .map(|item| item.trim().to_ascii_uppercase()) + .filter(|item| !item.is_empty()) + .collect(); let portfolio_drawdown_controller = config .portfolio_drawdown_control .clone() @@ -1310,6 +1317,7 @@ impl PlatformExprStrategy { stock_text_factors_required, stock_state_cache_date: RefCell::new(None), stock_state_cache: RefCell::new(AHashMap::new()), + normalized_universe_exclude, } } @@ -1340,6 +1348,7 @@ impl PlatformExprStrategy { /// This validates syntax only; identifiers and runtime values are resolved /// later against the point-in-time execution scope. pub fn validate_expression_syntax(&self) -> Result<(), BacktestError> { + self.validate_expression_value_kinds()?; let normalized_prelude = Self::normalize_prelude_for_eval(&self.config.prelude); let mut expressions = vec![ ( @@ -1557,6 +1566,88 @@ impl PlatformExprStrategy { Ok(()) } + fn trim_outer_parentheses_for_expression_literal(expression: &str) -> &str { + let bytes = expression.as_bytes(); + let mut start = 0usize; + let mut end = bytes.len(); + while start < end && bytes[start].is_ascii_whitespace() { + start += 1; + } + while end > start && bytes[end - 1].is_ascii_whitespace() { + end -= 1; + } + loop { + if end <= start + 1 || bytes[start] != b'(' || bytes[end - 1] != b')' { + break; + } + start += 1; + end -= 1; + while start < end && bytes[start].is_ascii_whitespace() { + start += 1; + } + while end > start && bytes[end - 1].is_ascii_whitespace() { + end -= 1; + } + } + &expression[start..end] + } + + fn is_single_string_literal_expression(expression: &str) -> bool { + let expression = Self::trim_outer_parentheses_for_expression_literal(expression).trim(); + let Some(quote) = expression.chars().next() else { + return false; + }; + if quote != '"' && quote != '\'' { + return false; + } + let mut escaped = false; + for (index, character) in expression.char_indices().skip(1) { + if escaped { + escaped = false; + continue; + } + if character == '\\' { + escaped = true; + continue; + } + if character == quote { + return expression[index + character.len_utf8()..].trim().is_empty(); + } + } + false + } + + fn validate_expression_value_kinds(&self) -> Result<(), BacktestError> { + if Self::is_single_string_literal_expression(&self.config.stock_filter_expr) { + return Err(BacktestError::Execution( + "field=stock_filter_expr must be a boolean expression; received a string literal" + .to_string(), + )); + } + for (field, expression) in [ + ("refresh_rate_expr", self.config.refresh_rate_expr.as_str()), + ("market_cap_lower_expr", self.config.market_cap_lower_expr.as_str()), + ("market_cap_upper_expr", self.config.market_cap_upper_expr.as_str()), + ("selection_limit_expr", self.config.selection_limit_expr.as_str()), + ( + "selection_candidate_limit_expr", + self.config.selection_candidate_limit_expr.as_str(), + ), + ("buy_scale_expr", self.config.buy_scale_expr.as_str()), + ("exposure_expr", self.config.exposure_expr.as_str()), + ("stop_loss_expr", self.config.stop_loss_expr.as_str()), + ("take_profit_expr", self.config.take_profit_expr.as_str()), + ("rank_expr", self.config.rank_expr.as_str()), + ] { + if Self::is_single_string_literal_expression(expression) { + return Err(BacktestError::Execution(format!( + "field={field} must be a numeric expression; received a string literal" + ))); + } + } + Ok(()) + } + /// 用 AST 缓存执行 script。命中：直接走 eval_ast_with_scope；未命中：先 /// engine.compile，再插入缓存，再 eval_ast_with_scope。任何编译/执行错误 /// 都按字符串包装为 BacktestError::Execution。 @@ -3885,12 +3976,19 @@ impl PlatformExprStrategy { symbol: symbol.to_string(), }) })?; - { + let reset_stock_state_cache = { let mut cache_date = self.stock_state_cache_date.borrow_mut(); if *cache_date != Some(date) { - self.stock_state_cache.borrow_mut().clear(); *cache_date = Some(date); + true + } else { + false } + }; + if reset_stock_state_cache { + let mut cache = self.stock_state_cache.borrow_mut(); + cache.clear(); + cache.reserve(ctx.data.factor_snapshot_rows_on(factor_date).len()); } let cache_key = ( date, @@ -4135,12 +4233,15 @@ impl PlatformExprStrategy { stock_volume_ma100, extra_factors, extra_text_factors: if self.stock_text_factors_required { - ctx.data - .factor_text_rows_on(date) - .iter() - .filter(|row| row.symbol == symbol) - .map(|row| (row.field.clone(), row.value.clone())) - .collect() + let mut values = BTreeMap::new(); + ctx.data.for_each_factor_text_row_for_symbol_on( + date, + symbol, + |row| { + values.insert(row.field.clone(), row.value.clone()); + }, + ); + values } else { BTreeMap::new() }, @@ -8881,9 +8982,13 @@ impl PlatformExprStrategy { selection_risk_deferral: SelectionRiskDeferral, collect_risk_decisions: bool, ) -> (Vec, Vec) { - let mut rows = Vec::new(); - let mut decisions = Vec::new(); let factor_rows = ctx.data.factor_snapshot_rows_on(factor_date); + let mut rows = Vec::with_capacity(factor_rows.len()); + let mut decisions = if collect_risk_decisions { + Vec::with_capacity(factor_rows.len()) + } else { + Vec::new() + }; let factor_symbol_ids = ctx.data.factor_symbol_ids_on(factor_date); debug_assert_eq!(factor_rows.len(), factor_symbol_ids.len()); for (factor, symbol_id) in factor_rows.iter().zip(factor_symbol_ids.iter().copied()) { @@ -8954,7 +9059,7 @@ impl PlatformExprStrategy { free_float_cap_bn, }); } - rows.sort_by(|left, right| { + rows.sort_unstable_by(|left, right| { left.market_cap_bn .partial_cmp(&right.market_cap_bn) .unwrap_or(std::cmp::Ordering::Equal) @@ -9060,9 +9165,29 @@ impl PlatformExprStrategy { _candidate: &crate::data::CandidateEligibility, market: &DailyMarketSnapshot, ) -> bool { - Self::universe_exclude_reason(&self.config.universe_exclude, &market.symbol).is_none() + self.configured_universe_exclude_reason(&market.symbol).is_none() } + fn configured_universe_exclude_reason(&self, symbol: &str) -> Option<&'static str> { + let normalized_symbol = symbol.trim().to_ascii_uppercase(); + if normalized_symbol.is_empty() { + return None; + } + let symbol_base = normalized_symbol.split('.').next().unwrap_or(""); + if self.normalized_universe_exclude.contains("BJSE") + && normalized_symbol.ends_with(".BJ") + { + return Some("bjse"); + } + if self.normalized_universe_exclude.contains(&normalized_symbol) + || (!symbol_base.is_empty() && self.normalized_universe_exclude.contains(symbol_base)) + { + return Some("universe_exclude"); + } + None + } + + #[cfg(test)] fn universe_exclude_reason(excludes: &[String], symbol: &str) -> Option<&'static str> { let normalized_symbol = symbol.trim().to_ascii_lowercase(); if normalized_symbol.is_empty() { @@ -9370,9 +9495,7 @@ impl PlatformExprStrategy { let market = ctx.data.require_market(date, symbol)?; let candidate = ctx.data.require_candidate(date, symbol)?; - if let Some(reason) = - Self::universe_exclude_reason(&self.config.universe_exclude, &market.symbol) - { + if let Some(reason) = self.configured_universe_exclude_reason(&market.symbol) { return Ok(Some(reason.to_string())); } let upper_limit_check_price = @@ -9522,7 +9645,7 @@ impl PlatformExprStrategy { candidates.push((candidate, stock, rank_value)); } if !self.rank_reuses_market_cap_order() { - candidates.sort_by(|lhs, rhs| { + candidates.sort_unstable_by(|lhs, rhs| { let lhs_value = lhs.2; let rhs_value = rhs.2; let ordering = if self.config.rank_desc { @@ -10164,7 +10287,7 @@ impl PlatformExprStrategy { } candidates.push((candidate.symbol.clone(), rank_value, stock)); } - candidates.sort_by(|lhs, rhs| { + candidates.sort_unstable_by(|lhs, rhs| { let ordering = if self.config.rank_desc { rhs.1 .partial_cmp(&lhs.1) diff --git a/scripts/profile-daily-loop.sh b/scripts/profile-daily-loop.sh new file mode 100755 index 0000000..c98fffe --- /dev/null +++ b/scripts/profile-daily-loop.sh @@ -0,0 +1,97 @@ +#!/usr/bin/env bash + +set -euo pipefail + +usage() { + cat <<'EOF' +Usage: profile-daily-loop.sh --pid PID [--seconds N] [--output-dir DIR] [--flamegraph-dir DIR] + +Attaches perf to an existing FIDC runner process. It never starts, stops, or +restarts a service. The allocation report is a CPU-sample proxy for allocator +pressure; it is not an allocation count or byte-accurate heap profile. +EOF +} + +pid="" +seconds=15 +output_dir="" +flamegraph_dir="" + +while [[ $# -gt 0 ]]; do + case "$1" in + --pid) + pid="${2:-}" + shift 2 + ;; + --seconds) + seconds="${2:-}" + shift 2 + ;; + --output-dir) + output_dir="${2:-}" + shift 2 + ;; + --flamegraph-dir) + flamegraph_dir="${2:-}" + shift 2 + ;; + -h|--help) + usage + exit 0 + ;; + *) + echo "unknown argument: $1" >&2 + usage >&2 + exit 2 + ;; + esac +done + +if [[ "${EUID}" -ne 0 ]]; then + echo "run as root so perf can attach to the runner" >&2 + exit 2 +fi +if [[ ! "${pid}" =~ ^[1-9][0-9]*$ ]]; then + echo "--pid must be a positive process id" >&2 + exit 2 +fi +if [[ ! "${seconds}" =~ ^[1-9][0-9]*$ ]]; then + echo "--seconds must be a positive integer" >&2 + exit 2 +fi +if ! kill -0 "${pid}" 2>/dev/null; then + echo "process is not running: ${pid}" >&2 + exit 1 +fi +if ! command -v perf >/dev/null 2>&1; then + echo "perf is required" >&2 + exit 2 +fi + +if [[ -z "${output_dir}" ]]; then + output_dir="$(pwd)/profile-$(date +%Y%m%d-%H%M%S)-${pid}" +fi +mkdir -p "${output_dir}" + +perf record --quiet --call-graph dwarf -g -e cycles:P -p "${pid}" \ + -o "${output_dir}/perf.data" -- sleep "${seconds}" +perf report --stdio --no-children --sort symbol --percent-limit 0.5 \ + -i "${output_dir}/perf.data" > "${output_dir}/perf-report.txt" + +grep -E 'malloc|calloc|realloc|free|cfree|drop_glue' "${output_dir}/perf-report.txt" \ + > "${output_dir}/allocation-samples.txt" || true + +if [[ -n "${flamegraph_dir}" ]]; then + if [[ ! -x "${flamegraph_dir}/stackcollapse-perf.pl" || ! -x "${flamegraph_dir}/flamegraph.pl" ]]; then + echo "flamegraph tools not found under ${flamegraph_dir}" >&2 + exit 2 + fi + perf script -i "${output_dir}/perf.data" > "${output_dir}/perf.script" + "${flamegraph_dir}/stackcollapse-perf.pl" "${output_dir}/perf.script" \ + > "${output_dir}/daily-loop.folded" + "${flamegraph_dir}/flamegraph.pl" \ + --title="FIDC engine daily loop CPU samples" \ + "${output_dir}/daily-loop.folded" > "${output_dir}/daily-loop-flamegraph.svg" +fi + +printf '%s\n' "profile complete: ${output_dir}"