优化日循环对象查找并加入 profiling 工具
This commit is contained in:
@@ -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 缓存和运行时逻辑视图。
|
`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/`。
|
||||||
|
|
||||||
本仓库只保留核心库构建和测试入口:
|
本仓库只保留核心库构建和测试入口:
|
||||||
|
|
||||||
## 测试与构建
|
## 测试与构建
|
||||||
|
|||||||
@@ -1314,6 +1314,8 @@ pub struct DataSet {
|
|||||||
factor_symbol_ids_by_date: Arc<BTreeMap<NaiveDate, Vec<u32>>>,
|
factor_symbol_ids_by_date: Arc<BTreeMap<NaiveDate, Vec<u32>>>,
|
||||||
factor_row_positions_by_date: Arc<Option<DenseRowPositionIndex>>,
|
factor_row_positions_by_date: Arc<Option<DenseRowPositionIndex>>,
|
||||||
factor_text_by_date: Arc<BTreeMap<NaiveDate, Vec<FactorTextValue>>>,
|
factor_text_by_date: Arc<BTreeMap<NaiveDate, Vec<FactorTextValue>>>,
|
||||||
|
factor_text_symbol_indices_by_date:
|
||||||
|
Arc<BTreeMap<NaiveDate, AHashMap<String, Vec<usize>>>>,
|
||||||
factor_text_index: Arc<HashMap<(NaiveDate, String, String), FactorTextValue>>,
|
factor_text_index: Arc<HashMap<(NaiveDate, String, String), FactorTextValue>>,
|
||||||
candidate_by_date: Arc<BTreeMap<NaiveDate, Vec<CandidateEligibility>>>,
|
candidate_by_date: Arc<BTreeMap<NaiveDate, Vec<CandidateEligibility>>>,
|
||||||
candidate_symbol_ids_by_date: Arc<BTreeMap<NaiveDate, Vec<u32>>>,
|
candidate_symbol_ids_by_date: Arc<BTreeMap<NaiveDate, Vec<u32>>>,
|
||||||
@@ -1711,6 +1713,19 @@ impl DataSet {
|
|||||||
})
|
})
|
||||||
.collect::<Vec<_>>();
|
.collect::<Vec<_>>();
|
||||||
let factor_text_by_date = group_by_date(factor_texts.clone(), |item| item.date);
|
let factor_text_by_date = group_by_date(factor_texts.clone(), |item| item.date);
|
||||||
|
let mut factor_text_symbol_indices_by_date =
|
||||||
|
BTreeMap::<NaiveDate, AHashMap<String, Vec<usize>>>::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
|
let factor_text_index = factor_texts
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(|item| ((item.date, item.symbol.clone(), item.field.clone()), item))
|
.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_symbol_ids_by_date: Arc::new(factor_symbol_ids_by_date),
|
||||||
factor_row_positions_by_date: Arc::new(factor_row_positions_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_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),
|
factor_text_index: Arc::new(factor_text_index),
|
||||||
candidate_by_date: Arc::new(candidate_by_date),
|
candidate_by_date: Arc::new(candidate_by_date),
|
||||||
candidate_symbol_ids_by_date: Arc::new(candidate_symbol_ids_by_date),
|
candidate_symbol_ids_by_date: Arc::new(candidate_symbol_ids_by_date),
|
||||||
@@ -2897,6 +2913,31 @@ impl DataSet {
|
|||||||
.unwrap_or(&[])
|
.unwrap_or(&[])
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn for_each_factor_text_row_for_symbol_on<F>(
|
||||||
|
&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> {
|
pub fn market_snapshots_on(&self, date: NaiveDate) -> Vec<&DailyMarketSnapshot> {
|
||||||
self.market_by_date
|
self.market_by_date
|
||||||
.get(&date)
|
.get(&date)
|
||||||
@@ -5909,4 +5950,50 @@ mod tests {
|
|||||||
Some((200.0 + 9_999.0) / 2.0)
|
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());
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -996,6 +996,7 @@ pub struct PlatformExprStrategy {
|
|||||||
stock_state_cache: RefCell<
|
stock_state_cache: RefCell<
|
||||||
AHashMap<(NaiveDate, NaiveDate, u32, Option<NaiveTime>, bool), Arc<StockExpressionState>>,
|
AHashMap<(NaiveDate, NaiveDate, u32, Option<NaiveTime>, bool), Arc<StockExpressionState>>,
|
||||||
>,
|
>,
|
||||||
|
normalized_universe_exclude: AHashSet<String>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, PartialEq)]
|
#[derive(Debug, Clone, PartialEq)]
|
||||||
@@ -1274,6 +1275,12 @@ impl PlatformExprStrategy {
|
|||||||
&normalized_stock_filter_expr,
|
&normalized_stock_filter_expr,
|
||||||
&prelude_declared_identifiers,
|
&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
|
let portfolio_drawdown_controller = config
|
||||||
.portfolio_drawdown_control
|
.portfolio_drawdown_control
|
||||||
.clone()
|
.clone()
|
||||||
@@ -1310,6 +1317,7 @@ impl PlatformExprStrategy {
|
|||||||
stock_text_factors_required,
|
stock_text_factors_required,
|
||||||
stock_state_cache_date: RefCell::new(None),
|
stock_state_cache_date: RefCell::new(None),
|
||||||
stock_state_cache: RefCell::new(AHashMap::new()),
|
stock_state_cache: RefCell::new(AHashMap::new()),
|
||||||
|
normalized_universe_exclude,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -3885,12 +3893,19 @@ impl PlatformExprStrategy {
|
|||||||
symbol: symbol.to_string(),
|
symbol: symbol.to_string(),
|
||||||
})
|
})
|
||||||
})?;
|
})?;
|
||||||
{
|
let reset_stock_state_cache = {
|
||||||
let mut cache_date = self.stock_state_cache_date.borrow_mut();
|
let mut cache_date = self.stock_state_cache_date.borrow_mut();
|
||||||
if *cache_date != Some(date) {
|
if *cache_date != Some(date) {
|
||||||
self.stock_state_cache.borrow_mut().clear();
|
|
||||||
*cache_date = Some(date);
|
*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 = (
|
let cache_key = (
|
||||||
date,
|
date,
|
||||||
@@ -4135,12 +4150,15 @@ impl PlatformExprStrategy {
|
|||||||
stock_volume_ma100,
|
stock_volume_ma100,
|
||||||
extra_factors,
|
extra_factors,
|
||||||
extra_text_factors: if self.stock_text_factors_required {
|
extra_text_factors: if self.stock_text_factors_required {
|
||||||
ctx.data
|
let mut values = BTreeMap::new();
|
||||||
.factor_text_rows_on(date)
|
ctx.data.for_each_factor_text_row_for_symbol_on(
|
||||||
.iter()
|
date,
|
||||||
.filter(|row| row.symbol == symbol)
|
symbol,
|
||||||
.map(|row| (row.field.clone(), row.value.clone()))
|
|row| {
|
||||||
.collect()
|
values.insert(row.field.clone(), row.value.clone());
|
||||||
|
},
|
||||||
|
);
|
||||||
|
values
|
||||||
} else {
|
} else {
|
||||||
BTreeMap::new()
|
BTreeMap::new()
|
||||||
},
|
},
|
||||||
@@ -8881,9 +8899,13 @@ impl PlatformExprStrategy {
|
|||||||
selection_risk_deferral: SelectionRiskDeferral,
|
selection_risk_deferral: SelectionRiskDeferral,
|
||||||
collect_risk_decisions: bool,
|
collect_risk_decisions: bool,
|
||||||
) -> (Vec<EligibleUniverseSnapshot>, Vec<FidcRiskDecisionAudit>) {
|
) -> (Vec<EligibleUniverseSnapshot>, Vec<FidcRiskDecisionAudit>) {
|
||||||
let mut rows = Vec::new();
|
|
||||||
let mut decisions = Vec::new();
|
|
||||||
let factor_rows = ctx.data.factor_snapshot_rows_on(factor_date);
|
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);
|
let factor_symbol_ids = ctx.data.factor_symbol_ids_on(factor_date);
|
||||||
debug_assert_eq!(factor_rows.len(), factor_symbol_ids.len());
|
debug_assert_eq!(factor_rows.len(), factor_symbol_ids.len());
|
||||||
for (factor, symbol_id) in factor_rows.iter().zip(factor_symbol_ids.iter().copied()) {
|
for (factor, symbol_id) in factor_rows.iter().zip(factor_symbol_ids.iter().copied()) {
|
||||||
@@ -9060,9 +9082,29 @@ impl PlatformExprStrategy {
|
|||||||
_candidate: &crate::data::CandidateEligibility,
|
_candidate: &crate::data::CandidateEligibility,
|
||||||
market: &DailyMarketSnapshot,
|
market: &DailyMarketSnapshot,
|
||||||
) -> bool {
|
) -> 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> {
|
fn universe_exclude_reason(excludes: &[String], symbol: &str) -> Option<&'static str> {
|
||||||
let normalized_symbol = symbol.trim().to_ascii_lowercase();
|
let normalized_symbol = symbol.trim().to_ascii_lowercase();
|
||||||
if normalized_symbol.is_empty() {
|
if normalized_symbol.is_empty() {
|
||||||
@@ -9370,9 +9412,7 @@ impl PlatformExprStrategy {
|
|||||||
let market = ctx.data.require_market(date, symbol)?;
|
let market = ctx.data.require_market(date, symbol)?;
|
||||||
let candidate = ctx.data.require_candidate(date, symbol)?;
|
let candidate = ctx.data.require_candidate(date, symbol)?;
|
||||||
|
|
||||||
if let Some(reason) =
|
if let Some(reason) = self.configured_universe_exclude_reason(&market.symbol) {
|
||||||
Self::universe_exclude_reason(&self.config.universe_exclude, &market.symbol)
|
|
||||||
{
|
|
||||||
return Ok(Some(reason.to_string()));
|
return Ok(Some(reason.to_string()));
|
||||||
}
|
}
|
||||||
let upper_limit_check_price =
|
let upper_limit_check_price =
|
||||||
|
|||||||
Executable
+97
@@ -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}"
|
||||||
Reference in New Issue
Block a user