00ae84dd88
Strip all 413 #[allow(dead_code)] annotations from 139 files and remove the actual dead code they were suppressing: unused struct fields (and their constructor sites), unused methods/functions, and entire dead structs. Key removals: - trading_engine compliance: ~50 dead structs/fields across audit, reporting, SOX modules - trading_service: dead execution engine fields, broker routing, paper trading methods - ml_training_service: dead TLS validation (~340 lines), GPU state, monitoring fields - backtesting_service: dead model cache, TLS validation, TradeSignal fields - risk: dead VaR engine fields, safety coordinator fields, position tracker fields - adaptive-strategy: dead ensemble methods, regime detection, sizing functions 147 files changed, -4264 net lines. Workspace compiles with 0 errors. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1255 lines
41 KiB
Rust
1255 lines
41 KiB
Rust
//! Historical Data Loader for ML Training
|
|
//!
|
|
//! This module provides the `HistoricalDataLoader` which loads training data from PostgreSQL
|
|
//! and converts it to `FinancialFeatures` for ML model training.
|
|
//!
|
|
//! ## Data Pipeline
|
|
//!
|
|
//! 1. **Load**: Query database tables (order_book_snapshots, trade_executions, market_events)
|
|
//! 2. **Filter**: Apply time range and symbol filters from configuration
|
|
//! 3. **Extract**: Compute technical indicators and microstructure features
|
|
//! 4. **Normalize**: Apply z-score, min-max, or robust scaling
|
|
//! 5. **Convert**: Transform to FinancialFeatures format
|
|
//!
|
|
//! ## Usage
|
|
//!
|
|
//! ```rust,no_run
|
|
//! use ml_training_service::data_loader::HistoricalDataLoader;
|
|
//! use ml_training_service::data_config::TrainingDataSourceConfig;
|
|
//!
|
|
//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
|
|
//! let config = TrainingDataSourceConfig::from_env()?;
|
|
//! let loader = HistoricalDataLoader::new(config).await?;
|
|
//! let (training_data, validation_data) = loader.load_training_data().await?;
|
|
//! # Ok(())
|
|
//! # }
|
|
//! ```
|
|
|
|
use anyhow::{Context, Result};
|
|
use chrono::{DateTime, Utc};
|
|
use sqlx::PgPool;
|
|
use std::collections::{HashMap, VecDeque};
|
|
use tracing::{debug, info, warn};
|
|
|
|
use crate::data_config::{DatabaseConfig, TrainingDataSourceConfig};
|
|
use crate::schema_types::{MarketEvent, OrderBookSnapshot, TradeExecution};
|
|
use crate::technical_indicators::{IndicatorConfig, TechnicalIndicatorCalculator};
|
|
use common::Price;
|
|
use ml::training_pipeline::{FinancialFeatures, MicrostructureFeatures, RiskFeatures};
|
|
|
|
/// Risk metrics calculator for portfolio risk assessment
|
|
///
|
|
/// Maintains rolling price history and calculates financial risk metrics:
|
|
/// - VaR (Value at Risk) at 5% confidence level
|
|
///
|
|
/// - Expected Shortfall (CVaR) - average loss beyond VaR
|
|
/// - Maximum Drawdown - largest peak-to-trough decline
|
|
///
|
|
/// - Sharpe Ratio - risk-adjusted return metric
|
|
///
|
|
/// Uses log returns for better statistical properties and annualizes
|
|
/// metrics assuming 252 trading days per year.
|
|
struct RiskMetricsCalculator {
|
|
/// Rolling window of historical prices
|
|
price_history: VecDeque<f64>,
|
|
|
|
/// Maximum window size for calculations
|
|
window_size: usize,
|
|
|
|
/// Risk-free rate for Sharpe ratio (annualized)
|
|
risk_free_rate: f64,
|
|
}
|
|
|
|
impl RiskMetricsCalculator {
|
|
/// Create new risk metrics calculator
|
|
///
|
|
/// # Arguments
|
|
/// * `window_size` - Number of historical prices to maintain (default: 100)
|
|
///
|
|
/// * `risk_free_rate` - Annual risk-free rate (default: 0.0)
|
|
fn new(window_size: usize, risk_free_rate: f64) -> Self {
|
|
Self {
|
|
price_history: VecDeque::with_capacity(window_size),
|
|
window_size,
|
|
risk_free_rate,
|
|
}
|
|
}
|
|
|
|
/// Update with new price observation
|
|
fn update(&mut self, price: f64) {
|
|
if !price.is_finite() || price <= 0.0 {
|
|
return; // Skip invalid prices
|
|
}
|
|
|
|
self.price_history.push_back(price);
|
|
|
|
// Maintain window size
|
|
if self.price_history.len() > self.window_size {
|
|
self.price_history.pop_front();
|
|
}
|
|
}
|
|
|
|
/// Calculate log returns from price history
|
|
///
|
|
/// Returns: Vector of log returns ln(P_t / P_{t-1})
|
|
fn calculate_log_returns(&self) -> Vec<f64> {
|
|
if self.price_history.len() < 2 {
|
|
return Vec::new();
|
|
}
|
|
|
|
self.price_history
|
|
.iter()
|
|
.zip(self.price_history.iter().skip(1))
|
|
.map(|(prev, curr)| (curr / prev).ln())
|
|
.filter(|r| r.is_finite())
|
|
.collect()
|
|
}
|
|
|
|
/// Calculate Value at Risk (VaR) at given confidence level
|
|
///
|
|
/// # Arguments
|
|
/// * `confidence` - Confidence level (e.g., 0.05 for 5% VaR)
|
|
///
|
|
/// Returns: VaR as negative percentage loss
|
|
fn calculate_var(&self, confidence: f64) -> f64 {
|
|
let mut returns = self.calculate_log_returns();
|
|
if returns.is_empty() {
|
|
return -0.02; // Default: -2% if insufficient data
|
|
}
|
|
|
|
returns.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
|
|
|
|
let index = (returns.len() as f64 * confidence).floor() as usize;
|
|
let index = index.min(returns.len().saturating_sub(1));
|
|
|
|
returns[index]
|
|
}
|
|
|
|
/// Calculate Expected Shortfall (CVaR) - average loss beyond VaR
|
|
///
|
|
/// # Arguments
|
|
/// * `var` - VaR threshold value
|
|
///
|
|
/// Returns: Expected shortfall as negative percentage
|
|
fn calculate_expected_shortfall(&self, var: f64) -> f64 {
|
|
let returns = self.calculate_log_returns();
|
|
if returns.is_empty() {
|
|
return -0.03; // Default: -3% if insufficient data
|
|
}
|
|
|
|
let tail_returns: Vec<f64> = returns.iter().filter(|&&r| r <= var).copied().collect();
|
|
|
|
if tail_returns.is_empty() {
|
|
return var; // If no tail, ES equals VaR
|
|
}
|
|
|
|
tail_returns.iter().sum::<f64>() / tail_returns.len() as f64
|
|
}
|
|
|
|
/// Calculate maximum drawdown from price history
|
|
///
|
|
/// Returns: Maximum peak-to-trough decline as negative percentage
|
|
fn calculate_max_drawdown(&self) -> f64 {
|
|
if self.price_history.len() < 2 {
|
|
return -0.05; // Default: -5% if insufficient data
|
|
}
|
|
|
|
let mut max_price = self.price_history.front().copied().unwrap_or(0.0);
|
|
let mut max_drawdown = 0.0;
|
|
|
|
for price in self.price_history.iter().skip(1) {
|
|
if *price > max_price {
|
|
max_price = *price;
|
|
} else {
|
|
let drawdown = (*price - max_price) / max_price;
|
|
if drawdown < max_drawdown {
|
|
max_drawdown = drawdown;
|
|
}
|
|
}
|
|
}
|
|
|
|
max_drawdown
|
|
}
|
|
|
|
/// Calculate Sharpe ratio - risk-adjusted return
|
|
///
|
|
/// Returns: Annualized Sharpe ratio (assumes 252 trading days)
|
|
fn calculate_sharpe_ratio(&self) -> f64 {
|
|
let returns = self.calculate_log_returns();
|
|
if returns.len() < 2 {
|
|
return 1.0; // Default: neutral Sharpe if insufficient data
|
|
}
|
|
|
|
let mean_return = returns.iter().sum::<f64>() / returns.len() as f64;
|
|
|
|
let variance = returns
|
|
.iter()
|
|
.map(|r| (r - mean_return).powi(2))
|
|
.sum::<f64>()
|
|
/ returns.len() as f64;
|
|
|
|
let std_dev = variance.sqrt();
|
|
|
|
if std_dev < 1e-10 {
|
|
return 0.0; // No volatility, undefined Sharpe
|
|
}
|
|
|
|
// Annualize: multiply mean by 252, std dev by sqrt(252)
|
|
let annualized_return = mean_return * 252.0;
|
|
let annualized_volatility = std_dev * (252.0_f64).sqrt();
|
|
|
|
(annualized_return - self.risk_free_rate) / annualized_volatility
|
|
}
|
|
|
|
/// Calculate all risk metrics at once
|
|
fn calculate_all_metrics(&self) -> RiskMetrics {
|
|
let var_5pct = self.calculate_var(0.05);
|
|
let expected_shortfall = self.calculate_expected_shortfall(var_5pct);
|
|
let max_drawdown = self.calculate_max_drawdown();
|
|
let sharpe_ratio = self.calculate_sharpe_ratio();
|
|
|
|
RiskMetrics {
|
|
var_5pct,
|
|
expected_shortfall,
|
|
max_drawdown,
|
|
sharpe_ratio,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Computed risk metrics
|
|
struct RiskMetrics {
|
|
var_5pct: f64,
|
|
expected_shortfall: f64,
|
|
max_drawdown: f64,
|
|
sharpe_ratio: f64,
|
|
}
|
|
|
|
/// Normalization method for feature scaling
|
|
#[derive(Debug, Clone)]
|
|
enum NormalizationMethod {
|
|
/// No normalization applied
|
|
None,
|
|
/// Z-score: (x - mean) / std_dev
|
|
ZScore,
|
|
/// Min-max: (x - min) / (max - min)
|
|
MinMax,
|
|
/// Robust: (x - median) / IQR
|
|
Robust,
|
|
}
|
|
|
|
impl NormalizationMethod {
|
|
fn from_str(s: &str) -> Self {
|
|
match s.to_lowercase().as_str() {
|
|
"zscore" | "z-score" => Self::ZScore,
|
|
"minmax" | "min-max" => Self::MinMax,
|
|
"robust" => Self::Robust,
|
|
_ => Self::None,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Normalization parameters for a single feature
|
|
#[derive(Debug, Clone)]
|
|
pub struct NormalizationParams {
|
|
pub mean: f64,
|
|
pub std_dev: f64,
|
|
pub min: f64,
|
|
pub max: f64,
|
|
pub median: f64,
|
|
pub q1: f64, // 25th percentile
|
|
pub q3: f64, // 75th percentile
|
|
}
|
|
|
|
/// Complete normalization parameters for all features
|
|
///
|
|
/// Used to prevent data leakage by fitting on training set and applying to validation set
|
|
#[derive(Debug, Clone)]
|
|
pub struct FeatureNormalizationParams {
|
|
pub indicator_params: HashMap<String, NormalizationParams>,
|
|
pub spread_params: NormalizationParams,
|
|
pub imbalance_params: NormalizationParams,
|
|
pub intensity_params: NormalizationParams,
|
|
pub var_params: NormalizationParams,
|
|
pub es_params: NormalizationParams,
|
|
pub dd_params: NormalizationParams,
|
|
pub sharpe_params: NormalizationParams,
|
|
}
|
|
|
|
impl NormalizationParams {
|
|
/// Fit normalization parameters from data
|
|
fn fit(values: &[f64]) -> Self {
|
|
if values.is_empty() {
|
|
return Self::default();
|
|
}
|
|
|
|
let valid_values: Vec<f64> = values.iter().filter(|v| v.is_finite()).copied().collect();
|
|
|
|
if valid_values.is_empty() {
|
|
return Self::default();
|
|
}
|
|
|
|
// Calculate mean and std dev
|
|
let mean = valid_values.iter().sum::<f64>() / valid_values.len() as f64;
|
|
let variance = valid_values.iter().map(|v| (v - mean).powi(2)).sum::<f64>()
|
|
/ valid_values.len() as f64;
|
|
let std_dev = variance.sqrt();
|
|
|
|
// Calculate min and max
|
|
let min = valid_values.iter().fold(f64::INFINITY, |a, &b| a.min(b));
|
|
let max = valid_values
|
|
.iter()
|
|
.fold(f64::NEG_INFINITY, |a, &b| a.max(b));
|
|
|
|
// Calculate median and quartiles
|
|
let mut sorted = valid_values;
|
|
sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
|
|
|
|
let median = Self::percentile(&sorted, 0.5);
|
|
let q1 = Self::percentile(&sorted, 0.25);
|
|
let q3 = Self::percentile(&sorted, 0.75);
|
|
|
|
Self {
|
|
mean,
|
|
std_dev,
|
|
min,
|
|
max,
|
|
median,
|
|
q1,
|
|
q3,
|
|
}
|
|
}
|
|
|
|
/// Calculate percentile from sorted data
|
|
fn percentile(sorted_data: &[f64], p: f64) -> f64 {
|
|
if sorted_data.is_empty() {
|
|
return 0.0;
|
|
}
|
|
|
|
let index = (sorted_data.len() as f64 * p).floor() as usize;
|
|
let index = index.min(sorted_data.len() - 1);
|
|
sorted_data[index]
|
|
}
|
|
|
|
/// Apply normalization to a value
|
|
fn normalize(&self, value: f64, method: &NormalizationMethod) -> f64 {
|
|
if !value.is_finite() {
|
|
return 0.0;
|
|
}
|
|
|
|
match method {
|
|
NormalizationMethod::None => value,
|
|
NormalizationMethod::ZScore => {
|
|
if self.std_dev < 1e-10 {
|
|
0.0 // No variance, return 0
|
|
} else {
|
|
(value - self.mean) / self.std_dev
|
|
}
|
|
},
|
|
NormalizationMethod::MinMax => {
|
|
let range = self.max - self.min;
|
|
if range < 1e-10 {
|
|
0.0 // No range, return 0
|
|
} else {
|
|
(value - self.min) / range
|
|
}
|
|
},
|
|
NormalizationMethod::Robust => {
|
|
let iqr = self.q3 - self.q1;
|
|
if iqr < 1e-10 {
|
|
0.0 // No IQR, return 0
|
|
} else {
|
|
(value - self.median) / iqr
|
|
}
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Default for NormalizationParams {
|
|
fn default() -> Self {
|
|
Self {
|
|
mean: 0.0,
|
|
std_dev: 1.0,
|
|
min: 0.0,
|
|
max: 1.0,
|
|
median: 0.0,
|
|
q1: 0.0,
|
|
q3: 1.0,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Historical data loader for ML training
|
|
///
|
|
/// Connects to Postgre`SQL` database and loads market data for ML model training.
|
|
///
|
|
/// Handles time-range filtering, symbol filtering, and feature extraction.
|
|
pub struct HistoricalDataLoader {
|
|
/// Database connection pool
|
|
pool: PgPool,
|
|
|
|
/// Data source configuration
|
|
config: TrainingDataSourceConfig,
|
|
|
|
/// Per-symbol technical indicator calculators
|
|
calculators: HashMap<String, TechnicalIndicatorCalculator>,
|
|
|
|
/// Per-symbol risk metrics calculators
|
|
risk_calculators: HashMap<String, RiskMetricsCalculator>,
|
|
}
|
|
|
|
impl HistoricalDataLoader {
|
|
/// Create a new historical data loader
|
|
///
|
|
/// Establishes database connection pool based on configuration.
|
|
///
|
|
/// # Arguments
|
|
///
|
|
/// * `config` - Training data source configuration with database settings
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Returns error if:
|
|
/// - Database configuration is missing
|
|
///
|
|
/// - Database connection fails
|
|
/// - Connection pool cannot be created
|
|
pub async fn new(config: TrainingDataSourceConfig) -> Result<Self> {
|
|
let database_config = config.database.as_ref().ok_or_else(|| {
|
|
anyhow::anyhow!("Database configuration required for historical data")
|
|
})?;
|
|
|
|
info!(
|
|
"Connecting to database: {}",
|
|
Self::sanitize_connection_url(&database_config.connection_url)
|
|
);
|
|
|
|
let pool = Self::create_connection_pool(database_config).await?;
|
|
|
|
info!("Database connection established successfully");
|
|
|
|
Ok(Self {
|
|
pool,
|
|
config,
|
|
calculators: HashMap::new(),
|
|
risk_calculators: HashMap::new(),
|
|
})
|
|
}
|
|
|
|
/// Create database connection pool with configuration
|
|
async fn create_connection_pool(database_config: &DatabaseConfig) -> Result<PgPool> {
|
|
let pool = sqlx::postgres::PgPoolOptions::new()
|
|
.max_connections(database_config.max_connections)
|
|
.acquire_timeout(std::time::Duration::from_secs(
|
|
database_config.query_timeout_secs,
|
|
))
|
|
.connect(&database_config.connection_url)
|
|
.await
|
|
.context("Failed to create database connection pool")?;
|
|
|
|
Ok(pool)
|
|
}
|
|
|
|
/// Sanitize connection URL for logging (hide password)
|
|
fn sanitize_connection_url(url: &str) -> String {
|
|
if let Some(at_pos) = url.find('@') {
|
|
if let Some(colon_pos) = url[..at_pos].rfind(':') {
|
|
let mut sanitized = url.to_string();
|
|
sanitized.replace_range(colon_pos + 1..at_pos, "****");
|
|
return sanitized;
|
|
}
|
|
}
|
|
url.to_string()
|
|
}
|
|
|
|
/// Load training data from database
|
|
///
|
|
/// Main entry point for data loading. Queries database tables, extracts features,
|
|
/// and splits into training/validation sets.
|
|
///
|
|
/// # Returns
|
|
///
|
|
/// Returns tuple of (training_data, validation_data) where each element is
|
|
/// `Vec<(FinancialFeatures, Vec<f64>)>` for model training.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Returns error if:
|
|
/// - Database queries fail
|
|
///
|
|
/// - Insufficient data available
|
|
/// - Feature extraction fails
|
|
///
|
|
/// - Data validation fails
|
|
pub async fn load_training_data(
|
|
&mut self,
|
|
) -> Result<(
|
|
Vec<(FinancialFeatures, Vec<f64>)>,
|
|
Vec<(FinancialFeatures, Vec<f64>)>,
|
|
)> {
|
|
info!("Starting training data load from database");
|
|
|
|
// Step 1: Load raw data from database tables
|
|
let order_book_data = self.load_order_book_data().await?;
|
|
let trade_data = self.load_trade_data().await?;
|
|
let market_events = self.load_market_events().await?;
|
|
|
|
info!(
|
|
"Loaded raw data: {} order book snapshots, {} trades, {} events",
|
|
order_book_data.len(),
|
|
trade_data.len(),
|
|
market_events.len()
|
|
);
|
|
|
|
// Step 2: Validate data quality
|
|
self.validate_data_quality(&order_book_data, &trade_data)?;
|
|
|
|
// Step 3: Convert to FinancialFeatures with targets
|
|
let all_features =
|
|
self.convert_to_features_with_targets(order_book_data, trade_data, market_events)?;
|
|
|
|
info!("Converted to {} feature samples", all_features.len());
|
|
|
|
// Step 4: Split into training and validation sets
|
|
let (mut training_data, mut validation_data) = self.split_train_validation(all_features)?;
|
|
|
|
info!(
|
|
"Split data: {} training samples, {} validation samples",
|
|
training_data.len(),
|
|
validation_data.len()
|
|
);
|
|
|
|
// Step 5: Apply normalization to features
|
|
// Fit on training data, apply to both training and validation
|
|
// This prevents data leakage by using only training set statistics
|
|
if !training_data.is_empty() {
|
|
// Fit normalization parameters on training data
|
|
let normalization_params = self.fit_normalization(&training_data);
|
|
|
|
// Apply fitted parameters to training data
|
|
self.transform_with_params(&mut training_data, &normalization_params);
|
|
|
|
// Apply same parameters to validation data (prevents data leakage)
|
|
if !validation_data.is_empty() {
|
|
self.transform_with_params(&mut validation_data, &normalization_params);
|
|
}
|
|
}
|
|
|
|
Ok((training_data, validation_data))
|
|
}
|
|
|
|
/// Load order book snapshots from database
|
|
async fn load_order_book_data(&self) -> Result<Vec<OrderBookSnapshot>> {
|
|
let time_range = &self.config.time_range;
|
|
let symbols = &self.config.symbols;
|
|
let db = self.config.database.as_ref().ok_or_else(|| {
|
|
anyhow::anyhow!("Database configuration required for historical data loading")
|
|
})?;
|
|
let tables = &db.tables;
|
|
|
|
let query = if symbols.is_empty() {
|
|
// Load all symbols
|
|
format!(
|
|
r#"
|
|
SELECT * FROM {}
|
|
WHERE timestamp >= $1 AND timestamp <= $2
|
|
AND data_quality >= 80
|
|
ORDER BY timestamp ASC
|
|
LIMIT 100000
|
|
"#,
|
|
tables.order_books
|
|
)
|
|
} else {
|
|
// Load specific symbols
|
|
format!(
|
|
r#"
|
|
SELECT * FROM {}
|
|
WHERE timestamp >= $1 AND timestamp <= $2
|
|
AND symbol = ANY($3)
|
|
AND data_quality >= 80
|
|
ORDER BY timestamp ASC
|
|
LIMIT 100000
|
|
"#,
|
|
tables.order_books
|
|
)
|
|
};
|
|
|
|
let start = time_range
|
|
.start
|
|
.unwrap_or_else(|| Utc::now() - chrono::Duration::days(30));
|
|
let end = time_range.end.unwrap_or_else(Utc::now);
|
|
|
|
debug!("Querying order books from {} to {}", start, end);
|
|
|
|
let snapshots = if symbols.is_empty() {
|
|
sqlx::query_as::<_, OrderBookSnapshot>(&query)
|
|
.bind(start)
|
|
.bind(end)
|
|
.fetch_all(&self.pool)
|
|
.await
|
|
.context("Failed to query order book snapshots")?
|
|
} else {
|
|
sqlx::query_as::<_, OrderBookSnapshot>(&query)
|
|
.bind(start)
|
|
.bind(end)
|
|
.bind(symbols)
|
|
.fetch_all(&self.pool)
|
|
.await
|
|
.context("Failed to query order book snapshots")?
|
|
};
|
|
|
|
Ok(snapshots)
|
|
}
|
|
|
|
/// Load trade executions from database
|
|
async fn load_trade_data(&self) -> Result<Vec<TradeExecution>> {
|
|
let time_range = &self.config.time_range;
|
|
let symbols = &self.config.symbols;
|
|
let db = self.config.database.as_ref().ok_or_else(|| {
|
|
anyhow::anyhow!("Database configuration required for historical data loading")
|
|
})?;
|
|
let tables = &db.tables;
|
|
|
|
let query = if symbols.is_empty() {
|
|
format!(
|
|
r#"
|
|
SELECT * FROM {}
|
|
WHERE timestamp >= $1 AND timestamp <= $2
|
|
AND data_quality >= 80
|
|
ORDER BY timestamp ASC
|
|
LIMIT 100000
|
|
"#,
|
|
tables.trades
|
|
)
|
|
} else {
|
|
format!(
|
|
r#"
|
|
SELECT * FROM {}
|
|
WHERE timestamp >= $1 AND timestamp <= $2
|
|
AND symbol = ANY($3)
|
|
AND data_quality >= 80
|
|
ORDER BY timestamp ASC
|
|
LIMIT 100000
|
|
"#,
|
|
tables.trades
|
|
)
|
|
};
|
|
|
|
let start = time_range
|
|
.start
|
|
.unwrap_or_else(|| Utc::now() - chrono::Duration::days(30));
|
|
let end = time_range.end.unwrap_or_else(Utc::now);
|
|
|
|
debug!("Querying trades from {} to {}", start, end);
|
|
|
|
let trades = if symbols.is_empty() {
|
|
sqlx::query_as::<_, TradeExecution>(&query)
|
|
.bind(start)
|
|
.bind(end)
|
|
.fetch_all(&self.pool)
|
|
.await
|
|
.context("Failed to query trade executions")?
|
|
} else {
|
|
sqlx::query_as::<_, TradeExecution>(&query)
|
|
.bind(start)
|
|
.bind(end)
|
|
.bind(symbols)
|
|
.fetch_all(&self.pool)
|
|
.await
|
|
.context("Failed to query trade executions")?
|
|
};
|
|
|
|
Ok(trades)
|
|
}
|
|
|
|
/// Load market events from database
|
|
async fn load_market_events(&self) -> Result<Vec<MarketEvent>> {
|
|
let time_range = &self.config.time_range;
|
|
let symbols = &self.config.symbols;
|
|
let db = self.config.database.as_ref().ok_or_else(|| {
|
|
anyhow::anyhow!("Database configuration required for historical data loading")
|
|
})?;
|
|
let tables = &db.tables;
|
|
|
|
let query = if symbols.is_empty() {
|
|
format!(
|
|
r#"
|
|
SELECT * FROM {}
|
|
WHERE timestamp >= $1 AND timestamp <= $2
|
|
ORDER BY timestamp ASC
|
|
LIMIT 10000
|
|
"#,
|
|
tables.market_data
|
|
)
|
|
} else {
|
|
format!(
|
|
r#"
|
|
SELECT * FROM {}
|
|
WHERE timestamp >= $1 AND timestamp <= $2
|
|
AND (symbol = ANY($3) OR symbol IS NULL)
|
|
ORDER BY timestamp ASC
|
|
LIMIT 10000
|
|
"#,
|
|
tables.market_data
|
|
)
|
|
};
|
|
|
|
let start = time_range
|
|
.start
|
|
.unwrap_or_else(|| Utc::now() - chrono::Duration::days(30));
|
|
let end = time_range.end.unwrap_or_else(Utc::now);
|
|
|
|
debug!("Querying market events from {} to {}", start, end);
|
|
|
|
let events = if symbols.is_empty() {
|
|
sqlx::query_as::<_, MarketEvent>(&query)
|
|
.bind(start)
|
|
.bind(end)
|
|
.fetch_all(&self.pool)
|
|
.await
|
|
.context("Failed to query market events")?
|
|
} else {
|
|
sqlx::query_as::<_, MarketEvent>(&query)
|
|
.bind(start)
|
|
.bind(end)
|
|
.bind(symbols)
|
|
.fetch_all(&self.pool)
|
|
.await
|
|
.context("Failed to query market events")?
|
|
};
|
|
|
|
Ok(events)
|
|
}
|
|
|
|
/// Validate data quality and quantity
|
|
fn validate_data_quality(
|
|
&self,
|
|
order_book_data: &[OrderBookSnapshot],
|
|
_trade_data: &[TradeExecution],
|
|
) -> Result<()> {
|
|
let validation = &self.config.validation;
|
|
|
|
// Check minimum samples
|
|
let total_samples = order_book_data.len();
|
|
if total_samples < validation.min_samples {
|
|
anyhow::bail!(
|
|
"Insufficient data: {} samples (minimum {} required)",
|
|
total_samples,
|
|
validation.min_samples
|
|
);
|
|
}
|
|
|
|
// Check data quality distribution
|
|
let high_quality_count = order_book_data
|
|
.iter()
|
|
.filter(|s| s.is_high_quality())
|
|
.count();
|
|
let quality_ratio = high_quality_count as f64 / total_samples as f64;
|
|
|
|
if quality_ratio < (1.0 - validation.max_missing_ratio) {
|
|
warn!(
|
|
"Data quality concern: only {:.1}% high quality samples",
|
|
quality_ratio * 100.0
|
|
);
|
|
}
|
|
|
|
debug!(
|
|
"Data validation passed: {} samples, {:.1}% high quality",
|
|
total_samples,
|
|
quality_ratio * 100.0
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Convert database rows to FinancialFeatures with targets
|
|
fn convert_to_features_with_targets(
|
|
&mut self,
|
|
order_book_data: Vec<OrderBookSnapshot>,
|
|
trade_data: Vec<TradeExecution>,
|
|
_market_events: Vec<MarketEvent>,
|
|
) -> Result<Vec<(FinancialFeatures, Vec<f64>)>> {
|
|
let mut result = Vec::new();
|
|
|
|
// Group trades by timestamp window for aggregation
|
|
let trade_map = self.group_trades_by_time(&trade_data);
|
|
|
|
// Convert each order book snapshot to features
|
|
for (i, snapshot) in order_book_data.iter().enumerate() {
|
|
// Extract features from order book
|
|
let features = self.snapshot_to_features(snapshot, &trade_map)?;
|
|
|
|
// Compute target (next price change) if we have future data
|
|
let target = if i + 1 < order_book_data.len() {
|
|
let next_snapshot = &order_book_data[i + 1];
|
|
vec![self.compute_price_change_target(snapshot, next_snapshot)]
|
|
} else {
|
|
vec![0.0] // Last sample has no target
|
|
};
|
|
|
|
result.push((features, target));
|
|
}
|
|
|
|
Ok(result)
|
|
}
|
|
|
|
/// Group trades by time windows for aggregation
|
|
fn group_trades_by_time<'a>(
|
|
&self,
|
|
trades: &'a [TradeExecution],
|
|
) -> HashMap<DateTime<Utc>, Vec<&'a TradeExecution>> {
|
|
let mut trade_map: HashMap<DateTime<Utc>, Vec<&TradeExecution>> = HashMap::new();
|
|
|
|
// Group trades into 1-second buckets
|
|
for trade in trades {
|
|
// Round timestamp to nearest second by truncating to second precision
|
|
let seconds = trade.timestamp.timestamp();
|
|
let bucket = DateTime::from_timestamp(seconds, 0).unwrap_or(trade.timestamp);
|
|
trade_map.entry(bucket).or_default().push(trade);
|
|
}
|
|
|
|
trade_map
|
|
}
|
|
|
|
/// Convert order book snapshot to FinancialFeatures
|
|
fn snapshot_to_features(
|
|
&mut self,
|
|
snapshot: &OrderBookSnapshot,
|
|
trade_map: &HashMap<DateTime<Utc>, Vec<&TradeExecution>>,
|
|
) -> Result<FinancialFeatures> {
|
|
// Price features
|
|
let raw_mid = snapshot.mid_price_f64();
|
|
let mid_price = Price::from_f64(raw_mid)
|
|
.or_else(|_| Price::new(raw_mid))
|
|
.map_err(|e| anyhow::anyhow!("Failed to construct mid_price: {}", e))?;
|
|
let prices = vec![mid_price];
|
|
|
|
// Volume features
|
|
let total_volume = (snapshot.bid_volume_f64() + snapshot.ask_volume_f64()) as i64;
|
|
let volumes = vec![total_volume];
|
|
|
|
// Technical indicators - Phase 3 implementation with stateful calculations
|
|
let mut technical_indicators = HashMap::new();
|
|
technical_indicators.insert("spread_bps".to_string(), snapshot.spread_bps as f64);
|
|
technical_indicators.insert("imbalance".to_string(), snapshot.imbalance);
|
|
|
|
// Get or create calculator for this symbol
|
|
let calculator = self
|
|
.calculators
|
|
.entry(snapshot.symbol.clone())
|
|
.or_insert_with(|| {
|
|
TechnicalIndicatorCalculator::new(
|
|
snapshot.symbol.clone(),
|
|
IndicatorConfig::default(),
|
|
)
|
|
});
|
|
|
|
// Update calculator with new price data
|
|
let price = snapshot.mid_price_f64();
|
|
let volume = snapshot.bid_volume_f64() + snapshot.ask_volume_f64();
|
|
calculator.update(price, volume, None, None);
|
|
|
|
// Extract calculated indicators
|
|
let calc_indicators = calculator.current_indicators();
|
|
|
|
// Merge calculator indicators into technical_indicators map
|
|
for (key, value) in calc_indicators {
|
|
technical_indicators.insert(key, value);
|
|
}
|
|
|
|
// Microstructure features
|
|
let vwap = self.calculate_vwap(snapshot, trade_map);
|
|
let trade_intensity = self.calculate_trade_intensity(snapshot.timestamp, trade_map);
|
|
|
|
let microstructure = MicrostructureFeatures {
|
|
spread_bps: snapshot.spread_bps,
|
|
imbalance: snapshot.imbalance,
|
|
trade_intensity,
|
|
vwap: Price::from_f64(vwap).unwrap_or(mid_price),
|
|
};
|
|
|
|
// Risk metrics calculated from rolling price history
|
|
let risk_calc = self
|
|
.risk_calculators
|
|
.entry(snapshot.symbol.clone())
|
|
.or_insert_with(|| RiskMetricsCalculator::new(100, 0.0));
|
|
|
|
// Update risk calculator with current price
|
|
risk_calc.update(price);
|
|
|
|
// Calculate all risk metrics
|
|
let risk_metrics_calc = risk_calc.calculate_all_metrics();
|
|
|
|
let risk_metrics = RiskFeatures {
|
|
var_5pct: risk_metrics_calc.var_5pct,
|
|
expected_shortfall: risk_metrics_calc.expected_shortfall,
|
|
max_drawdown: risk_metrics_calc.max_drawdown,
|
|
sharpe_ratio: risk_metrics_calc.sharpe_ratio,
|
|
};
|
|
|
|
Ok(FinancialFeatures {
|
|
prices,
|
|
volumes,
|
|
technical_indicators,
|
|
microstructure,
|
|
risk_metrics,
|
|
timestamp: snapshot.timestamp,
|
|
})
|
|
}
|
|
|
|
/// Calculate volume-weighted average price (VWAP)
|
|
fn calculate_vwap(
|
|
&self,
|
|
snapshot: &OrderBookSnapshot,
|
|
trade_map: &HashMap<DateTime<Utc>, Vec<&TradeExecution>>,
|
|
) -> f64 {
|
|
let seconds = snapshot.timestamp.timestamp();
|
|
let bucket = DateTime::from_timestamp(seconds, 0).unwrap_or(snapshot.timestamp);
|
|
|
|
if let Some(trades) = trade_map.get(&bucket) {
|
|
let mut total_value = 0.0;
|
|
let mut total_volume = 0.0;
|
|
|
|
for trade in trades {
|
|
if trade.symbol == snapshot.symbol {
|
|
total_value += trade.price_f64() * trade.quantity_f64();
|
|
total_volume += trade.quantity_f64();
|
|
}
|
|
}
|
|
|
|
if total_volume > 0.0 {
|
|
total_value / total_volume
|
|
} else {
|
|
snapshot.mid_price_f64()
|
|
}
|
|
} else {
|
|
snapshot.mid_price_f64()
|
|
}
|
|
}
|
|
|
|
/// Calculate trade intensity (trades per second)
|
|
fn calculate_trade_intensity(
|
|
&self,
|
|
timestamp: DateTime<Utc>,
|
|
trade_map: &HashMap<DateTime<Utc>, Vec<&TradeExecution>>,
|
|
) -> f64 {
|
|
let seconds = timestamp.timestamp();
|
|
let bucket = DateTime::from_timestamp(seconds, 0).unwrap_or(timestamp);
|
|
|
|
trade_map
|
|
.get(&bucket)
|
|
.map(|trades| trades.len() as f64)
|
|
.unwrap_or(0.0)
|
|
}
|
|
|
|
/// Compute price change target for prediction
|
|
fn compute_price_change_target(
|
|
&self,
|
|
current: &OrderBookSnapshot,
|
|
next: &OrderBookSnapshot,
|
|
) -> f64 {
|
|
let current_price = current.mid_price_f64();
|
|
let next_price = next.mid_price_f64();
|
|
|
|
if current_price > 0.0 {
|
|
(next_price - current_price) / current_price
|
|
} else {
|
|
0.0
|
|
}
|
|
}
|
|
|
|
/// Split data into training and validation sets
|
|
fn split_train_validation(
|
|
&self,
|
|
mut all_data: Vec<(FinancialFeatures, Vec<f64>)>,
|
|
) -> Result<(
|
|
Vec<(FinancialFeatures, Vec<f64>)>,
|
|
Vec<(FinancialFeatures, Vec<f64>)>,
|
|
)> {
|
|
let train_split = self.config.time_range.train_split;
|
|
let split_index = (all_data.len() as f64 * train_split) as usize;
|
|
|
|
if split_index >= all_data.len() {
|
|
anyhow::bail!(
|
|
"Invalid train split: {:.2} results in {} training samples from {} total",
|
|
train_split,
|
|
split_index,
|
|
all_data.len()
|
|
);
|
|
}
|
|
|
|
let validation_data = all_data.split_off(split_index);
|
|
let training_data = all_data;
|
|
|
|
Ok((training_data, validation_data))
|
|
}
|
|
|
|
/// Fit normalization parameters on training data
|
|
///
|
|
/// Computes statistics (mean, std, min, max, etc.) from training data only.
|
|
///
|
|
/// These parameters are then applied to both training and validation data
|
|
/// to prevent data leakage.
|
|
///
|
|
/// # Arguments
|
|
/// * `features_list` - Training data to fit parameters on
|
|
///
|
|
/// # Returns
|
|
/// * `FeatureNormalizationParams` - Fitted parameters for all features
|
|
pub fn fit_normalization(
|
|
&self,
|
|
features_list: &[(FinancialFeatures, Vec<f64>)],
|
|
) -> FeatureNormalizationParams {
|
|
if features_list.is_empty() {
|
|
return FeatureNormalizationParams {
|
|
indicator_params: HashMap::new(),
|
|
spread_params: NormalizationParams::default(),
|
|
imbalance_params: NormalizationParams::default(),
|
|
intensity_params: NormalizationParams::default(),
|
|
var_params: NormalizationParams::default(),
|
|
es_params: NormalizationParams::default(),
|
|
dd_params: NormalizationParams::default(),
|
|
sharpe_params: NormalizationParams::default(),
|
|
};
|
|
}
|
|
|
|
info!(
|
|
"Fitting normalization parameters on {} training samples",
|
|
features_list.len()
|
|
);
|
|
|
|
// Collect all technical indicator keys
|
|
let mut all_indicator_keys: Vec<String> = features_list
|
|
.first()
|
|
.map(|(f, _)| f.technical_indicators.keys().cloned().collect())
|
|
.unwrap_or_default();
|
|
all_indicator_keys.sort();
|
|
|
|
// Fit normalization parameters for each technical indicator
|
|
let mut indicator_params: HashMap<String, NormalizationParams> = HashMap::new();
|
|
|
|
for key in &all_indicator_keys {
|
|
let values: Vec<f64> = features_list
|
|
.iter()
|
|
.filter_map(|(f, _)| f.technical_indicators.get(key).copied())
|
|
.collect();
|
|
|
|
let params = NormalizationParams::fit(&values);
|
|
indicator_params.insert(key.clone(), params);
|
|
}
|
|
|
|
// Fit parameters for microstructure features
|
|
let spread_values: Vec<f64> = features_list
|
|
.iter()
|
|
.map(|(f, _)| f.microstructure.spread_bps as f64)
|
|
.collect();
|
|
let spread_params = NormalizationParams::fit(&spread_values);
|
|
|
|
let imbalance_values: Vec<f64> = features_list
|
|
.iter()
|
|
.map(|(f, _)| f.microstructure.imbalance)
|
|
.collect();
|
|
let imbalance_params = NormalizationParams::fit(&imbalance_values);
|
|
|
|
let intensity_values: Vec<f64> = features_list
|
|
.iter()
|
|
.map(|(f, _)| f.microstructure.trade_intensity)
|
|
.collect();
|
|
let intensity_params = NormalizationParams::fit(&intensity_values);
|
|
|
|
// Fit parameters for risk metrics
|
|
let var_values: Vec<f64> = features_list
|
|
.iter()
|
|
.map(|(f, _)| f.risk_metrics.var_5pct)
|
|
.collect();
|
|
let var_params = NormalizationParams::fit(&var_values);
|
|
|
|
let es_values: Vec<f64> = features_list
|
|
.iter()
|
|
.map(|(f, _)| f.risk_metrics.expected_shortfall)
|
|
.collect();
|
|
let es_params = NormalizationParams::fit(&es_values);
|
|
|
|
let dd_values: Vec<f64> = features_list
|
|
.iter()
|
|
.map(|(f, _)| f.risk_metrics.max_drawdown)
|
|
.collect();
|
|
let dd_params = NormalizationParams::fit(&dd_values);
|
|
|
|
let sharpe_values: Vec<f64> = features_list
|
|
.iter()
|
|
.map(|(f, _)| f.risk_metrics.sharpe_ratio)
|
|
.collect();
|
|
let sharpe_params = NormalizationParams::fit(&sharpe_values);
|
|
|
|
info!(
|
|
"Fitted normalization parameters for {} technical indicators",
|
|
all_indicator_keys.len()
|
|
);
|
|
|
|
FeatureNormalizationParams {
|
|
indicator_params,
|
|
spread_params,
|
|
imbalance_params,
|
|
intensity_params,
|
|
var_params,
|
|
es_params,
|
|
dd_params,
|
|
sharpe_params,
|
|
}
|
|
}
|
|
|
|
/// Apply pre-fitted normalization parameters to features
|
|
///
|
|
/// Uses normalization parameters fitted on training data to transform
|
|
/// features. This prevents data leakage when normalizing validation data.
|
|
///
|
|
/// # Arguments
|
|
/// * `features_list` - Mutable reference to features to normalize
|
|
///
|
|
/// * `params` - Pre-fitted normalization parameters (from fit_normalization)
|
|
pub fn transform_with_params(
|
|
&self,
|
|
features_list: &mut [(FinancialFeatures, Vec<f64>)],
|
|
params: &FeatureNormalizationParams,
|
|
) {
|
|
let method = NormalizationMethod::from_str(&self.config.features.normalization);
|
|
|
|
if matches!(method, NormalizationMethod::None) {
|
|
debug!("Normalization disabled, skipping");
|
|
return;
|
|
}
|
|
|
|
info!(
|
|
"Applying {:?} normalization to {} samples",
|
|
method,
|
|
features_list.len()
|
|
);
|
|
|
|
if features_list.is_empty() {
|
|
return;
|
|
}
|
|
|
|
// Apply normalization to all features using pre-fitted parameters
|
|
for (features, _) in features_list.iter_mut() {
|
|
// Normalize technical indicators
|
|
for (key, value) in features.technical_indicators.iter_mut() {
|
|
if let Some(indicator_params) = params.indicator_params.get(key) {
|
|
*value = indicator_params.normalize(*value, &method);
|
|
}
|
|
}
|
|
|
|
// Normalize microstructure features
|
|
features.microstructure.imbalance = params
|
|
.imbalance_params
|
|
.normalize(features.microstructure.imbalance, &method);
|
|
features.microstructure.trade_intensity = params
|
|
.intensity_params
|
|
.normalize(features.microstructure.trade_intensity, &method);
|
|
|
|
// Note: spread_bps is u16, so we normalize separately if needed
|
|
let normalized_spread = params
|
|
.spread_params
|
|
.normalize(features.microstructure.spread_bps as f64, &method);
|
|
// Store in technical_indicators for reference
|
|
features
|
|
.technical_indicators
|
|
.insert("spread_bps_normalized".to_string(), normalized_spread);
|
|
|
|
// Normalize risk metrics
|
|
features.risk_metrics.var_5pct = params
|
|
.var_params
|
|
.normalize(features.risk_metrics.var_5pct, &method);
|
|
features.risk_metrics.expected_shortfall = params
|
|
.es_params
|
|
.normalize(features.risk_metrics.expected_shortfall, &method);
|
|
features.risk_metrics.max_drawdown = params
|
|
.dd_params
|
|
.normalize(features.risk_metrics.max_drawdown, &method);
|
|
features.risk_metrics.sharpe_ratio = params
|
|
.sharpe_params
|
|
.normalize(features.risk_metrics.sharpe_ratio, &method);
|
|
}
|
|
|
|
info!("Normalization complete");
|
|
}
|
|
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[allow(clippy::unwrap_used, clippy::expect_used)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[tokio::test]
|
|
async fn test_price_change_calculation() {
|
|
let loader = create_test_loader();
|
|
|
|
let current = create_test_snapshot(100.0);
|
|
let next = create_test_snapshot(101.0);
|
|
|
|
let change = loader.compute_price_change_target(¤t, &next);
|
|
assert!((change - 0.01).abs() < 1e-6); // 1% increase
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_vwap_calculation() {
|
|
let loader = create_test_loader();
|
|
let snapshot = create_test_snapshot(100.0);
|
|
let trade_map = HashMap::new();
|
|
|
|
let vwap = loader.calculate_vwap(&snapshot, &trade_map);
|
|
assert_eq!(vwap, 100.0); // No trades, should return mid price
|
|
}
|
|
|
|
fn create_test_loader() -> HistoricalDataLoader {
|
|
use crate::data_config::*;
|
|
|
|
let config = TrainingDataSourceConfig {
|
|
source_type: DataSourceType::Historical,
|
|
database: None,
|
|
s3: None,
|
|
time_range: TimeRangeConfig::default(),
|
|
symbols: vec![],
|
|
features: FeatureExtractionConfig::default(),
|
|
validation: DataValidationConfig::default(),
|
|
cache: CacheConfig::default(),
|
|
};
|
|
|
|
// Create a test pool that won't actually be used
|
|
// We use a minimal PgPoolOptions that will create an unconnected pool
|
|
let pool = sqlx::postgres::PgPoolOptions::new()
|
|
.max_connections(1)
|
|
.connect_lazy("postgres://test:test@localhost:5432/test_db")
|
|
.expect("Failed to create test pool");
|
|
|
|
HistoricalDataLoader {
|
|
pool,
|
|
config,
|
|
calculators: HashMap::new(),
|
|
risk_calculators: HashMap::new(),
|
|
}
|
|
}
|
|
|
|
fn create_test_snapshot(mid_price: f64) -> OrderBookSnapshot {
|
|
OrderBookSnapshot {
|
|
id: 1,
|
|
timestamp: Utc::now(),
|
|
symbol: "TEST".to_string(),
|
|
best_bid: rust_decimal::Decimal::from_f64_retain(mid_price - 0.01).unwrap(),
|
|
best_ask: rust_decimal::Decimal::from_f64_retain(mid_price + 0.01).unwrap(),
|
|
bid_volume: rust_decimal::Decimal::from(1000),
|
|
ask_volume: rust_decimal::Decimal::from(1000),
|
|
spread_bps: 2,
|
|
mid_price: rust_decimal::Decimal::from_f64_retain(mid_price).unwrap(),
|
|
imbalance: 0.0,
|
|
bid_levels: None,
|
|
ask_levels: None,
|
|
exchange: Some("TEST".to_string()),
|
|
data_quality: Some(100),
|
|
created_at: Utc::now(),
|
|
}
|
|
}
|
|
}
|