SIENTIAPDE-1430: Refactor DataPreprocessor date filtering, make rce_train radius optional, and introduce constants for model names.

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
+    # Non-parametric radius: Silverman Radius (compute if not provided)
-    radius = silverman_radius(distances.flatten())
+    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)

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
+        start = self._parse_datetime(self.start_date)
-        if self.start_date:
+        end = self._parse_datetime(self.end_date)
-            try:
+        input_data = self._filter_by_date_range(input_data, start, end)
-                start = pd.to_datetime(self.start_date)
-                input_data = input_data[input_data.index >= start]
-            except (ValueError, TypeError):
-                pass  # Invalid date format, skip filtering
 
-        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:
+                input_data = self._remove_interval(input_data, interval)
-                    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
 
         return input_data
 

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:
+        if self.model_name == MODEL_LINEAR_REGRESSION and self.degree != 1:
-            raise ValueError(f'degree must be 1 for Linear Regression, got {self.degree}')
+            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."""