zsb/fidc-backtest-engine
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

清理CSV回测demo入口

Browse Source
This commit is contained in:
boris
2026-07-03 23:40:33 +08:00
parent fea09ce93c
commit 73dd006bb2
18 changed files with 593 additions and 1711 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/fidc-core/src/data.rs
+1 -677
View File
@@ -1,6 +1,4 @@
use std::collections::{BTreeMap, HashMap, HashSet};
use std::fs;
use std::path::Path;
use std::sync::{Arc, OnceLock, RwLock};
use chrono::{NaiveDate, NaiveDateTime};
@@ -8,7 +6,7 @@ use serde::{Deserialize, Serialize};
use thiserror::Error;
use crate::calendar::TradingCalendar;
use crate::futures::{FuturesCommissionType, FuturesTradingParameter};
use crate::futures::FuturesTradingParameter;
use crate::instrument::Instrument;
use crate::risk_control::{ChinaAShareRiskControl, FidcRiskControlConfig};
@@ -58,18 +56,6 @@ mod datetime_format {
#[derive(Debug, Error)]
pub enum DataSetError {
#[error("failed to read file {path}: {source}")]
Io {
path: String,
#[source]
source: std::io::Error,
},
#[error("invalid csv row in {path} at line {line}: {message}")]
InvalidRow {
path: String,
line: usize,
message: String,
},
#[error("benchmark file contains multiple benchmark codes")]
MultipleBenchmarks,
#[error("missing data for {kind} on {date} / {symbol}")]
@@ -1006,96 +992,6 @@ pub struct DataSet {
}
impl DataSet {
pub fn from_csv_dir(path: &Path) -> Result<Self, DataSetError> {
let instruments = read_instruments(&path.join("instruments.csv"))?;
let market = read_market(&path.join("market.csv"))?;
let factors = read_factors(&path.join("factors.csv"))?;
let factor_texts = read_factor_texts(&path.join("factors.csv"))?;
let candidates = read_candidates(&path.join("candidate_flags.csv"))?;
let benchmarks = read_benchmarks(&path.join("benchmark.csv"))?;
let corporate_actions_path = path.join("corporate_actions.csv");
let corporate_actions = if corporate_actions_path.exists() {
read_corporate_actions(&corporate_actions_path)?
} else {
Vec::new()
};
let execution_quotes_path = path.join("execution_quotes.csv");
let execution_quotes = if execution_quotes_path.exists() {
read_execution_quotes(&execution_quotes_path)?
} else {
Vec::new()
};
let futures_params_path = path.join("futures_trading_parameters.csv");
let futures_params = if futures_params_path.exists() {
read_futures_trading_parameters(&futures_params_path)?
} else {
Vec::new()
};
let order_book_depth_path = path.join("order_book_depth.csv");
let order_book_depth = if order_book_depth_path.exists() {
read_order_book_depth(&order_book_depth_path)?
} else {
Vec::new()
};
Self::from_components_with_actions_quotes_futures_depth_and_factor_texts(
instruments,
market,
factors,
candidates,
benchmarks,
corporate_actions,
execution_quotes,
futures_params,
order_book_depth,
factor_texts,
)
}
pub fn from_partitioned_dir(path: &Path) -> Result<Self, DataSetError> {
let instruments = read_instruments(&path.join("instruments.csv"))?;
let benchmarks = read_partitioned_dir(&path.join("benchmark"), read_benchmarks)?;
let market = read_partitioned_dir(&path.join("market"), read_market)?;
let factors = read_partitioned_dir(&path.join("factors"), read_factors)?;
let factor_texts = read_partitioned_dir(&path.join("factors"), read_factor_texts)?;
let candidates = read_partitioned_dir(&path.join("candidates"), read_candidates)?;
let corporate_actions_dir = path.join("corporate_actions");
let corporate_actions = if corporate_actions_dir.exists() {
read_partitioned_dir(&corporate_actions_dir, read_corporate_actions)?
} else {
Vec::new()
};
let execution_quotes_dir = path.join("execution_quotes");
let execution_quotes = if execution_quotes_dir.exists() {
read_partitioned_dir(&execution_quotes_dir, read_execution_quotes)?
} else {
Vec::new()
};
let futures_params_dir = path.join("futures_trading_parameters");
let futures_params = if futures_params_dir.exists() {
read_partitioned_dir(&futures_params_dir, read_futures_trading_parameters)?
} else {
Vec::new()
};
let order_book_depth_dir = path.join("order_book_depth");
let order_book_depth = if order_book_depth_dir.exists() {
read_partitioned_dir(&order_book_depth_dir, read_order_book_depth)?
} else {
Vec::new()
};
Self::from_components_with_actions_quotes_futures_depth_and_factor_texts(
instruments,
market,
factors,
candidates,
benchmarks,
corporate_actions,
execution_quotes,
futures_params,
order_book_depth,
factor_texts,
)
}
pub fn from_components(
instruments: Vec<Instrument>,
market: Vec<DailyMarketSnapshot>,
@@ -2665,157 +2561,6 @@ impl DataSet {
}
}
fn read_instruments(path: &Path) -> Result<Vec<Instrument>, DataSetError> {
let rows = read_rows(path)?;
let mut instruments = Vec::new();
for row in rows {
instruments.push(Instrument {
symbol: row.get(0)?.to_string(),
name: row.get(1)?.to_string(),
board: row.get(2)?.to_string(),
round_lot: row.parse_optional_u32(3).unwrap_or(100),
listed_at: row.parse_optional_date(4)?,
delisted_at: row.parse_optional_date(5)?,
status: row
.fields
.get(6)
.map(|value| value.trim())
.filter(|value| !value.is_empty())
.unwrap_or("active")
.to_string(),
});
}
Ok(instruments)
}
fn read_market(path: &Path) -> Result<Vec<DailyMarketSnapshot>, DataSetError> {
let rows = read_rows(path)?;
let mut snapshots = Vec::new();
for row in rows {
let open = row.parse_f64(2)?;
let close = row.parse_f64(5)?;
let prev_close = row.parse_f64(6)?;
let price_tick = row.parse_optional_f64(15).unwrap_or(0.01);
let derived_upper_limit = round_price_to_tick(prev_close * 1.10, price_tick);
let derived_lower_limit = round_price_to_tick(prev_close * 0.90, price_tick);
snapshots.push(DailyMarketSnapshot {
date: row.parse_date(0)?,
symbol: row.get(1)?.to_string(),
timestamp: row
.fields
.get(16)
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty()),
day_open: row.parse_optional_f64(11).unwrap_or(open),
open,
high: row.parse_f64(3)?,
low: row.parse_f64(4)?,
close,
last_price: row.parse_optional_f64(12).unwrap_or(close),
bid1: row.parse_optional_f64(13).unwrap_or(close),
ask1: row.parse_optional_f64(14).unwrap_or(close),
prev_close,
volume: row.parse_u64(7)?,
minute_volume: row.parse_optional_u64(17).unwrap_or_default(),
bid1_volume: row.parse_optional_u64(18).unwrap_or_default(),
ask1_volume: row.parse_optional_u64(19).unwrap_or_default(),
trading_phase: row
.fields
.get(20)
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty()),
paused: row.parse_bool(8)?,
upper_limit: row.parse_optional_f64(9).unwrap_or(derived_upper_limit),
lower_limit: row.parse_optional_f64(10).unwrap_or(derived_lower_limit),
price_tick,
});
}
Ok(snapshots)
}
fn read_factors(path: &Path) -> Result<Vec<DailyFactorSnapshot>, DataSetError> {
let rows = read_rows(path)?;
let mut snapshots = Vec::new();
for row in rows {
let (extra_factors, _) = parse_extra_factor_maps(&row);
snapshots.push(DailyFactorSnapshot {
date: row.parse_date(0)?,
symbol: row.get(1)?.to_string(),
market_cap_bn: row.parse_f64(2)?,
free_float_cap_bn: row.parse_f64(3)?,
pe_ttm: row.parse_f64(4)?,
turnover_ratio: row.parse_optional_f64(5),
effective_turnover_ratio: row.parse_optional_f64(6),
extra_factors,
});
}
Ok(snapshots)
}
fn read_factor_texts(path: &Path) -> Result<Vec<FactorTextValue>, DataSetError> {
let rows = read_rows(path)?;
let mut text_values = Vec::new();
for row in rows {
let date = row.parse_date(0)?;
let symbol = row.get(1)?.to_string();
let (_, extra_text_factors) = parse_extra_factor_maps(&row);
for (field, value) in extra_text_factors {
text_values.push(FactorTextValue {
date,
symbol: symbol.clone(),
field,
value,
});
}
}
Ok(text_values)
}
fn parse_extra_factor_maps(row: &CsvRow) -> (BTreeMap<String, f64>, BTreeMap<String, String>) {
let mut numeric = BTreeMap::new();
let mut text = BTreeMap::new();
for value in row.fields.get(7).into_iter().chain(row.fields.get(8)) {
merge_extra_factor_json(value, &mut numeric, &mut text);
}
(numeric, text)
}
fn merge_extra_factor_json(
raw: &str,
numeric: &mut BTreeMap<String, f64>,
text: &mut BTreeMap<String, String>,
) {
let trimmed = raw.trim();
if trimmed.is_empty() {
return;
}
let Ok(serde_json::Value::Object(map)) = serde_json::from_str::<serde_json::Value>(trimmed)
else {
return;
};
for (key, value) in map {
let key = normalize_field(&key);
if key.is_empty() {
continue;
}
match value {
serde_json::Value::Number(number) => {
if let Some(value) = number.as_f64().filter(|value| value.is_finite()) {
numeric.insert(key, value);
}
}
serde_json::Value::String(value) => {
text.insert(key, value);
}
serde_json::Value::Bool(value) => {
numeric.insert(key.clone(), if value { 1.0 } else { 0.0 });
text.insert(key, value.to_string());
}
_ => {}
}
}
}
fn normalized_aliases(values: &[String]) -> Vec<String> {
let mut aliases = Vec::new();
for value in values {
@@ -3123,374 +2868,6 @@ fn take_last<T>(mut rows: Vec<T>, count: usize) -> Vec<T> {
rows.split_off(rows.len() - count)
}
fn read_candidates(path: &Path) -> Result<Vec<CandidateEligibility>, DataSetError> {
let rows = read_rows(path)?;
let mut snapshots = Vec::new();
for row in rows {
snapshots.push(CandidateEligibility {
date: row.parse_date(0)?,
symbol: row.get(1)?.to_string(),
is_st: row.parse_bool(2)?,
is_star_st: row.parse_bool(3)?,
is_new_listing: row.parse_bool(4)?,
is_paused: row.parse_bool(5)?,
allow_buy: row.parse_bool(6)?,
allow_sell: row.parse_bool(7)?,
is_kcb: row.parse_optional_bool(8).unwrap_or(false),
is_one_yuan: row.parse_optional_bool(9).unwrap_or(false),
risk_level_code: row
.fields
.get(10)
.map(String::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned),
});
}
Ok(snapshots)
}
fn read_benchmarks(path: &Path) -> Result<Vec<BenchmarkSnapshot>, DataSetError> {
let rows = read_rows(path)?;
let mut snapshots = Vec::new();
for row in rows {
snapshots.push(BenchmarkSnapshot {
date: row.parse_date(0)?,
benchmark: row.get(1)?.to_string(),
open: row.parse_f64(2)?,
close: row.parse_f64(3)?,
prev_close: row.parse_f64(4)?,
volume: row.parse_u64(5)?,
});
}
Ok(snapshots)
}
fn read_corporate_actions(path: &Path) -> Result<Vec<CorporateAction>, DataSetError> {
let rows = read_rows(path)?;
let mut snapshots = Vec::new();
for row in rows {
let has_payable_date = row.fields.len() >= 10;
let payable_date = if has_payable_date {
row.parse_optional_date(2)?
} else {
None
};
let offset = if has_payable_date { 1 } else { 0 };
snapshots.push(CorporateAction {
date: row.parse_date(0)?,
symbol: row.get(1)?.to_string(),
payable_date,
share_cash: row.parse_optional_f64(2 + offset).unwrap_or(0.0),
share_bonus: row.parse_optional_f64(3 + offset).unwrap_or(0.0),
share_gift: row.parse_optional_f64(4 + offset).unwrap_or(0.0),
issue_quantity: row.parse_optional_f64(5 + offset).unwrap_or(0.0),
issue_price: row.parse_optional_f64(6 + offset).unwrap_or(0.0),
reform: row.parse_optional_bool(7 + offset).unwrap_or(false),
adjust_factor: row.parse_optional_f64(8 + offset),
successor_symbol: row
.fields
.get(9 + offset)
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty()),
successor_ratio: row.parse_optional_f64(10 + offset),
successor_cash: row.parse_optional_f64(11 + offset),
});
}
Ok(snapshots)
}
fn read_execution_quotes(path: &Path) -> Result<Vec<IntradayExecutionQuote>, DataSetError> {
let rows = read_rows(path)?;
let mut quotes = Vec::new();
for row in rows {
quotes.push(IntradayExecutionQuote {
date: row.parse_date(0)?,
symbol: row.get(1)?.to_string(),
timestamp: row.parse_datetime(2)?,
last_price: row.parse_optional_f64(3).unwrap_or_default(),
bid1: row.parse_optional_f64(4).unwrap_or_default(),
ask1: row.parse_optional_f64(5).unwrap_or_default(),
bid1_volume: row.parse_optional_u64(6).unwrap_or_default(),
ask1_volume: row.parse_optional_u64(7).unwrap_or_default(),
volume_delta: row.parse_optional_u64(8).unwrap_or_default(),
amount_delta: row.parse_optional_f64(9).unwrap_or_default(),
trading_phase: row
.fields
.get(10)
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty()),
});
}
Ok(quotes)
}
fn read_order_book_depth(path: &Path) -> Result<Vec<IntradayOrderBookDepthLevel>, DataSetError> {
let rows = read_rows(path)?;
let mut levels = Vec::new();
for row in rows {
levels.push(IntradayOrderBookDepthLevel {
date: row.parse_date(0)?,
symbol: row.get(1)?.to_string(),
timestamp: row.parse_datetime(2)?,
level: row
.parse_optional_u32(3)
.unwrap_or(1)
.clamp(1, u8::MAX as u32) as u8,
bid_price: row.parse_optional_f64(4).unwrap_or_default(),
bid_volume: row.parse_optional_u64(5).unwrap_or_default(),
ask_price: row.parse_optional_f64(6).unwrap_or_default(),
ask_volume: row.parse_optional_u64(7).unwrap_or_default(),
});
}
Ok(levels)
}
fn read_futures_trading_parameters(
path: &Path,
) -> Result<Vec<FuturesTradingParameter>, DataSetError> {
let rows = read_rows(path)?;
let mut params = Vec::new();
for row in rows {
let first = row.get(0)?.trim();
let (effective_date, symbol_index) = if NaiveDate::parse_from_str(first, "%Y-%m-%d").is_ok()
{
(row.parse_optional_date(0)?, 1)
} else {
(None, 0)
};
params.push(FuturesTradingParameter {
effective_date,
symbol: row.get(symbol_index)?.to_string(),
contract_multiplier: row.parse_optional_f64(symbol_index + 1).unwrap_or(1.0),
long_margin_rate: row.parse_optional_f64(symbol_index + 2).unwrap_or(0.0),
short_margin_rate: row.parse_optional_f64(symbol_index + 3).unwrap_or(0.0),
commission_type: row
.fields
.get(symbol_index + 4)
.map(|value| FuturesCommissionType::parse(value))
.unwrap_or(FuturesCommissionType::ByMoney),
open_commission_ratio: row.parse_optional_f64(symbol_index + 5).unwrap_or(0.0),
close_commission_ratio: row.parse_optional_f64(symbol_index + 6).unwrap_or(0.0),
close_today_commission_ratio: row
.parse_optional_f64(symbol_index + 7)
.unwrap_or_else(|| row.parse_optional_f64(symbol_index + 6).unwrap_or(0.0)),
price_tick: row.parse_optional_f64(symbol_index + 8).unwrap_or(1.0),
});
}
Ok(params)
}
struct CsvRow {
path: String,
line: usize,
fields: Vec<String>,
}
impl CsvRow {
fn get(&self, index: usize) -> Result<&str, DataSetError> {
self.fields
.get(index)
.map(String::as_str)
.ok_or_else(|| DataSetError::InvalidRow {
path: self.path.clone(),
line: self.line,
message: format!("missing column {index}"),
})
}
fn parse_date(&self, index: usize) -> Result<NaiveDate, DataSetError> {
NaiveDate::parse_from_str(self.get(index)?, "%Y-%m-%d").map_err(|err| {
DataSetError::InvalidRow {
path: self.path.clone(),
line: self.line,
message: format!("invalid date: {err}"),
}
})
}
fn parse_f64(&self, index: usize) -> Result<f64, DataSetError> {
self.get(index)?
.parse::<f64>()
.map_err(|err| DataSetError::InvalidRow {
path: self.path.clone(),
line: self.line,
message: format!("invalid f64: {err}"),
})
}
fn parse_u64(&self, index: usize) -> Result<u64, DataSetError> {
self.get(index)?
.parse::<u64>()
.map_err(|err| DataSetError::InvalidRow {
path: self.path.clone(),
line: self.line,
message: format!("invalid u64: {err}"),
})
}
fn parse_optional_f64(&self, index: usize) -> Option<f64> {
self.fields.get(index).and_then(|value| {
let trimmed = value.trim();
if trimmed.is_empty() {
None
} else {
trimmed.parse::<f64>().ok()
}
})
}
fn parse_bool(&self, index: usize) -> Result<bool, DataSetError> {
self.get(index)?
.parse::<bool>()
.map_err(|err| DataSetError::InvalidRow {
path: self.path.clone(),
line: self.line,
message: format!("invalid bool: {err}"),
})
}
fn parse_optional_bool(&self, index: usize) -> Option<bool> {
self.fields
.get(index)
.and_then(|value| value.parse::<bool>().ok())
}
fn parse_optional_date(&self, index: usize) -> Result<Option<NaiveDate>, DataSetError> {
let Some(value) = self.fields.get(index) else {
return Ok(None);
};
let trimmed = value.trim();
if trimmed.is_empty() {
return Ok(None);
}
NaiveDate::parse_from_str(trimmed, "%Y-%m-%d")
.map(Some)
.map_err(|err| DataSetError::InvalidRow {
path: self.path.clone(),
line: self.line,
message: format!("invalid optional date: {err}"),
})
}
fn parse_datetime(&self, index: usize) -> Result<NaiveDateTime, DataSetError> {
NaiveDateTime::parse_from_str(self.get(index)?, "%Y-%m-%d %H:%M:%S").map_err(|err| {
DataSetError::InvalidRow {
path: self.path.clone(),
line: self.line,
message: format!("invalid datetime: {err}"),
}
})
}
fn parse_optional_u32(&self, index: usize) -> Option<u32> {
self.fields.get(index).and_then(|value| {
let trimmed = value.trim();
if trimmed.is_empty() {
None
} else {
trimmed.parse::<u32>().ok()
}
})
}
fn parse_optional_u64(&self, index: usize) -> Option<u64> {
self.fields.get(index).and_then(|value| {
let trimmed = value.trim();
if trimmed.is_empty() {
None
} else {
trimmed.parse::<u64>().ok()
}
})
}
}
fn read_partitioned_dir<T, F>(dir: &Path, mut loader: F) -> Result<Vec<T>, DataSetError>
where
F: FnMut(&Path) -> Result<Vec<T>, DataSetError>,
{
let mut rows = Vec::new();
let mut stack = vec![dir.to_path_buf()];
while let Some(current_dir) = stack.pop() {
let mut entries = fs::read_dir(&current_dir)
.map_err(|source| DataSetError::Io {
path: current_dir.display().to_string(),
source,
})?
.collect::<Result<Vec<_>, _>>()
.map_err(|source| DataSetError::Io {
path: current_dir.display().to_string(),
source,
})?;
entries.sort_by_key(|entry| entry.path());
for entry in entries.into_iter().rev() {
let path = entry.path();
if path.is_dir() {
stack.push(path);
continue;
}
if path.extension().and_then(|x| x.to_str()) != Some("csv") {
continue;
}
rows.extend(loader(&path)?);
}
}
Ok(rows)
}
fn read_rows(path: &Path) -> Result<Vec<CsvRow>, DataSetError> {
let content = fs::read_to_string(path).map_err(|source| DataSetError::Io {
path: path.display().to_string(),
source,
})?;
let mut rows = Vec::new();
for (line_idx, line) in content.lines().enumerate() {
let line_no = line_idx + 1;
if line_no == 1 || line.trim().is_empty() {
continue;
}
rows.push(CsvRow {
path: path.display().to_string(),
line: line_no,
fields: split_csv_line(line),
});
}
Ok(rows)
}
fn split_csv_line(line: &str) -> Vec<String> {
let mut fields = Vec::new();
let mut field = String::new();
let mut chars = line.trim_start_matches('\u{feff}').chars().peekable();
let mut in_quotes = false;
while let Some(ch) = chars.next() {
match ch {
'"' if in_quotes && chars.peek() == Some(&'"') => {
field.push('"');
chars.next();
}
'"' => {
in_quotes = !in_quotes;
}
',' if !in_quotes => {
fields.push(field.trim().to_string());
field.clear();
}
_ => field.push(ch),
}
}
fields.push(field.trim().to_string());
fields
}
fn group_by_date<T, F>(rows: Vec<T>, mut date_of: F) -> BTreeMap<NaiveDate, Vec<T>>
where
F: FnMut(&T) -> NaiveDate,
@@ -3549,15 +2926,6 @@ fn collect_benchmark_code(benchmarks: &[BenchmarkSnapshot]) -> Result<String, Da
}
}
fn round_price_to_tick(value: f64, tick: f64) -> f64 {
let effective_tick = if tick.is_finite() && tick > 0.0 {
tick
} else {
0.01
};
((value / effective_tick).round() * effective_tick * 10000.0).round() / 10000.0
}
fn prefix_sums(values: &[f64]) -> Vec<f64> {
let mut prefix = Vec::with_capacity(values.len() + 1);
prefix.push(0.0);
@@ -3770,15 +3138,6 @@ fn instrument_passes_baseline_selection(instrument: Option<&Instrument>, date: N
#[cfg(test)]
mod tests {
use super::*;
use std::time::{SystemTime, UNIX_EPOCH};
fn temp_csv_path(name: &str) -> std::path::PathBuf {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
std::env::temp_dir().join(format!("{}_{}_{}.csv", name, std::process::id(), nanos))
}
fn market_row(date: &str, prev_close: f64, volume: u64) -> DailyMarketSnapshot {
DailyMarketSnapshot {
@@ -4193,39 +3552,4 @@ mod tests {
Some((200.0 + 9_999.0) / 2.0)
);
}
#[test]
fn reads_mixed_numeric_and_text_extra_factors_from_quoted_csv_json() {
let path = temp_csv_path("mixed_factor_maps");
fs::write(
&path,
concat!(
"date,symbol,market_cap_bn,free_float_cap_bn,pe_ttm,turnover_ratio,effective_turnover_ratio,extra_factors\n",
"2025-01-02,000001.SZ,12,10,8,1,1,\"{\"\"custom_alpha\"\":7,\"\"industry_name\"\":\"\"electronics,hardware\"\",\"\"flag\"\":true}\"\n"
),
)
.unwrap();
let factors = read_factors(&path).unwrap();
let text_factors = read_factor_texts(&path).unwrap();
fs::remove_file(&path).ok();
assert_eq!(factors.len(), 1);
assert_eq!(
factors[0].extra_factors.get("custom_alpha").copied(),
Some(7.0)
);
assert_eq!(factors[0].extra_factors.get("flag").copied(), Some(1.0));
assert_eq!(text_factors.len(), 2);
assert!(
text_factors
.iter()
.any(|row| row.field == "industry_name" && row.value == "electronics,hardware")
);
assert!(
text_factors
.iter()
.any(|row| row.field == "flag" && row.value == "true")
);
}
}