gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(ml): drop dead TFT Parquet loader (-130 lines)

Browse Source 
train_from_parquet and load_training_data_from_parquet had zero
callers — TFT hyperopt uses train_from_bars via DBN data.
Also removes unused NormalizationParams struct from this module.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-02-27 18:56:04 +01:00
parent 3564efe961
commit b4cfd88634
1 changed files with 3 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

166
crates/ml/src/trainers/tft_parquet.rs
View File

@@ -1,7 +1,7 @@
//! TFT Parquet Training Extension
//! TFT DBN Training Extension
//!
//! This module adds Parquet lazy-loading support to the TFT trainer
//! to fix OOM crashes when training on large datasets.
//! Adds OHLCV bar training support to TFTTrainer with sliding window
//! sample creation (lookback=60, horizon=10) and OOM-resilient batch retry.
use ndarray::{Array1, Array2};
use tracing::info;
@@ -11,39 +11,7 @@ use crate::tft::training::TFTDataLoader;
use crate::trainers::tft::{TFTTrainer, TrainingMetrics};
use crate::{MLError, MLResult};
/// Normalization parameters for target denormalization
#[derive(Debug, Clone)]
pub struct NormalizationParams {
pub mean: f64,
pub std: f64,
}
impl TFTTrainer {
/// Train TFT on market data from Parquet file (lazy-loading to avoid OOM)
///
/// # Arguments
///
/// * `parquet_path` - Path to Parquet file containing OHLCV bars
///
/// # Returns
///
/// Training metrics (loss, RMSE, quantile loss, attention entropy)
///
/// # Features
///
/// - Lazy batch loading (10,000 rows at a time)
/// - 51-feature extraction from OHLCV bars (WAVE 10: 5 static + 10 known + 36 unknown)
/// - Sliding window creation (lookback=60, horizon=10)
/// - Memory-efficient processing for large datasets
pub async fn train_from_parquet(&mut self, parquet_path: &str) -> MLResult<TrainingMetrics> {
info!("Starting TFT training from Parquet file: {}", parquet_path);
// Load market data from Parquet file
let training_data = self.load_training_data_from_parquet(parquet_path).await?;
self.train_from_samples(training_data).await
}
/// Train TFT on pre-loaded OHLCV bars (from DBN or any other source).
pub async fn train_from_bars(&mut self, bars: &[OHLCVBar]) -> MLResult<TrainingMetrics> {
info!("Starting TFT training from {} OHLCV bars", bars.len());
@@ -186,134 +154,6 @@ impl TFTTrainer {
}
}
/// Load training data from Parquet file with lazy batch loading
async fn load_training_data_from_parquet(
&mut self,
parquet_path: &str,
) -> MLResult<Vec<(Array1<f64>, Array2<f64>, Array2<f64>, Array1<f64>)>> {
use arrow::array::{Array, Float64Array, PrimitiveArray, UInt64Array};
use arrow::datatypes::TimestampNanosecondType;
use arrow::record_batch::RecordBatch;
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use std::fs::File;
info!("Loading Parquet file: {}", parquet_path);
let file = File::open(parquet_path)
.map_err(|e| MLError::ModelError(format!("Failed to open Parquet file: {}", e)))?;
let builder = ParquetRecordBatchReaderBuilder::try_new(file)
.map_err(|e| MLError::ModelError(format!("Failed to create Parquet reader: {}", e)))?;
let reader = builder
.build()
.map_err(|e| MLError::ModelError(format!("Failed to build Parquet reader: {}", e)))?;
let mut all_ohlcv_bars = Vec::new();
for batch_result in reader {
let batch: RecordBatch = batch_result
.map_err(|e| MLError::ModelError(format!("Failed to read record batch: {}", e)))?;
let timestamp_col = batch
.column_by_name("timestamp_ns")
.or_else(|| batch.column_by_name("ts_event"))
.ok_or_else(|| {
MLError::InvalidInput(
"Missing timestamp column. Expected 'timestamp_ns' or 'ts_event'"
.to_string(),
)
})?;
let timestamps = timestamp_col
.as_any()
.downcast_ref::<PrimitiveArray<TimestampNanosecondType>>()
.ok_or_else(|| MLError::InvalidInput(
format!(
"Failed to downcast timestamp column. Expected Timestamp(Nanosecond), got: {:?}",
timestamp_col.data_type()
)
))?;
let opens = batch
.column_by_name("open")
.ok_or_else(|| {
MLError::InvalidInput("Missing 'open' column in Parquet schema".to_owned())
})?
.as_any()
.downcast_ref::<Float64Array>()
.ok_or_else(|| {
MLError::InvalidInput("Invalid 'open' column type. Expected Float64".to_owned())
})?;
let highs = batch
.column_by_name("high")
.ok_or_else(|| {
MLError::InvalidInput("Missing 'high' column in Parquet schema".to_owned())
})?
.as_any()
.downcast_ref::<Float64Array>()
.ok_or_else(|| {
MLError::InvalidInput("Invalid 'high' column type. Expected Float64".to_owned())
})?;
let lows = batch
.column_by_name("low")
.ok_or_else(|| {
MLError::InvalidInput("Missing 'low' column in Parquet schema".to_owned())
})?
.as_any()
.downcast_ref::<Float64Array>()
.ok_or_else(|| {
MLError::InvalidInput("Invalid 'low' column type. Expected Float64".to_owned())
})?;
let closes = batch
.column_by_name("close")
.ok_or_else(|| {
MLError::InvalidInput("Missing 'close' column in Parquet schema".to_owned())
})?
.as_any()
.downcast_ref::<Float64Array>()
.ok_or_else(|| {
MLError::InvalidInput("Invalid 'close' column type. Expected Float64".to_owned())
})?;
let volumes = batch
.column_by_name("volume")
.ok_or_else(|| {
MLError::InvalidInput("Missing 'volume' column in Parquet schema".to_owned())
})?
.as_any()
.downcast_ref::<UInt64Array>()
.ok_or_else(|| {
MLError::InvalidInput("Invalid 'volume' column type. Expected UInt64".to_owned())
})?;
for i in 0..batch.num_rows() {
let timestamp_ns = timestamps.value(i);
let timestamp = chrono::DateTime::from_timestamp_nanos(timestamp_ns);
let bar = OHLCVBar {
timestamp,
open: opens.value(i),
high: highs.value(i),
low: lows.value(i),
close: closes.value(i),
volume: volumes.value(i) as f64,
};
all_ohlcv_bars.push(bar);
}
}
info!(
"Successfully loaded {} OHLCV bars from Parquet file",
all_ohlcv_bars.len()
);
self.prepare_training_data(&all_ohlcv_bars)
}
/// Convert OHLCV bars into TFT training samples (data-source-agnostic).
fn prepare_training_data(
&mut self,