From b4cfd88634dd7705c1f0f48a5b2f778ec1abb5a3 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Fri, 27 Feb 2026 18:56:04 +0100 Subject: [PATCH] refactor(ml): drop dead TFT Parquet loader (-130 lines) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- crates/ml/src/trainers/tft_parquet.rs | 166 +------------------------- 1 file changed, 3 insertions(+), 163 deletions(-) diff --git a/crates/ml/src/trainers/tft_parquet.rs b/crates/ml/src/trainers/tft_parquet.rs index cce66daeb..d32aa2929 100644 --- a/crates/ml/src/trainers/tft_parquet.rs +++ b/crates/ml/src/trainers/tft_parquet.rs @@ -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 { - 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 { 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, Array2, Array2, Array1)>> { - 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::>() - .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::() - .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::() - .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::() - .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::() - .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::() - .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,