From 4d761c15ac3071b0c2df113dafeec6db2991aab0 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 13 Apr 2026 10:40:21 -0300 Subject: [PATCH] chore: remove sientia models module and relocate date format validation - Deleted the sientia/models.py file, which contained the LinearRegressionModel and related functionality. - Moved the frontend date format validation logic to train_model_params.py, ensuring a single source of truth for date formats. --- model_manager/sientia/models.py | 804 ------------------ .../utils/models/train_model_params.py | 20 +- 2 files changed, 19 insertions(+), 805 deletions(-) delete mode 100644 model_manager/sientia/models.py diff --git a/model_manager/sientia/models.py b/model_manager/sientia/models.py deleted file mode 100644 index 80b38c8..0000000 --- a/model_manager/sientia/models.py +++ /dev/null @@ -1,804 +0,0 @@ -from typing import Any - -import numpy as np -import pandas as pd -from sientia_do.operations.df_preprocessor import create_features, limit_dataset, treat_nan -from sientia_do.timeseries.analyzer import TimeSeriesDiscontinuityAnalyzer -from sklearn.base import BaseEstimator, TransformerMixin -from sklearn.linear_model import LinearRegression -from sklearn.preprocessing import PolynomialFeatures, StandardScaler - -DISCONTINUITY_TREATMENT = 'Discontinuity Treatment' -LAG_SELECTION = 'Lag Selection' -RANGE_SELECTION = 'Range Selection & Data Removal' -STATIC_WINDOW_REMOVAL = 'Static Window Removal' -DEFINE_VARIABLES_LIMITS = 'Define Variables Limits' -NORMALIZATION = 'Normalization' -FEATURE_CREATION = 'Feature Creation' -LAG_CREATION = 'Lag Creation' - -# Allowed frontend date formats and their strftime equivalents (single source of truth) -FRONTEND_DATE_FORMAT_TO_STRFTIME = { - 'dd/MM/yyyy HH:mm:ss': '%d/%m/%Y %H:%M:%S', - 'MM/dd/yyyy HH:mm:ss': '%m/%d/%Y %H:%M:%S', - 'yyyy/MM/dd HH:mm:ss': '%Y/%m/%d %H:%M:%S', - 'dd-MM-yyyy HH:mm:ss': '%d-%m-%Y %H:%M:%S', - 'MM-dd-yyyy HH:mm:ss': '%m-%d-%Y %H:%M:%S', - 'yyyy-MM-dd HH:mm:ss': '%Y-%m-%d %H:%M:%S', -} -ALLOWED_FRONTEND_DATE_FORMATS = frozenset(FRONTEND_DATE_FORMAT_TO_STRFTIME.keys()) - - -def validate_frontend_date_format(fmt: str | None) -> None: - """Raise ValueError if fmt is set and not one of the allowed frontend date formats.""" - if not fmt or not fmt.strip(): - return - if fmt not in ALLOWED_FRONTEND_DATE_FORMATS: - allowed = ', '.join(sorted(ALLOWED_FRONTEND_DATE_FORMATS)) - raise ValueError(f'Invalid date_format "{fmt}". Allowed formats: {allowed}') - - -def _frontend_date_format_to_strftime(fmt: str | None) -> str | None: - """Convert front-end date format to Python strftime. Validates format; returns None for empty.""" - if not fmt: - return None - validate_frontend_date_format(fmt) - return FRONTEND_DATE_FORMAT_TO_STRFTIME[fmt] - - -class LinearRegressionModel(BaseEstimator, TransformerMixin): - """ - Linear Regression Model for Time Series Analysis. - - Supports both simple linear regression and polynomial regression. - - Thread-safety: This class is NOT thread-safe during fit() operations. - Do not call fit() on the same instance from multiple threads simultaneously. - After fitting, predict() is thread-safe for read-only operations. - - For multi-threaded environments: - - Fit the model in a single thread - - Share the fitted instance across threads for prediction only - - Or create separate instances per thread - """ - - def __init__( - self, - target_variable: str = '', - variable_columns: list[str] | None = None, - model_params: dict[str, Any] | None = None, - clipping: dict[str, float] | None = None, - weights: dict[str, float] | None = None, - degree: int = 1, - interaction_only: bool = False, - ): - """ - Linear Regression Model for Time Series Analysis - - Args: - target_variable (str): The target variable name - variable_columns (list): The input columns names in a list - model_params (dict): The parameters used for training the model \\ - clipping (dict): The lower and upper limits for the target variable to be clipped \\ - *Format: {'min': min_value, 'max': max_value}* - weights (dict): The weights for the Linear Regression model \\ - *Format: {'variable_name': weight}* - degree (int): The degree of the polynomial features (1 = linear, >1 = polynomial) - interaction_only (bool): If True, only interaction features are produced - - Returns: - LinearRegressionModel: The prediction model object - """ - self.target_variable: str = target_variable - self.variable_columns: list[str] | None = variable_columns - self.model_params: dict[str, Any] | None = model_params - self.clipping: dict[str, float] | None = clipping - self.regr = LinearRegression() - self.q1_target: float | None = None - self.q3_target: float | None = None - self.weights: dict[str, float] | None = weights - self.degree: int = degree - self.interaction_only: bool = interaction_only - self.poly: PolynomialFeatures | None = None - self.poly_feature_names: list[str] | None = None - - def create_poly_features(self, input_data: pd.DataFrame, fit: bool = False) -> pd.DataFrame: - """ - Create polynomial features from input data. - - Args: - input_data (pd.DataFrame): Input data with feature columns - fit (bool): If True, fit the PolynomialFeatures transformer - - Returns: - pd.DataFrame: DataFrame with polynomial features - """ - if self.degree <= 1: - return input_data - - if fit: - self.poly = PolynomialFeatures( - degree=self.degree, - interaction_only=self.interaction_only, - include_bias=False, - ) - poly_features = self.poly.fit_transform(input_data) - self.poly_feature_names = list(self.poly.get_feature_names_out(input_data.columns)) - else: - if self.poly is None: - raise ValueError('PolynomialFeatures not fitted. Call fit() first.') - poly_features = self.poly.transform(input_data) - - return pd.DataFrame( - poly_features, - columns=self.poly_feature_names, - index=input_data.index, - ) - - def fit(self, input_data: pd.DataFrame) -> 'LinearRegressionModel': - """ - Function to fit the model - - Args: - input_data (pandas.DataFrame): The data used to fit the Linear Regression model - - Returns: - LinearRegressionModel: The prediction model object - """ - if not self.target_variable: - raise ValueError('target_variable must be set before fitting') - - # Infer variable_columns if not provided - if self.variable_columns is None: - self.variable_columns = [ - col for col in input_data.columns if col != self.target_variable - ] - - # Validate columns exist - missing_cols = [col for col in self.variable_columns if col not in input_data.columns] - if missing_cols: - raise ValueError(f'Columns not found in input data: {missing_cols}') - - if self.target_variable not in input_data.columns: - raise ValueError(f'Target variable {self.target_variable} not found in input data') - - X_train = input_data[self.variable_columns].copy() - y_train = input_data[self.target_variable].copy() - - # Handle infinite values - X_train = X_train.replace([np.inf, -np.inf], np.nan) - y_train = y_train.replace([np.inf, -np.inf], np.nan) - - # Remove rows with NaN - valid_mask = ~(X_train.isna().any(axis=1) | y_train.isna()) - X_train = X_train[valid_mask] - y_train = y_train[valid_mask] - - # Remove columns with all NaN values - cols_to_drop = X_train.columns[X_train.isna().all()].tolist() - if cols_to_drop: # pragma: no cover - X_train = X_train.drop(columns=cols_to_drop) - self.variable_columns = [c for c in self.variable_columns if c not in cols_to_drop] - - self.q1_target = y_train.quantile(0.25) - self.q3_target = y_train.quantile(0.75) - - # Apply polynomial features if degree > 1 - if self.degree > 1: - X_train = self.create_poly_features(X_train, fit=True) - - # Fit the model - self.regr.fit(X_train, y_train) - - # Get the weights - round_coef = np.round(self.regr.coef_, 3) - round_intercept = np.round(self.regr.intercept_, 3) - - # Save the weights - feature_names = self.poly_feature_names if self.degree > 1 else self.variable_columns - assert feature_names is not None, 'feature_names should be set at this point' - weights = dict(zip(feature_names, [float(c) for c in round_coef], strict=True)) - weights = dict(sorted(weights.items(), key=lambda item: abs(item[1]), reverse=True)) - weights = {'Bias': float(round_intercept), **weights} - self.weights = weights - - return self - - def predict(self, input_data: pd.DataFrame) -> np.ndarray: - """ - Function to predict the target variable. - If clipping is True, the predictions are clipped based on the target variable quartiles. - - Args: - input_data (pandas.DataFrame): The data used to predict the target variable - - Returns: - numpy.ndarray: The predicted target variable - """ - assert self.variable_columns is not None, 'variable_columns must be set before predict' - X_test: pd.DataFrame = input_data[self.variable_columns].copy() - - # Handle infinite values - X_test = X_test.replace([np.inf, -np.inf], np.nan) - - # Apply polynomial features if degree > 1 - if self.degree > 1: - X_test = self.create_poly_features(X_test, fit=False) - - y_pred = self.regr.predict(X_test) - - if self.clipping: - for i in range(len(y_pred)): - if y_pred[i] > self.clipping['max']: - y_pred[i] = self.q3_target - elif y_pred[i] < self.clipping['min']: - y_pred[i] = self.q1_target - - return y_pred - - def get_regressor(self) -> LinearRegression: - """ - Get the underlying LinearRegression model. - - Returns: - LinearRegression: The sklearn LinearRegression model - """ - return self.regr - - -class DataPreprocessor(BaseEstimator, TransformerMixin): - """ - Data Preprocessor for Time Series Analysis. - - Thread-safety: This class is NOT thread-safe during fit() operations. - Do not call fit() on the same instance from multiple threads simultaneously. - After fitting, transform() is thread-safe for read-only operations IF the - input DataFrames are not shared between threads. - - For multi-threaded environments: - - Fit the preprocessor in a single thread - - Share the fitted instance across threads for transform() only - - Ensure each thread passes its own DataFrame copy to transform() - - Or create separate instances per thread - """ - - def __init__( - self, - date_column: str = '', - target_variable: str = '', - input_columns: list[str] | None = None, - nan_treatment: str | None = None, - lag_train: dict[str, int] | None = None, - lag_transform: dict[str, int] | None = None, - start_date: str | None = None, - end_date: str | None = None, - date_format: str | None = None, - removed_intervals: list[tuple[str, str]] | None = None, - static_threshold: int | None = None, - low_lim: dict[str, float] | None = None, - upp_lim: dict[str, float] | None = None, - window: int | None = None, - scaler_name: str | None = None, - scaler_params: dict[str, Any] | None = None, - ar_var: str | None = None, - self_operations: list[str] | None = None, - cross_operations: list[str] | None = None, - created_lags: dict[str, int] | None = None, - steps_order: list[str] | None = None, - ): - """ - Data Preprocessor for Time Series Analysis - - Args: - date_column (str): The column name of the date in the dataset - target_variable (str): The target variable name - input_columns (list): The input columns names in a list - nan_treatment (str): The treatment for missing values \\ - *Options: 'drop', 'fill linear', 'linear interpolation'* - lag_train (dict): The lags for each variable to be applyed during training \\ - *Format: {'variable_name': lag}* - lag_transform (dict): The lags for each variable to be applyed during transformation \\ - *Format: {'variable_name': lag}* - start_date (str): The start date for filtering data - end_date (str): The end date for filtering data - date_format (str | None): Frontend date format for start/end (e.g. dd/MM/yyyy HH:mm:ss or MM/dd/yyyy HH:mm:ss). When set, parsing matches the CSV date column. - removed_intervals (list): List of tuples with intervals to remove from data \\ - *Format: [('start_date', 'end_date'), ...]* - static_threshold (int): The number of repeated values to be considered as static - low_lim (dict): The lower limits for each variable \\ - *Format: {'variable_name': limit}* - upp_lim (dict): The upper limits for each variable \\ - *Format: {'variable_name': limit}* - window (int): The window size for rolling window. **Not implemented yet** - scaler_name (str): The scaler name. If no scaler is used, it is 'None' \\ - *Options: 'None', 'Standard Scaler'* - scaler_params (dict): The parameters for the scaler object, if it is used \\ - *Format for Standard Scaler: {'variable_name': {'mean': mean, 'variance': variance}}* - ar_var (str): The autoregressive variable name. If None, it is not created - self_operations (list): The operations for feature creation using the same variable \\ - *Format: ['{variable_name}\\_{operation}\\_{scalar}']* \\ - *Operations: 'exp', 'pow', 'log', 'root'* - cross_operations (list): The operations for feature creation using two variables \\ - *Format: ['{variable_name1}\\_{operation}\\_{variable_name2}']* \\ - *Operations: '\\*', '/'* - created_lags (dict): Variables created by lagging existing ones \\ - *Format: {'original_variable_name': lag}* - steps_order (list): The order of the steps to be executed in the pipeline \\ - *Options for list: 'Discontinuity Treatment', - 'Lag Selection', - 'Range Selection & Data Removal', - 'Static Window Removal', - 'Define Variables Limits', - 'Normalization', - 'Feature Creation', - 'Lag Creation'* - - Returns: - DataPreprocessor: The data preprocessor object - """ - self.date_column = date_column - self.target_variable = target_variable - self.input_columns = input_columns - self.nan_treatment = nan_treatment - self.lag_train = lag_train if lag_train else {} - self.lag_transform = lag_transform if lag_transform else {} - self.start_date = start_date - self.end_date = end_date - self.date_format = date_format - self.removed_intervals = removed_intervals if removed_intervals else [] - self.ar_var = ar_var - self.self_operations = self_operations - self.cross_operations = cross_operations - self.created_lags = created_lags - self.static_threshold = static_threshold - self.low_lim = low_lim - self.upp_lim = upp_lim - # self.window = window # Not implemented yet - self.scaler_name = scaler_name - self.scaler_params = scaler_params - self.feature_names_order: list[str] = [] # Initialize to avoid AttributeError - self._fitted_feature_order: list[str] | None = None # Track feature order after fit - - if self.scaler_name == 'Standard Scaler': - self.scaler = StandardScaler() - elif self.scaler_name == 'None': - self.scaler = None - else: - self.scaler = None - - # Filter steps for preprocessor class - possible_steps = [ - DISCONTINUITY_TREATMENT, - LAG_SELECTION, - RANGE_SELECTION, - STATIC_WINDOW_REMOVAL, - DEFINE_VARIABLES_LIMITS, - NORMALIZATION, - FEATURE_CREATION, - LAG_CREATION, - ] - self.steps_order = steps_order or possible_steps - for step in possible_steps: - if step not in self.steps_order: - self.steps_order.append(step) - - def get_required_columns(self, existing_columns: list) -> list: - """ - Get the required columns to generate the input columns - - Args: - existing_columns (list): The existing columns in the data - - Returns: - list: The required columns - """ - required_columns: list[str] = [] - - # Columns for feature creation - if self.self_operations is not None: - for name in self.self_operations: - var, operation, scalar = name.split('}_{') - var = var.split('{')[1] - operation = operation.split('}')[0] - scalar = scalar.split('}')[0] - required_columns.append(var) - if self.cross_operations is not None: - for name in self.cross_operations: - var1, operation, var2 = name.split('}_{') - var1 = var1.split('{')[1] - operation = operation.split('}')[0] - var2 = var2.split('}')[0] - required_columns.append(var1) - required_columns.append(var2) - - # Columns for lag creation - if self.created_lags is not None: - for var in self.created_lags.keys(): - required_columns.append(var) - - # Check if any column in required_columns is not in existing_columns - required_columns = list(set(required_columns)) - _to_remove: list[str] = [] - for column in required_columns: - # If column was already in self_operations list, remove it - if ( - self.self_operations is not None - and column not in existing_columns - and column in self.self_operations - ): - _to_remove.append(column) - # If column was already in cross_operations list, remove it - if ( - self.cross_operations is not None - and column not in existing_columns - and column in self.cross_operations - ): - _to_remove.append(column) - # If column was already in created_lags list, remove it - if ( - self.created_lags is not None - and column not in existing_columns - and column in self.created_lags - ): - _to_remove.append(column) - for column in set(_to_remove): - required_columns.remove(column) - - return required_columns - - def get_scaler(self) -> Any: - """ - Get the scaler object - - Returns: - Scaler: The scaler object - """ - return self.scaler - - def treat_discontinuities(self, input_data: pd.DataFrame) -> pd.DataFrame: - """ - Treat the discontinuities in the data - - Args: - input_data (pandas.DataFrame): The input data - - Returns: - pandas.DataFrame: The treated data - """ - if self.nan_treatment: - # Map 'linear interpolation' to 'fill linear' for compatibility - treatment = self.nan_treatment - if treatment == 'linear interpolation': - treatment = 'fill linear' - 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 using date_format when set. - - When date_format is set (e.g. dd/MM/yyyy HH:mm:ss or MM/dd/yyyy HH:mm:ss), - parsing matches the CSV date column so start_date/end_date filter correctly. - """ - if not date_str: - return None - try: - python_fmt = ( - _frontend_date_format_to_strftime(self.date_format) if self.date_format else None - ) - if python_fmt: - return pd.to_datetime(date_str, format=python_fmt) - 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. - - Args: - input_data (pandas.DataFrame): The input data with datetime index - - Returns: - pandas.DataFrame: The filtered data - """ - 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.removed_intervals: - for interval in self.removed_intervals: - input_data = self._remove_interval(input_data, interval) - - return input_data - - def lag_selection(self, input_data: pd.DataFrame, lag_dict: dict) -> pd.DataFrame: - """ - Select the lags for the variables - - Args: - input_data (pandas.DataFrame): The input data - lag_dict (dict): The lags for each variable \\ - *Format: {'variable_name': lag}* - - Returns: - pandas.DataFrame: The treated data - - Note: - This method modifies input_data in-place. Ensure the caller passes - a copy if the original DataFrame needs to be preserved. - """ - if lag_dict: - for var, lag in lag_dict.items(): - if lag > 0: - input_data[var] = input_data[var].shift(lag) - # WARNING: Modifies DataFrame in-place - input_data.dropna(inplace=True) - return input_data - - def treat_static_windows(self, input_data: pd.DataFrame) -> pd.DataFrame: - """ - Treat the static windows in the data - - Args: - input_data (pandas.DataFrame): The input data - - Returns: - pandas.DataFrame: The treated data - """ - if self.static_threshold: - ts_analyzer = TimeSeriesDiscontinuityAnalyzer(input_data) - ts_analyzer.infer_frequency() - for col in input_data.columns: - ts_analyzer.identify_static_windows(column=col, threshold=self.static_threshold) - ts_analyzer.treat_static_windows( - column=col, remove_window=True, threshold=self.static_threshold - ) - ts_analyzer.update_total_discontinuities(col) - input_data = ts_analyzer.get_treated_data() - return input_data - - def adjust_limits(self, input_data: pd.DataFrame) -> pd.DataFrame: - """ - Adjust the limits for the variables - - Args: - input_data (pandas.DataFrame): The input data - - Returns: - pandas.DataFrame: The treated data - """ - input_data, self.low_lim, self.upp_lim = limit_dataset( - input_data, self.low_lim, self.upp_lim - ) - return input_data - - def create_features(self, input_data: pd.DataFrame) -> pd.DataFrame: - """ - Create features in the data - - Args: - input_data (pandas.DataFrame): The input data - - Returns: - pandas.DataFrame: The treated data - """ - input_data = create_features(input_data, self.self_operations, self.cross_operations) - return input_data - - def create_ar(self, input_data: pd.DataFrame) -> pd.DataFrame: - """ - Create the autoregressive variable in the data - - Args: - input_data (pandas.DataFrame): The input data - - Returns: - pandas.DataFrame: The treated data - - Note: - This method modifies input_data in-place. - """ - if self.ar_var: - input_data[self.ar_var] = input_data[self.target_variable].shift(1) - # WARNING: Modifies DataFrame in-place - input_data.dropna(inplace=True) - return input_data - - def create_lags(self, input_data: pd.DataFrame) -> pd.DataFrame: - """ - Create additional lags in the data - - Args: - input_data (pandas.DataFrame): The input data - - Returns: - pandas.DataFrame: The treated data - - Note: - This method modifies input_data in-place. - """ - if self.created_lags: - for var, lag in self.created_lags.items(): - if lag > 0 and var in input_data.columns: - new_col = f'{var}_lag{lag}' - input_data[new_col] = input_data[var].shift(lag) - # WARNING: Modifies DataFrame in-place - input_data.dropna(inplace=True) - return input_data - - def fit(self, x: pd.DataFrame, y: None | pd.Series = None) -> 'DataPreprocessor': - """ - Function to preprocess the data and split it into training and testing sets - - Args: - x (pandas.DataFrame): The input data - y (pandas.Series): The target variable - - Returns: - DataPreprocessor: The data preprocessor object - """ - if x is not None and y is not None: - data_treat = pd.concat([x.copy(), y.copy()], axis=1) - elif x is not None: - data_treat = x.copy() - else: - raise ValueError('No data was provided') - assert self.input_columns is not None, 'input_columns must be set' - existing_columns = [col for col in data_treat.columns if col in self.input_columns] - data_treat = data_treat[existing_columns + [self.target_variable]] - - for step in self.steps_order: - # Discontinuity Treatment - if step == DISCONTINUITY_TREATMENT: - data_treat = self.treat_discontinuities(data_treat) - - # Lag for Model Training - if step == LAG_SELECTION: - data_treat = self.lag_selection(data_treat, self.lag_train) - - # Range Selection & Data Removal - if step == RANGE_SELECTION: - data_treat = self.range_selection(data_treat) - - # Static Window Treatment - if step == STATIC_WINDOW_REMOVAL: - data_treat = self.treat_static_windows(data_treat) - - # Adjust limits - if step == DEFINE_VARIABLES_LIMITS: - data_treat = self.adjust_limits(data_treat) - - # Normalization - if step == NORMALIZATION and self.scaler: - self.scaler = self.scaler.fit(data_treat[existing_columns]) - self.feature_names_order = list(data_treat[existing_columns].columns) - data_treat[existing_columns] = self.scaler.transform(data_treat[existing_columns]) - - # Save scaler parameters - assert self.scaler_params is not None, 'scaler_params must be initialized' - for index, column in enumerate(list(existing_columns)): - mean = self.scaler.mean_[index] - variance = self.scaler.var_[index] - self.scaler_params[column] = { - 'mean': round(mean, 3), - 'variance': round(variance, 3), - } - - # Store fitted feature order for predict method - self._fitted_feature_order = list(existing_columns) - - return self - - def transform(self, x: pd.DataFrame) -> pd.DataFrame: - """ - Function to preprocess the data - - Args: - x (pandas.DataFrame): The input data - - Returns: - pandas.DataFrame: The treated data - """ - if 'timestamp' in x.columns: - data_treat = x.drop(columns='timestamp') - else: - data_treat = x.copy() - assert self.input_columns is not None, 'input_columns must be set' - existing_columns = [col for col in data_treat.columns if col in self.input_columns] - required_columns = self.get_required_columns(existing_columns) - all_cols = required_columns + existing_columns + [self.target_variable] - all_cols = list(set(all_cols)) - data_treat = data_treat[all_cols] - - for step in self.steps_order: - # Discontinuity Treatment - if step == DISCONTINUITY_TREATMENT: - data_treat = self.treat_discontinuities(data_treat) - - # Lag for Model Training - if step == LAG_SELECTION: - data_treat = self.lag_selection(data_treat, self.lag_transform) - - # Range Selection & Data Removal (typically skipped in transform) - if step == RANGE_SELECTION: - data_treat = self.range_selection(data_treat) - - # Static Window Treatment - if step == STATIC_WINDOW_REMOVAL: - data_treat = self.treat_static_windows(data_treat) - - # Adjust limits - if step == DEFINE_VARIABLES_LIMITS: - data_treat = self.adjust_limits(data_treat) - - # Normalization - if step == NORMALIZATION and self.scaler: - # Only transform feature columns, preserve target and any other required columns - feature_cols = self.feature_names_order - data_treat[feature_cols] = self.scaler.transform(data_treat[feature_cols]) - - # Feature Creation - if step == FEATURE_CREATION: - data_treat = self.create_features(data_treat) - - # Lag Creation - if step == LAG_CREATION: - # Autoregressive Variable - if self.input_columns is not None and self.ar_var in self.input_columns: - data_treat = self.create_ar(data_treat) - - # Additonal Lags - data_treat = self.create_lags(data_treat) - - return data_treat - - def predict(self, x: pd.DataFrame) -> pd.DataFrame: - """ - Transform data for prediction (removes target variable). - - This method is a wrapper around transform() that: - 1. Transforms the input data - 2. Removes the target variable column - 3. Ensures features are in the same order as during fit - - Args: - x (pandas.DataFrame): The input data - - Returns: - pandas.DataFrame: The transformed data without target variable, - with features in the same order as during fit - """ - data_treat = self.transform(x) - - # Remove target variable if present - if self.target_variable in data_treat.columns: # pragma: no branch - data_treat = data_treat.drop(columns=self.target_variable) - - # Ensure features are in the same order as during fit - if self._fitted_feature_order is not None: - # Filter to only include columns that exist in both - available_cols = [c for c in self._fitted_feature_order if c in data_treat.columns] - data_treat = data_treat[available_cols] - - return data_treat diff --git a/model_manager/utils/models/train_model_params.py b/model_manager/utils/models/train_model_params.py index 21f40b7..9d1f7b8 100644 --- a/model_manager/utils/models/train_model_params.py +++ b/model_manager/utils/models/train_model_params.py @@ -3,7 +3,25 @@ from typing import Any from jsonschema import Draft202012Validator, ValidationError # type: ignore[import-untyped] -from model_manager.sientia.models import validate_frontend_date_format +# Allowed frontend date formats and their strftime equivalents (single source of truth) +FRONTEND_DATE_FORMAT_TO_STRFTIME = { + 'dd/MM/yyyy HH:mm:ss': '%d/%m/%Y %H:%M:%S', + 'MM/dd/yyyy HH:mm:ss': '%m/%d/%Y %H:%M:%S', + 'yyyy/MM/dd HH:mm:ss': '%Y/%m/%d %H:%M:%S', + 'dd-MM-yyyy HH:mm:ss': '%d-%m-%Y %H:%M:%S', + 'MM-dd-yyyy HH:mm:ss': '%m-%d-%Y %H:%M:%S', + 'yyyy-MM-dd HH:mm:ss': '%Y-%m-%d %H:%M:%S', +} +ALLOWED_FRONTEND_DATE_FORMATS = frozenset(FRONTEND_DATE_FORMAT_TO_STRFTIME.keys()) + + +def validate_frontend_date_format(fmt: str | None) -> None: + """Raise ValueError if fmt is set and not one of the allowed frontend date formats.""" + if not fmt or not fmt.strip(): + return + if fmt not in ALLOWED_FRONTEND_DATE_FORMATS: + allowed = ', '.join(sorted(ALLOWED_FRONTEND_DATE_FORMATS)) + raise ValueError(f'Invalid date_format "{fmt}". Allowed formats: {allowed}') # Model name constants MODEL_LINEAR_REGRESSION = 'Linear Regression'