""" Model Evaluation Pipelines """ from absl import logging import os import time import mlflow import seaborn as sns from sklearn.metrics import ( mean_absolute_error, mean_absolute_percentage_error, mean_squared_error ) from uuid import uuid1 from sklearn.model_selection import TimeSeriesSplit # set mlflow tracking server TRACKING_SERVER = os.environ.get("TRACK_URI", "https://dev-orch-mlflow-service.dev.theorchard.io/") mlflow.set_tracking_uri(TRACKING_SERVER) # default test size (7 days) default_timeseries_splitcv = TimeSeriesSplit(n_splits=5, test_size=7, gap=1) def evaluate_generalization_performance_sklearn(model_factory, X, y, cv=default_timeseries_splitcv, model_params={}, results_store={}, mlflow_experiment="data-platform/debut-forecasting", mlflow_experiment_tags={ "team": "data-platform", "data-type": "time-series" }): """ Evaluates the generalisation performance of the a pipeline as a whole using Time series Cross Validation Time series Cross validation allows us to evaluate models in a rolling fashion - so our dataset will be trained as follows for 3 fold cross validation X = [x_1, x2_,x_3 ....x_n] fold_1 : - X_train[x_1], X_test[x_2,x_3 ....x_n] fold_2: - X_train[x_1, x_2], X_test[x_3,x_4 ....x_n] fold_3: - X_train[x_1, x_2, x_3], X_test[x_4 ....x_n] params: model_factory - model factory to evaluate the mode X (DataFrame/numpy-array) Features with shape (n_samples, n_features) y (Series) - Target variables with shape (n_samples) results_store (dictionary) - dictionary to store results mlflow_experiment (str) - mlflow experiment name mlflow_experiment_tags (dict) - mlflow experiment tags returns: None """ # create experiment (if not exist) mlflow.set_experiment(f"{mlflow_experiment}") mlflow.sklearn.autolog() with mlflow.start_run(run_name=mlflow_experiment + f"/{uuid1()}", tags=mlflow_experiment_tags): # log model params mlflow.log_params(model_params) # Timseries Cross validation for cv_fold, (train_indicies, val_indicies) in enumerate(cv.split(X,y)): cv_fold_str = f'fold_{cv_fold + 1}' with mlflow.start_run(run_name=f"{cv_fold_str}", nested=True) as cv_fold_run: logging.info(f'**Running CV Fold : {cv_fold + 1}**') try: results_store[cv_fold_str] = {} logging.debug(results_store.keys()) #split train validation split X_train, X_val = X[train_indicies], X[val_indicies] y_train, y_val = y[train_indicies], y[val_indicies] # log train-val split mlflow.log_params({"train_batch_size": len(train_indicies), "val_batch_size": len(val_indicies)}) # create model logging.info('Creating model with model factory using defined parameters') model = model_factory(model_params=model_params) # fit model logging.info('Fitting Model') fit_start = time.time() fitted_model = model.fit(X_train, y_train) fit_took_seconds = time.time() - fit_start results_store[cv_fold_str]["fit_time_seconds"] = fit_took_seconds mlflow.log_metrics({"fit_time_seconds": fit_took_seconds}, step=cv_fold) logging.info('Evaluating Model') logging.debug('-running inference') # predict using dataset predict_time = time.time() y_hat_train = fitted_model.predict(X_train) y_hat_val = fitted_model.predict(X_val) predict_took_seconds = time.time() - predict_time results_store[cv_fold_str]["inference_time_seconds"] = predict_took_seconds # plot mlflow.log_metrics({"inference_time_seconds": predict_took_seconds}, step=cv_fold) # evaluate metrics logging.debug('-evaluating forecast predictions') # MAE mae_train = mean_absolute_error(y_pred=y_hat_train, y_true=y_train) mae_val = mean_absolute_error(y_pred=y_hat_val, y_true=y_val) mlflow.log_metrics({"mae_val": mae_val, "mae_train": mae_train}, step=cv_fold) # save results results_store[cv_fold_str]["mae_val"] = mae_val results_store[cv_fold_str]["mae_train"] = mae_train logging.debug(f'MAE (train) : {results_store[cv_fold_str]["mae_train"]: .2f}') logging.debug(f'MAE (val) : {results_store[cv_fold_str]["mae_val"]: .2f}') # RMSE mse_train = mean_squared_error(y_pred=y_hat_train, y_true=y_train, squared=True) mse_val = mean_squared_error(y_pred=y_hat_val, y_true=y_val, squared=True) # save results results_store[cv_fold_str]["mse_train"] = mse_train results_store[cv_fold_str]["mse_val"] = mse_val logging.debug(f'MSE (train) : {results_store[cv_fold_str]["mse_train"]: .2f}') logging.debug(f'MSE (val) : {results_store[cv_fold_str]["mse_val"]: .2f}\n') # log mse mlflow.log_metrics({"mse_train": mse_train, "mse_val": mse_val}, step=cv_fold) # MAPE mape_train = mean_absolute_percentage_error(y_pred=y_hat_train, y_true=y_train) mape_val = mean_absolute_percentage_error(y_pred=y_hat_val, y_true=y_val) # save results results_store[cv_fold_str]["mape_train"] = mape_train results_store[cv_fold_str]["mape_val"] = mape_val logging.debug(f'MAPE (train) : {results_store[cv_fold_str]["mape_train"]: .2f}') logging.debug(f'MAPE (val) : {results_store[cv_fold_str]["mape_val"]: .2f}\n') # log mape mlflow.log_metrics({"mape_train": mape_train, "mape_val": mape_val}, step=cv_fold) except Exception as err: logging.debug("Something went wrong while evaluating model") logging.error(err)