import logging import time from scikit_transformers import do_transformations from sklearn.ensemble import RandomForestRegressor # ,TODO try ExtraTreesRegressor from sklearn.model_selection import train_test_split from sklearn.utils import shuffle from utils.data_processor import merge_csv from utils.data_sets import * logging.getLogger().setLevel(logging.INFO) from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score import numpy as np from scikit_plot_metrics import draw_feature_importance import mlflow.sklearn import sys def eval_metrics(actual, pred): """ Calculate evaluation metrics :param actual: :param pred: :return: """ #print(repr(type(actual))) #print('---------------') #print(repr(pred)) rmse = np.sqrt(mean_squared_error(actual, pred)) mae = mean_absolute_error(actual, pred) r2 = r2_score(actual, pred) # Calculate roc curve # TODO! implement for Regressor """ Idea for the above implementation: - turn pred values into binary at a given threshold - thresholds contain in a pip range of actual & pred value diff This would then allow to calculate roc (and use that to guide model selection). """ #roc_auc = auc(fpr, tpr) #fpr, tpr, thresholds = roc_curve(actual, pred) # limited to binary classificiation #acc = roc_auc_score(actual, np.array(pred)) # ROC area under curve fpr, tpr, roc_auc, acc = None, None, None, None return rmse, mae, r2, acc, fpr, tpr, roc_auc def train_model(model_name, df, cols, target): """ Train a model using df as a dateset. Features defined in cols, and label in target. Save artifacts with "now" argument. """ logging.info("Shuffling training dataset...") train_data = shuffle(df) x = train_data[cols] y = train_data[target] # to free up some RAM. #TODO we shouldn't have to do this, but helps with local when limited resources and big data train_data = None # TODO! do more advanced interpolation instead fillna """ # null interpolation """ # fill NaN with 0 x = x.fillna(0) y = y.fillna(0) logging.info("Split data into training and scoring sets ...") X_train, X_test, y_train, y_test = train_test_split(x, y, test_size=0.10, random_state=75844) #gbt = GradientBoostingClassifier(n_estimators=100) # TODO! gridsearch rf = RandomForestRegressor(n_estimators=300, oob_score=True, max_features=None, n_jobs=-1) # All features """ # Model training """ logging.info("Fit model to data ...") train = rf.fit(X_train, y_train, sample_weight=None) # y_train # Model hyperparams # Draw feature importance logging.info("Drawing feature importance...") feature_importance = train.feature_importances_ estimators = train.estimators_ feature_importance_plot = draw_feature_importance(feature_importance, cols) feature_importance_plot.savefig('data/ml_models/{}_feature_importance.png'.format(model_name)) """ # Model evaluation """ prediction = rf.predict(X_test) # [:, 1] # Evaluate basic metrics logging.info("Running evaluation metrics...") actual = y_test # y_test pred = prediction (rmse, mae, r2, acc, fpr, tpr, roc_auc) = eval_metrics(actual, pred) # Draw a roc curve plot # TODO! fix logging.info("Skipping drawing roc area under curve...") #roc_curve_plot = draw_roc_curve(fpr, tpr, roc_auc) #roc_curve_plot.savefig('data/ml_models/rf_{}_roc_auc.png'.format(model_name)) # , bbox_inches='tight' logging.info("Done modelling!") #joblib.dump(rf, 'data/ml_models/rf_{}.pkl'.format(model_name)) return rf, rmse, mae, r2, acc, fpr, tpr, roc_auc, feature_importance, estimators if __name__ == '__main__': start_time = time.time() # TODO! when calling this, provide in arguments for source, target ccy & time unit # Set model parameters. company_id = sys.argv[1] runtype = sys.argv[2] # sets tag names for analytics, use'manual' or 'auto' runmode = sys.argv[3] # sets tag names for analytics, use'prod' or 'dev' model_name = 'rf_' + company_id # Get data logging.info("Training model {}. Loading data...".format(model_name)) #df, cols, cat_cols, index_col, target = get_feature_vector('training') csv1 = 'data/TMW-2019-Festival-Conference-Pass-FREE-24.10.2019.csv' csv2 = 'data/TMW-2019-Festival-Conference-Pass-FREE-24.10.2019_enriched.csv' df = merge_csv(csv1, csv2, 'email', 'email', 'inner') df = df.fillna(0) cat_cols = ['skip3', 'gender', 'attendeeType'] cols = ['skip3', 'gender', 'age', 'attendeeType'] target = 'quantity' # Do transformations # Do transformations if there are predefined categorical features #df[cols].to_csv('dataframe.csv') print(df[cols].dtypes) # Need to convert datatypes of columns to strings if they are objects df[cat_cols] = df[cat_cols].astype(str) print(df[cols].dtypes) df = do_transformations(df=df, categorical_columns=cat_cols) logging.info(f'Processed data sample: {df[cols]}') """ We are doing this split here because we are too lazy to get another data set for predictions """ # Take the first 90% for training model, and 10% for predicting scores. first_slice = int(len(df.index) * 0.9) logging.info("Slicing dataset...training set: {} rows; testing set: {} rows".format( first_slice, len(df.index) - first_slice)) train_df = df.iloc[1 : first_slice] test_df = df.iloc[first_slice+1: -1] # TODO! add this somewhere - ALTER TABLE public.params ALTER COLUMN value TYPE VARCHAR; # mlflow.tracking.set_tracking_uri('postgresql+psycopg2://{user}:{password}@{host}/{database}'.format(**db_args)) try: """ Even though 'set_experiment' creates new when not exists, we explicitly create it to have control over artifact_location (set_experiment) doesn't expose this option """ mlflow.create_experiment(model_name) # , artifact_location="s3://fansifter-model-artifacts" mlflow.set_experiment(model_name) logging.info("Created new experiment {}".format(model_name)) except Exception as e: logging.info("Using existing experiment {}".format(model_name)) mlflow.set_experiment(model_name) # Creates a new one if doesn't exist with same name #mlflow.set_experiment(model_name) with mlflow.start_run(): mlflow.set_tag('runtype', runtype) mlflow.set_tag('runmode', runmode) # TODO! in a fine tuning experiment, feed in different class_weight types (e.g. balanced_subsample) (rf, rmse, mae, r2, acc, fpr, tpr, roc_auc, feature_importance, estimators) = \ train_model(model_name, train_df, cols, target) mlflow.log_param('data_params', { "traindata_shape": df.shape, "features": cols, "targetlabel": target }) # data_params is too big, must be shorter than 250 chars mlflow.log_param('model_params', { 'algorithm': 'RandomForestRegressor', # TODO! this is currently hardcoded, but shouldsn't 'n_estimators': len(estimators) }) mlflow.log_param('model_eval', { 'feature_importance': [round(x,2) for x in feature_importance], 'fpr': fpr, 'tpr': tpr, 'roc_auc': roc_auc }) mlflow.log_metric("rmse", rmse) mlflow.log_metric("mae", mae) mlflow.log_metric("r2", r2) #mlflow.log_metric("area_under_curve", acc) # TODO! add back #mlflow.log_artifact('machine_learning/models/rf_{}_roc_auc.png'.format(model_name)) # add back mlflow.log_artifact('data/ml_models/{}_feature_importance.png'.format(model_name)) mlflow.sklearn.log_model(rf, model_name) #mlflow.mleap.log_model(spark_model, sample_input, artifact_path) # # Score model for sanity checking if it works # predictions_df, model_id, best_rmse = scoring.score_best_model(model_name, test_df, cols, index_col) #print(predictions_df.tail(5)) logging.info("--- Total runtime: {} minutes ---".format(str(round(((time.time() - start_time)/60),2)))) # TODO! mlflow server into systemd service that can be restarted easily if needed # TODO! model tracking uri Dashboard to metabase # TODO!