From 7cbb7022e5f3ba9e045096cfe03041072699b032 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Fri, 19 Dec 2025 12:28:07 -0300 Subject: [PATCH] SIENTIAPDE-1430: Refactor DataPreprocessor date filtering, make rce_train radius optional, and introduce constants for model names. --- model_manager/sientia/metrics.py | 10 ++-- model_manager/sientia/models.py | 60 +++++++++++-------- .../utils/models/train_model_params.py | 18 +++--- 3 files changed, 50 insertions(+), 38 deletions(-) diff --git a/model_manager/sientia/metrics.py b/model_manager/sientia/metrics.py index 9cd82cf..a3a3265 100644 --- a/model_manager/sientia/metrics.py +++ b/model_manager/sientia/metrics.py @@ -45,13 +45,13 @@ def silverman_radius(data: np.ndarray) -> float: return radius -def rce_train(training_set: pd.DataFrame, radius: float) -> pd.DataFrame: +def rce_train(training_set: pd.DataFrame, radius: float | None = None) -> pd.DataFrame: """ Get the Reduced Coulomb Energy (RCE) prototypes. Args: training_set (pd.DataFrame): The training set - radius (float): The radius of the RCE prototypes + radius (float | None): The radius of the RCE prototypes. If None, computed using Silverman's rule. Returns: pd.DataFrame: The RCE prototypes @@ -62,8 +62,8 @@ def rce_train(training_set: pd.DataFrame, radius: float) -> pd.DataFrame: diff_vectors = train_vectors[:, np.newaxis] - train_vectors[np.newaxis, :] distances = np.linalg.norm(diff_vectors, axis=-1) - # Non-parametric radius: Silverman Radius - radius = silverman_radius(distances.flatten()) + # Non-parametric radius: Silverman Radius (compute if not provided) + effective_radius = radius if radius is not None else silverman_radius(distances.flatten()) # Initialize prototypes with the first vector prototypes = [train_vectors[0]] @@ -73,7 +73,7 @@ def rce_train(training_set: pd.DataFrame, radius: float) -> pd.DataFrame: distances_to_prototypes = np.linalg.norm(prototypes - vector, axis=1) # If no prototype is close, add the current vector as a new prototype - if np.all(distances_to_prototypes > radius): + if np.all(distances_to_prototypes > effective_radius): prototypes.append(vector) return pd.DataFrame(prototypes) diff --git a/model_manager/sientia/models.py b/model_manager/sientia/models.py index 6517fae..b3abc3f 100644 --- a/model_manager/sientia/models.py +++ b/model_manager/sientia/models.py @@ -442,6 +442,36 @@ class DataPreprocessor(BaseEstimator, TransformerMixin): input_data = treat_nan(input_data, treatment) return input_data + def _parse_datetime(self, date_str: str | None) -> pd.Timestamp | None: + """Parse a date string to Timestamp, returning None on failure.""" + if not date_str: + return None + try: + return pd.to_datetime(date_str) + except (ValueError, TypeError): + return None + + def _filter_by_date_range( + self, input_data: pd.DataFrame, start: pd.Timestamp | None, end: pd.Timestamp | None + ) -> pd.DataFrame: + """Filter DataFrame by start and end dates.""" + if start is not None: + input_data = input_data[input_data.index >= start] + if end is not None: + input_data = input_data[input_data.index <= end] + return input_data + + def _remove_interval(self, input_data: pd.DataFrame, interval: tuple | list) -> pd.DataFrame: + """Remove a single interval from the DataFrame.""" + if len(interval) < 2: + return input_data + interval_start = self._parse_datetime(interval[0]) + interval_end = self._parse_datetime(interval[1]) + if interval_start is None or interval_end is None: + return input_data + mask = ~((input_data.index >= interval_start) & (input_data.index <= interval_end)) + return input_data[mask] + def range_selection(self, input_data: pd.DataFrame) -> pd.DataFrame: """ Filter data by date range and remove specified intervals. @@ -452,35 +482,13 @@ class DataPreprocessor(BaseEstimator, TransformerMixin): Returns: pandas.DataFrame: The filtered data """ - # Filter by start_date and end_date - if self.start_date: - try: - start = pd.to_datetime(self.start_date) - input_data = input_data[input_data.index >= start] - except (ValueError, TypeError): - pass # Invalid date format, skip filtering + start = self._parse_datetime(self.start_date) + end = self._parse_datetime(self.end_date) + input_data = self._filter_by_date_range(input_data, start, end) - if self.end_date: - try: - end = pd.to_datetime(self.end_date) - input_data = input_data[input_data.index <= end] - except (ValueError, TypeError): - pass # Invalid date format, skip filtering - - # Remove specified intervals if self.removed_intervals: for interval in self.removed_intervals: - if len(interval) >= 2: - try: - interval_start = pd.to_datetime(interval[0]) - interval_end = pd.to_datetime(interval[1]) - mask = ~( - (input_data.index >= interval_start) - & (input_data.index <= interval_end) - ) - input_data = input_data[mask] - except (ValueError, TypeError): - pass # Invalid date format, skip this interval + input_data = self._remove_interval(input_data, interval) return input_data diff --git a/model_manager/utils/models/train_model_params.py b/model_manager/utils/models/train_model_params.py index 09f45bc..e88a3f0 100644 --- a/model_manager/utils/models/train_model_params.py +++ b/model_manager/utils/models/train_model_params.py @@ -1,6 +1,10 @@ from dataclasses import dataclass from typing import Any +# Model name constants +MODEL_LINEAR_REGRESSION = 'Linear Regression' +MODEL_POLYNOMIAL_REGRESSION = 'Polynomial Regression' + @dataclass class TrainModelParams: @@ -239,23 +243,23 @@ class TrainModelParams: if self.scaler_name not in valid_scalers: raise ValueError(f'scaler_name must be one of {valid_scalers}, got {self.scaler_name}') - valid_models = ['Linear Regression', 'Polynomial Regression'] + valid_models = [MODEL_LINEAR_REGRESSION, MODEL_POLYNOMIAL_REGRESSION] if self.model_name not in valid_models: raise ValueError(f'model_name must be one of {valid_models}, got {self.model_name}') - if self.model_name == 'Polynomial Regression' and self.degree < 2: + if self.model_name == MODEL_POLYNOMIAL_REGRESSION and self.degree < 2: raise ValueError( - f'degree must be at least 2 for Polynomial Regression, got {self.degree}' + f'degree must be at least 2 for {MODEL_POLYNOMIAL_REGRESSION}, got {self.degree}' ) - if self.model_name == 'Polynomial Regression' and self.scaler_name == 'None': + if self.model_name == MODEL_POLYNOMIAL_REGRESSION and self.scaler_name == 'None': raise ValueError( - 'scaler_name must be set (e.g., "Standard Scaler") for Polynomial Regression ' + f'scaler_name must be set (e.g., "Standard Scaler") for {MODEL_POLYNOMIAL_REGRESSION} ' 'to avoid numerical overflow with large feature values' ) - if self.model_name == 'Linear Regression' and self.degree != 1: - raise ValueError(f'degree must be 1 for Linear Regression, got {self.degree}') + if self.model_name == MODEL_LINEAR_REGRESSION and self.degree != 1: + raise ValueError(f'degree must be 1 for {MODEL_LINEAR_REGRESSION}, got {self.degree}') def _validate_intervals_and_dates(self) -> None: """Validate removed_intervals format and date parameters."""