diff --git a/model_manager/sientia/__init__.py b/model_manager/sientia/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/model_manager/sientia/exceptions.py b/model_manager/sientia/exceptions.py new file mode 100644 index 0000000..8a8cca4 --- /dev/null +++ b/model_manager/sientia/exceptions.py @@ -0,0 +1,3 @@ +from mlflow.exceptions import MlflowException + +SientiaMlException = MlflowException diff --git a/model_manager/sientia/metrics.py b/model_manager/sientia/metrics.py new file mode 100644 index 0000000..184cb4b --- /dev/null +++ b/model_manager/sientia/metrics.py @@ -0,0 +1,28 @@ +import numpy as np +import pandas as pd +from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score + + +def mse(real_data: pd.Series, predictions: pd.Series) -> float: + """ + Calculates the mean squared error between the real data and the predictions. + """ + return round( + mean_squared_error(real_data.astype(np.float64), predictions.astype(np.float64)), 2 + ) + + +def mae(real_data: pd.Series, predictions: pd.Series) -> float: + """ + Calculates the mean absolute error between the real data and the predictions. + """ + return round( + mean_absolute_error(real_data.astype(np.float64), predictions.astype(np.float64)), 2 + ) + + +def r2(real_data: pd.Series, predictions: pd.Series) -> float: + """ + Calculates the R2 score between the real data and the predictions. + """ + return round(r2_score(real_data.astype(np.float64), predictions.astype(np.float64)), 2) diff --git a/model_manager/sientia/model_serving.py b/model_manager/sientia/model_serving.py new file mode 100644 index 0000000..b0b6c76 --- /dev/null +++ b/model_manager/sientia/model_serving.py @@ -0,0 +1,158 @@ +import logging +import os +from typing import Any + +import mlflow +import mlflow.sklearn +import pandas as pd + +from model_manager.sientia.exceptions import SientiaMlException + + +class ModelServing: + def __init__( + self, + tracking_uri: str, + username: str | None = None, + password: str | None = None, + logger: Any | None = None, + ): + # set tracking uri + mlflow.set_tracking_uri(tracking_uri) + + if username is not None: + os.environ['MLFLOW_TRACKING_USERNAME'] = username + if password is not None: + os.environ['MLFLOW_TRACKING_PASSWORD'] = password + # Create an MLflow client + self.client = mlflow.tracking.MlflowClient() + + # Function to list runs for a given experiment + def search_runs_by_name( + self, experiment_names: list[str], order_by: None | list[str] = None + ) -> pd.DataFrame: + """ + List runs for a specified MLflow experiment. + + Args: + experiment_names (list[str]): List with experiment_names to retrieve runs from. + + Returns: + pandas.DataFrame: A DataFrame containing run information. + + Raise: + SientiaMlException if unable to search runs + """ + try: + runs = mlflow.search_runs(experiment_names=experiment_names, order_by=order_by) + except SientiaMlException as e: + logging.error(e) + raise SientiaMlException from e + return runs + + def set_experiment(self, experiment_identifier: str) -> None: + """ + Set the given experiment as the active experiment. + + Args: + experiment_identifier (str): name or id of the experiment to be setted + """ + mlflow.set_experiment(experiment_identifier) + + def log_model(self, sk_model: Any, artifact_path: Any, **kwargs) -> None: + """ + Log a sklearn model. + + Args: + sk_model: scikit-learn model to be saved. + artifact_path: Run-relative artifact path. + + Returns: + None + """ + mlflow.sklearn.log_model( + sk_model, + artifact_path, + extra_pip_requirements=[ + 'git+https://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@github.com/Aignosi/sientia-mlops-library.git' + ], + **kwargs, + ) + + def log_param(self, key: str, value: Any) -> None: + """ + Log a param in the active run. + + Args: + key (str): Param name + value (any): Param value + + Returns: + None + """ + mlflow.log_param(key, value) + + def log_metric(self, key: str, value: Any) -> None: + """ + Log a metric in the active run. + + Args: + key (str): Metric name + value (any): Metric value + + Returns: + None + """ + mlflow.log_metric(key, value) + + def log_artifact( + self, local_path: str, artifact_path: str | None = None, run_id: str | None = None + ) -> None: + """ + Log an artifact. + + Args: + local_path: Local path of the artifact to log. + artifact_path: If provided, the directory in artifact_uri to write to. + run_id: optional id of current run + + Returns: + None + """ + mlflow.log_artifact(local_path=local_path, artifact_path=artifact_path, run_id=run_id) + + def save_experiment( + self, + run_id: str | None = None, + experiment_id: str | None = None, + run_name: str | None = None, + nested: bool = False, + tags: dict[str, Any] | None = None, + description: str | None = None, + log_system_metrics: bool | None = None, + ) -> mlflow.ActiveRun: + """ + Save a experiment. + + Args: + run_id: If specified, get the run with the specified UUID and log parameters and metrics under that run. + experiment_id: ID of the experiment under which to create the current run (applicable only when run_id is not specified). + run_name: Name of new run. Used only when run_id is unspecified. + nested: Controls whether run is nested in parent run. True creates a nested run. + tags: An optional dictionary of string keys and values to set as tags on the run. If a run is being resumed, these tags are set on the resumed run. If a new run is being created, these tags are set on the new run. + description: An optional string that populates the description box of the run. + log_system_metrics: If True, system metrics will be logged. If None, we will check environment variable + + Returns: + ActiveRun: object that acts as a context manager wrapping the run's state. + """ + run = mlflow.start_run( + run_id=run_id, + experiment_id=experiment_id, + run_name=run_name, + nested=nested, + tags=tags, + description=description, + log_system_metrics=log_system_metrics, + ) + return run diff --git a/model_manager/sientia/models.py b/model_manager/sientia/models.py new file mode 100644 index 0000000..fd3cacc --- /dev/null +++ b/model_manager/sientia/models.py @@ -0,0 +1,511 @@ +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 StandardScaler + + +class LinearRegressionModel(BaseEstimator, TransformerMixin): + 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, + ): + """ + 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}* + + 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 + + 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 + """ + assert self.variable_columns is not None, 'variable_columns must be set before fitting' + X_train = input_data[self.variable_columns] + y_train = input_data[self.target_variable] + + self.q1_target = y_train.quantile(0.25) + self.q3_target = y_train.quantile(0.75) + + # 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 + weights = dict(zip(self.variable_columns, [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 + """ + X_test = input_data[self.variable_columns] + 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 + + +class DataPreprocessor(BaseEstimator, TransformerMixin): + 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, + 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'* + 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}* + 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', + '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.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 + self.scaler_name = scaler_name + self.scaler_params = scaler_params + 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', + '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: + input_data = treat_nan(input_data, self.nan_treatment) + 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 + """ + if lag_dict: + for var, lag in lag_dict.items(): + if lag > 0: + input_data[var] = input_data[var].shift(lag) + 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 + """ + if self.ar_var: + input_data[self.ar_var] = input_data[self.target_variable].shift(1) + 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 + """ + 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) + 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) + + # 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': + if 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), + } + + 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) + + # 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': + if self.scaler: + data_treat = data_treat[self.feature_names_order] + data_treat[existing_columns] = self.scaler.transform( + data_treat[existing_columns] + ) + + # 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 diff --git a/model_manager/sientia/reports.py b/model_manager/sientia/reports.py new file mode 100644 index 0000000..7a12fee --- /dev/null +++ b/model_manager/sientia/reports.py @@ -0,0 +1,226 @@ +import os +from collections.abc import Sequence +from typing import Any + +from bs4 import BeautifulSoup +from evidently.metric_preset import DataDriftPreset +from evidently.metrics import ( + ColumnSummaryMetric, + ConflictTargetMetric, + DatasetCorrelationsMetric, + DatasetSummaryMetric, + RegressionAbsPercentageErrorPlot, + RegressionDummyMetric, + RegressionErrorDistribution, + RegressionErrorPlot, + RegressionPerformanceMetrics, + RegressionPredictedVsActualPlot, + RegressionPredictedVsActualScatter, +) +from evidently.metrics.base_metric import generate_column_metrics +from evidently.options import ColorOptions +from evidently.report import Report + +COLOR_DISCRETE_SEQUENCE = ( + '#ed0400', + '#0a5f38', + '#6c3461', + '#71aa34', + '#d8dcd6', + '#6b8ba4', +) + + +def load_html_from_file(file_path): + try: + with open(file_path, encoding='utf-8') as file: + return file.read() + except FileNotFoundError: + print(f'File not found: {file_path}') + return None + except OSError as e: # noqa: BLE001 + print(f'Error reading file: {e}') + return None + + +def inject_content(main_html, section_id, content): + soup = BeautifulSoup(main_html, 'html.parser') + section = soup.find(id=section_id) + if section: + section.clear() + section.append(BeautifulSoup(content, 'html.parser')) + else: + print(f"Section with id '{section_id}' not found in the main HTML template.") + return str(soup) + + +class Reports: + def __init__( + self, reference_data: Any, current_data: Any, base_path: str | None = None + ) -> None: + """ + Initializes an instance of the AigReport class. + + Args: + reference_data: The reference data for the report. + current_data: The current data for the report. + base_path: The base path for the report. + """ + self.metrics: list[Any] = [] + self.options: list[Any] | None = None + self.sections: dict[str, Any] = {} + self.report: Any = None + self.ref_data = reference_data + self.cur_data = current_data + self.set_color_options(primary_color='#0F4C81', secondary_color='#001E60') + self.base_path = base_path + + def add_data_quality_section(self, columns: list[str] | None = None, run: bool = True) -> None: + """ + Adds a data quality section to the report. + + Args: + columns: The list of columns to include in the data quality section. If None, all columns will be included. + run: Indicates whether to run the report immediately after adding the section. + """ + metrics = [ + DatasetSummaryMetric(), + generate_column_metrics(ColumnSummaryMetric, columns=columns, skip_id_column=True), + ConflictTargetMetric(), + DatasetCorrelationsMetric(), + ] + self.metrics.extend(metrics) + if run: + report = Report(metrics=metrics, options=self.options) + report.run(reference_data=self.ref_data, current_data=self.cur_data) + self.sections['data_quality'] = report.as_dict() + if self.base_path: + report.save_html(os.path.join(self.base_path, 'data_quality.html')) + + def add_data_drift_section(self, columns: list[str] | None = None, run: bool = True) -> None: + """ + Adds a data drift section to the report. + + Args: + columns: The list of columns to include in the data drift section. If None, all columns will be included. + run: Indicates whether to run the report immediately after adding the section. + """ + self.metrics.append(DataDriftPreset(columns=columns)) + if run: + report = Report(metrics=[DataDriftPreset(columns=columns)], options=self.options) + report.run(reference_data=self.ref_data, current_data=self.cur_data) + self.sections['data_drift'] = report.as_dict() + if self.base_path: + report.save_html(os.path.join(self.base_path, 'data_drift.html')) + + def add_regression_section(self, run: bool = True) -> None: + """ + Adds a regression section to the report. + + Args: + run: Indicates whether to run the report immediately after adding the section. + """ + metrics = [ + RegressionPerformanceMetrics(), + RegressionDummyMetric(), + RegressionPredictedVsActualScatter(), + RegressionPredictedVsActualPlot(), + RegressionErrorPlot(), + RegressionAbsPercentageErrorPlot(), + RegressionErrorDistribution(), + ] + self.metrics.extend(metrics) + if run: + report = Report(metrics=metrics, options=self.options) + report.run(reference_data=self.ref_data, current_data=self.cur_data) + self.sections['regression'] = report.as_dict() + if self.base_path: + report.save_html(os.path.join(self.base_path, 'regression.html')) + + def set_color_options( + self, + primary_color: str = '#0F4C81', + secondary_color: str = '#001E60', + current_data_color: str | None = None, + reference_data_color: str | None = None, + additional_data_color: str = '#0a5f38', + color_sequence: Sequence[str] = COLOR_DISCRETE_SEQUENCE, + fill_color: str = 'LightGreen', + zero_line_color: str = 'green', + non_visible_color: str = 'white', + underestimation_color: str = '#6574f7', + overestimation_color: str = '#ee5540', + majority_color: str = '#1acc98', + vertical_lines: str = 'green', + heatmap: str = 'RdBu_r', + ) -> None: + """ + Sets the color options for the report. + + Args: + primary_color: The primary color for the report. + secondary_color: The secondary color for the report. + current_data_color: The color for the current data. + reference_data_color: The color for the reference data. + additional_data_color: The color for additional data. + color_sequence: The color sequence for discrete values. + fill_color: The fill color for visualizations. + zero_line_color: The color for the zero line. + non_visible_color: The color for non-visible elements. + underestimation_color: The color for underestimation. + overestimation_color: The color for overestimation. + majority_color: The color for majority elements. + vertical_lines: The color for vertical lines. + heatmap: The color map for heatmaps. + """ + color_scheme = ColorOptions( + primary_color=primary_color, + secondary_color=secondary_color, + current_data_color=current_data_color, + reference_data_color=reference_data_color, + additional_data_color=additional_data_color, + color_sequence=color_sequence, + fill_color=fill_color, + zero_line_color=zero_line_color, + non_visible_color=non_visible_color, + underestimation_color=underestimation_color, + overestimation_color=overestimation_color, + majority_color=majority_color, + vertical_lines=vertical_lines, + heatmap=heatmap, + ) + + if self.options is None: + self.options = [color_scheme] + else: + self.options.append(color_scheme) + + def save_all_sections_html(self, report_path): + """ + Saves the report with all sections as HTML. + + Args: + report_path: The path to save the report HTML file. + """ + if not self.base_path: + raise ValueError('base_path is required to save all sections HTML') + + # Load main HTML template + main_html_path = os.path.join(self.base_path, 'header.html') + main_html = load_html_from_file(main_html_path) + + # Load content from data_drift.html, data_quality.html, and regression.html + data_drift_content = load_html_from_file(os.path.join(self.base_path, 'data_drift.html')) + data_quality_content = load_html_from_file( + os.path.join(self.base_path, 'data_quality.html') + ) + regression_content = load_html_from_file(os.path.join(self.base_path, 'regression.html')) + + # Inject content into the main HTML template + main_html = inject_content(main_html, 'data_drift', data_drift_content) + main_html = inject_content(main_html, 'data_quality', data_quality_content) + main_html = inject_content(main_html, 'regression', regression_content) + + # Save the final HTML to a new file (report.html) + with open(report_path, 'w', encoding='utf-8') as report_file: + report_file.write(main_html) diff --git a/model_manager/sientia/utils.py b/model_manager/sientia/utils.py new file mode 100644 index 0000000..38d5582 --- /dev/null +++ b/model_manager/sientia/utils.py @@ -0,0 +1,36 @@ +from typing import Any + +from numpy.typing import ArrayLike +from sklearn.model_selection import train_test_split + + +def split_train_test( + *data: Any, + test_size: float | None = None, + train_size: float | None = None, + random_state: int | None = None, + shuffle: bool = True, + stratify: ArrayLike | None = None, +) -> tuple[Any, Any, Any, Any]: + """ + Split arrays or matrices into random train and test subsets. + + Args: + *data: data to be splitted. + test_size: size of test subset. + train_size: size of train subset. + random_state: Seed applied to the data before applying the split. + shuffle: Whether or not to shuffle the data before splitting. + stratify: If not None, data is split in a stratified fashion, using this as the class labels. + Returns: + X_train, X_test, y_train, y_test + """ + X_train, X_test, y_train, y_test = train_test_split( + *data, + test_size=test_size, + train_size=train_size, + random_state=random_state, + shuffle=shuffle, + stratify=stratify, + ) + return X_train, X_test, y_train, y_test diff --git a/model_manager/utils/models/train_model_result.py b/model_manager/utils/models/train_model_result.py index 189cfe0..13bf150 100644 --- a/model_manager/utils/models/train_model_result.py +++ b/model_manager/utils/models/train_model_result.py @@ -1,9 +1,8 @@ from dataclasses import dataclass import pandas as pd -from sientia.linear_models import LinearRegressionModel -from sientia.preprocessing import DataPreprocessor +from model_manager.sientia.models import DataPreprocessor, LinearRegressionModel from model_manager.utils.models.train_model_params import TrainModelParams @@ -40,8 +39,8 @@ class TrainModelResult: process_data: DataPreprocessor x_train: pd.DataFrame x_test: pd.DataFrame - y_train: pd.DataFrame - y_test: pd.DataFrame + y_train: pd.Series + y_test: pd.Series regr: LinearRegressionModel scaler_dict: dict y_pred: pd.Series | None = None diff --git a/model_manager/utils/repository/model_repository.py b/model_manager/utils/repository/model_repository.py index 6c74a5e..dcde880 100644 --- a/model_manager/utils/repository/model_repository.py +++ b/model_manager/utils/repository/model_repository.py @@ -15,10 +15,10 @@ from os import makedirs, path import numpy as np import pandas as pd -from sientia.ModelServing import ModelServing # type: ignore[import-untyped] -from sientia.reports import Reports # type: ignore[import-untyped] from sientia_do.observability.logger import Logger +from model_manager.sientia.model_serving import ModelServing # type: ignore[import-untyped] +from model_manager.sientia.reports import Reports # type: ignore[import-untyped] from model_manager.utils.models.train_model_result import TrainModelResult diff --git a/model_manager/utils/repository/training_repository.py b/model_manager/utils/repository/training_repository.py index 88833f0..79a4fcf 100644 --- a/model_manager/utils/repository/training_repository.py +++ b/model_manager/utils/repository/training_repository.py @@ -10,14 +10,13 @@ from io import BytesIO import numpy as np import pandas as pd -from sientia.linear_models import LinearRegressionModel -from sientia.metrics import mae, mse, r2 -from sientia.preprocessing import DataPreprocessor -from sientia.utils import split_train_test from sientia_do.observability.logger import Logger from sientia_do.operations.df_preprocessor import load_data from sientia_do.operations.normalization import MinMaxScaler, Z_Scaler +from model_manager.sientia.metrics import mae, mse, r2 +from model_manager.sientia.models import DataPreprocessor, LinearRegressionModel +from model_manager.sientia.utils import split_train_test from model_manager.utils.models.train_model_params import TrainModelParams from model_manager.utils.models.train_model_result import TrainModelResult @@ -174,7 +173,7 @@ class TrainingRepository: and metrics (mse_val, mae_val, r2_val) """ # Make predictions on test set - tmr.y_pred = tmr.regr.predict(tmr.x_test) + y_pred_array = tmr.regr.predict(tmr.x_test) # Denormalize data if scaler was used if params.use_scaler: @@ -188,10 +187,10 @@ class TrainingRepository: # Denormalize target variable tmr.y_train = scaler.denormalize_single_input(tmr.y_train, params.target_variable) tmr.y_test = scaler.denormalize_single_input(tmr.y_test, params.target_variable) - tmr.y_pred = scaler.denormalize_predictions(tmr.y_pred, params.target_variable) + y_pred_array = scaler.denormalize_predictions(y_pred_array, params.target_variable) # Add index to predictions - tmr.y_pred = pd.Series(tmr.y_pred, index=tmr.y_test.index) + tmr.y_pred = pd.Series(y_pred_array, index=tmr.y_test.index) tmr.y_pred.name = f'{params.target_variable}_pred' # Reorder all data by index @@ -202,6 +201,7 @@ class TrainingRepository: tmr.y_pred = tmr.y_pred.sort_index() # Calculate evaluation metrics + assert tmr.y_pred is not None, 'y_pred should be set at this point' tmr.mse_val = round( mse(tmr.y_test.astype(np.float64), tmr.y_pred.astype(np.float64)), 2, diff --git a/pyproject.toml b/pyproject.toml index a7d7439..7356343 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -98,11 +98,23 @@ module = "prometheus_client.*" ignore_missing_imports = true [[tool.mypy.overrides]] -module = "sientia.*" +module = "pandas.*" ignore_missing_imports = true [[tool.mypy.overrides]] -module = "pandas.*" +module = "bs4.*" +ignore_missing_imports = true + +[[tool.mypy.overrides]] +module = "evidently.*" +ignore_missing_imports = true + +[[tool.mypy.overrides]] +module = "sklearn.*" +ignore_missing_imports = true + +[[tool.mypy.overrides]] +module = "yaml" ignore_missing_imports = true [tool.pytest.ini_options] diff --git a/requirements.txt b/requirements.txt index 57c0af1..c928484 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,8 +1,11 @@ -temporalio -psycopg2-binary -sqlalchemy -boto3 -botocore +temporalio==1.18.1 +psycopg2-binary==2.9.11 +sqlalchemy==2.0.44 +boto3==1.40.55 +botocore==1.40.55 git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.6 -git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.39.0 -prometheus-client +prometheus-client==0.23.1 +mlflow==2.10.1 +evidently==0.4.21 +beautifulsoup4==4.12.3 +scikit-learn==1.4.2