3
\$\begingroup\$

This code aims to make very easy to train new models in SageMaker and quickly decide whether a new feature should be introduced in our model or not, getting metrics (recall, accuracy and so on) for a model with and without certain variable, or simply make quick experiments. My specific case is a fraud detection model.

Almost half of the code are docstrings (the def of the functions needs one level more of indentation and the content a level less, I know, but it has been pasted this way and I don't know a way to easily change it. In my notebook it is correct anyway):

import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.feature_extraction import FeatureHasher
import boto3
import sagemaker
from sagemaker.amazon.amazon_estimator import get_image_uri
from sagemaker import get_execution_role
from sagemaker.predictor import csv_serializer, json_deserializer
from io import StringIO    
import warnings
import logging
import gc

class HoldOutSample:
def __init__(self):
    logging.basicConfig(level=logging.DEBUG)

def load_data(self, data_file_name, bucket='sagemaker', prefix = 'tests', **kwargs):
        """This function load the data that will be used in the rest of the script

        Args:
            data_file_name (str): name of the file with the data.
            bucket (str): the bucket in which is located the data and where will be stored the model and so
            on (in following functions). It should not be changed unless a good reason, 
            for the correct order maintainability.
            prefix (str): folder in the bucket that has the data. Same as above, it is recommended to keep the
            default value.

        Returns:
            raw_data (dataframe): the loaded data as dataframe. It includes label and features

        """
        self.bucket = bucket
        self.prefix = prefix
        self.sagemaker_session = sagemaker.Session()
        self.role = get_execution_role()
        data_location = f's3://{self.bucket}/{self.prefix}/{data_file_name}'
        raw_data = pd.read_csv(data_location)
        if raw_data.shape[0]<300000: #providing the rows of the dataset are less than 300k, it shows a warning
            warnings.warn("Your dataset seems small. Maybe you should consider to use CrossValidation notebook instead.")
        logging.info('Data loaded with success')

        return raw_data

def preprocess(self, raw_data, feature_hashing_big=10, feature_hashing_small=3,
                  columns_to_transform_big=['variable_1',
                                   'variable_2'],
                  columns_to_transform_small=['a',
                                  'b',
                                  'c'],
                  features_to_drop_after_transformation=['variable_1',
                                  'b',                          
                                  'c']
                  ,**kwargs):
        """It will split the data between features(dropping unnecesary variables that are contained in the raw_data)
        and label (fraud or no fraud), make the feature hashing for the columns needed, fill the nulls
        with 0 for security reasons and format it as float32 for XGBoost compatibility.

        Args:
            raw_data(DataFrame): data to be processed.
            feature_hashing_big (int): in how many features will be splitted the variables that are going to
            be featured hashed. The big is because the variable has a big variability and hence there are
            needed several hashed features to catch all the possible values of the original variable. 
            feature_hashing_small (int): same as above but for the small variability variables (i.e variables
            that just can take 5 o 6 different values).
            columns_to_transform_big (list): variables that will be feature hashed with the big number.
            columns_to_transform_small (list): variables that will be transformed into the small number.
            features_to_drop_after_transformation (list): those original variables which are going to be
            dropped after the transformation, for avoiding duplicities. 

        Returns:
            processed_data (DataFrame): data with all the transformations needed to create the XGBoost model

        """
        features = raw_data.drop(['order_id','domain','product_id','date'], axis=1)
        label = features.pop('fraud')
        hasher_big = FeatureHasher(n_features=feature_hashing_big, input_type="string")
        hasher_small = FeatureHasher(n_features=feature_hashing_small, input_type="string")    
        d = {}
        columns_to_transform_10 = columns_to_transform_big
        columns_to_transform_3 = columns_to_transform_small

        for column in columns_to_transform_big:
            d[column] = pd.DataFrame(hasher_big.transform(features[column].astype(str)).toarray())
            d[column].columns = [column + f'_{i}' for i in range(feature_hashing_big)]

        for column in columns_to_transform_small:
            d[column] = pd.DataFrame(hasher_small.transform(features[column].astype(str)).toarray())
            d[column].columns = [column + f'_{i}' for i in range(feature_hashing_small)]

        transformed_features = pd.concat(d,axis=1)
        transformed_features.columns = transformed_features.columns.droplevel(0)

        features = features.drop(features_to_drop_after_transformation, axis=1)
        concatenated_data = pd.concat([label,features,transformed_features], axis=1)
        #The format float32 is necessary for XGBoost models
        processed_data = concatenated_data.fillna(0).astype('float32')
        logging.info('Data preprocessed correctly')

        return processed_data


def split_data(self, data, test_size=0.2, validation_size = 0.25, **kwargs):
    """First, the data will be splitted in two datasets: train/validation and test. After, the first one will
    be splitted in train and validation.

    Args:
        data(DataFrame): self-explanatory
        test_size (float): percentage of data that is kept for the second dataset that is returned.

    """
    self.train_and_validation_data, self.test_data = train_test_split(data, test_size = test_size)
    self.train_data, self.validation_data = train_test_split(self.train_and_validation_data, 
                                                             test_size = validation_size)

def save_data_in_S3_for_XGB(self, train_route='train.csv', validation_route='validation.csv', **kwargs):
    """XGBoost takes the data from S3 in a special way, and this function will save the train and validation
    in that way in S3.

    Args:
        train_route (str): name of the train dataset csv file that will be saved in S3.
        validation_route (str): same as above but for validation dataset.
    """
    csv_buffer = StringIO()
    self.train_data.to_csv(csv_buffer, header = False, index = False)
    s3_resource = boto3.resource('s3')
    s3_resource.Object(self.bucket, f'{self.prefix}/{train_route}').put(Body=csv_buffer.getvalue())

    csv_buffer = StringIO()
    self.validation_data.to_csv(csv_buffer, header = False, index = False)
    s3_resource.Object(self.bucket, f'{self.prefix}/{validation_route}').put(Body=csv_buffer.getvalue())

    self.xgb_s3_input_validation = sagemaker.s3_input(s3_data=f's3://{self.bucket}/{self.prefix}/validation.csv', 
                                                      content_type = 'csv')
    self.container = get_image_uri(boto3.Session().region_name, 'xgboost')
    self.xgb_s3_input_train = sagemaker.s3_input(s3_data=f's3://{self.bucket}/{self.prefix}/train.csv', 
                                                 content_type = 'csv')

def train(self, objective='binary:logistic', num_round=400, eval_metric='error@0.1', scale_pos_weight=10,
         train_instance_count=1, train_instance_type='ml.m4.2xlarge', output_path='outputs', **kwargs):
    """This function makes the three steeps necessaries for training the model: creating the estimator,
    set hyperparameters and fit the model.

    Args:
        objective (str): the default value is the one needed for binary classification. Don't change.
        num_round (int): number of rounds (iterations) for the training of the model. It has been proved
        that somewhere between 300 and 450 rounds (depending on the size of the dataset) the algorithm does 
        not get bette, but until that cipher is reached, there are sligthly improvements. 
        eval_metric (str): in each round, the model is trained using the train dataset and evaluated against
        the validation dataset. This is the metric that takes place in the process. 
        scale_pos_weight (int): this is useful for umbalanced datasets (as our) and gives the less frequent label
        an extra importance. It has been proven that using the recommended calculation for this gives bad results.
        train_instance_count (str): Number of instances that will execute the train job.
        train_instance_type (str): the type of the instance that will execute the train job. The more powerful
        it is, the more it costs and the less it takes to train the model. The most powerful ones (x24large) 
        are not totally worth it because the time is not exactly the half of the previous ones (x12large) so we
        get charged for more money. Anyway that difference is not that much and if we need it to be trained fast
        is a total feasible option.
        output_path (str): folder inside the bucket and prefix in where the model artifact will be saved.

    """
    self.xgb = sagemaker.estimator.Estimator(self.container,
                                            self.role,
                                            train_instance_count = train_instance_count,
                                            train_instance_type = train_instance_type,
                                            output_path = f's3://{self.bucket}/{self.prefix}/{output_path}',
                                            sagemaker_session=sagemaker.Session())

    self.xgb.set_hyperparameters(objective = objective,
                    num_round = num_round,
                    eval_metric = eval_metric,
                    scale_pos_weight = scale_pos_weight
                   )

    self.xgb.fit({'train': self.xgb_s3_input_train,
                  'validation':self.xgb_s3_input_validation})

def batch_predict(self, data, rows=500, **kwargs):
    """Batch prediction for the data using the recent created inference endpoint. This function is used 
    in the next one for actually getting the predictions.

    Args:
        rows (int): number of rows per batch. Probably it is pointless to change the default value.
        data

    Returns:
        predictions in a numpy data structure separated by a comma

    """
    split_array = np.array_split(data, int(data.shape[0] / float(rows) + 1))
    predictions = ''
    for array in split_array:
        predictions = ','.join([predictions, self.xgb_predictor.predict(array).decode('utf-8')])
    logging.info('Predictions made')

    return np.fromstring(predictions[1:], sep=',')

def get_predictions(self, delete_endpoint='Yes', instance_type='ml.t2.medium', **kwargs):
    """This deploys an endpoint and makes the predictions on the test data. 

    Args:
        delete_endpoint (str): whether to delete the endpoint after getting the predictions or not.
        instance_type (str): instance type for the inference endpoint. 

    """

    self.xgb_predictor = self.xgb.deploy(initial_instance_count=1, instance_type=instance_type)
    logging.info('Endpoint deployed')
    self.xgb_predictor.content_type = 'text/csv' #This three lines are necessary for the endpoint
    self.xgb_predictor.serializer = csv_serializer #to read the data
    self.xgb_predictor.deserializer = None
    predictions = self.batch_predict(self.test_data_features.as_matrix()) #it gets the scores for each datum
    if delete_endpoint == 'Yes':
        sagemaker.Session().delete_endpoint(self.xgb_predictor.endpoint)
        logging.info('Endpoint deleted')
    self.predictions = pd.DataFrame(predictions,columns=['score'])

def calculate_metrics(self, score = 0.1, **kwargs):
    """It calculates the following metrics for evaluating the performance of the model:
    recall,precision,accuracy,f1_score,FP_rate,perdidas_eviatadas,fraudes y revisados.
    For that, the function needs to calculate first the true negatives, false positives, 
    true positives and false negatives. Those variables are not accessible though. The calculation
    is saved in the dictionary previously created with the function 'create_metrics_variables'

    Args:
        score (float): the treshold to calculate the metrics.
    """
    #The indexes have to be reseted. Otherwise the concats (1 and 2) will be wrong.
    test_data_features_reset_index = self.test_data_features.reset_index(inplace=False,drop=True)
    test_data_label_reset_index = self.test_data_label.reset_index(inplace=False,drop=True)

    self.score_and_test_label = pd.concat([self.predictions, #concat 1
                                           test_data_label_reset_index], 
                                          axis = 1)

    self.si_fraude = self.score_and_test_label[self.score_and_test_label['fraud']==1]
    self.no_fraude = self.score_and_test_label[self.score_and_test_label['fraud']==0]

    tn = self.no_fraude[self.no_fraude['score']<score]['fraud'].count()
    fp = self.no_fraude[self.no_fraude['score']>=score]['fraud'].count()
    tp = self.si_fraude[self.si_fraude['score']>=score]['fraud'].count()
    fn = self.si_fraude[self.si_fraude['score']<score]['fraud'].count()

    self.metrics_dict['Recall'].append(tp / (tp + fn))
    self.metrics_dict['Precision'].append(tp / (tp + fp))
    self.metrics_dict['Accuracy'].append((tp + tn) / (tp + fp + tn + fn))
    #the F1 score is calculated this way because the way recall and precision are stored in lists makes it 
    #pointless to implement a solution with those 2 metrics (since they are stored in a list, so you will have 
    #to add a counter variable to know if you have to access to the value [0] or [1] of the list)
    self.metrics_dict['F1_score'].append(2 * (((tp/(tp+fp))*(tp/(tp+fn)))/((tp/(tp+fp))+(tp/(tp+fn)))))
    self.metrics_dict['FP_rate'].append(fp/(fp+tn))

    self.pred = pd.concat([self.predictions, #concat 2
                           test_data_features_reset_index,
                           test_data_label_reset_index],
                          axis=1) 
    #the termination .item() in the next line is to convert np.float32 to native python float
    self.metrics_dict['Perdidas_evitadas'].append(self.pred[(self.pred['score']>=score) & (self.pred['fraud']==1)]['mt_perdida_potencial_nr'].sum().item())
    self.metrics_dict['Fraudes'].append(self.pred[(self.pred['score']>=score) & (self.pred['fraud']==1)]['mt_perdida_potencial_nr'].count())
    self.metrics_dict['Revisados'].append(self.pred[(self.pred['score']>=score)]['score'].count())

    logging.info('Metrics got successfully')

def create_metrics_variables(self, **kwargs):
    """It creates a dictionary that contains a list for each metric that is going to be calculated."""

    self.metrics_dict = {}
    self.metrics_list = ['Recall','Precision','Accuracy','F1_score','FP_rate', 'Perdidas_evitadas',
              'Fraudes','Revisados']

    for metric in self.metrics_list:
        self.metrics_dict[metric] = []

def process(self, **kwargs):
    """The whole process that is make right after preprocessing the data until the metrics are obtained.

    """
    gc.collect() #it frees RAM
    self.save_data_in_S3_for_XGB(**kwargs)
    self.train(**kwargs)
    self.test_data_label = self.test_data.loc[:,'fraud'] 
    self.test_data_label.columns = ['fraud']
    self.test_data_features = self.test_data.drop(columns =['fraud'], axis=1)
    self.get_predictions(**kwargs)
    self.calculate_metrics(**kwargs)

def feature_hashing_changed(self, **kwargs):
    """Function for complying with DRY (Don't repeat yourself) principle in the below function"""
    self.changed_preprocess_train_data = self.preprocess(self.raw_train_data, **kwargs)
    self.changed_preprocess_validation_data = self.preprocess(self.raw_validation_data, **kwargs)
    self.changed_preprocess_test_data = self.preprocess(self.raw_test_data, **kwargs)

def test_feature_hashing(self, data_file_name, variable_to_change, new_variable_value, **kwargs):
    """It runs the entire process to test different configurations in the feature hashing. First,
    it has to split the raw data and save it, so that you can access whenever you want to the data
    before being processed. The preprocessing is made in the three datasets (train,validation and test)
    to assure consistency across both models. Then, it get the metrics for the a model with the default 
    preprocessing. After that, a second model with a different value for one chosen variable for the
    feature hashing is trained and the new metrics are obtained and saved for comparison."""
    raw_data = self.load_data(data_file_name)
    self.create_metrics_variables()
    self.split_data(raw_data)
    self.raw_train_data = self.train_data
    self.raw_validation_data = self.validation_data
    self.raw_test_data = self.test_data

    original_preprocess_train_data = self.preprocess(self.raw_train_data)
    original_preprocess_validation_data = self.preprocess(self.raw_validation_data)
    original_preprocess_test_data = self.preprocess(self.raw_test_data)
    #the next 3 following variables are used in the training and prediction of the model
    self.train_data = original_preprocess_train_data
    self.validation_data = original_preprocess_validation_data
    self.test_data = original_preprocess_test_data
    self.process(**kwargs)

    if variable_to_change == 'feature_hashing_big':
        self.feature_hashing_changed(feature_hashing_big = new_variable_value)
    elif variable_to_change == 'feature_hashing_small':
        self.feature_hashing_changed(feature_hashing_small = new_variable_value)
    elif variable_to_change == 'columns_to_transform_big':
        self.feature_hashing_changed(columns_to_transform_big = new_variable_value)
    elif variable_to_change == 'columns_to_transform_small':
        self.feature_hashing_changed(columns_to_transform_small = new_variable_value)
    elif variable_to_change == 'features_to_drop_after_transformation':
        self.feature_hashing_changed(features_to_drop_after_transformation = new_variable_value)

    self.train_data = self.changed_preprocess_train_data
    self.validation_data = self.changed_preprocess_validation_data
    self.test_data = self.changed_preprocess_test_data
    self.process(**kwargs)

    self.results = pd.DataFrame.from_dict(self.metrics_dict, 
                                          orient = 'index', 
                                          columns = ['1st test','2nd test'])

def test_new_variable(self, data_file_name, variable_to_delete, **kwargs):
    """Gathering function that execute the entire process needed to decide whether to preserve a variable or not.

    Args:
        data_file_name (str): pretty self explanatory
        variable_to_delete (str): same as above

    """
    raw_data = self.load_data(data_file_name)
    original_data_preprocessed = self.preprocess(raw_data)
    self.split_data(original_data_preprocessed)
    self.create_metrics_variables(**kwargs)

    self.process(**kwargs)

    #To maintain the same data for both models (and hence avoid changes due to randomness)
    #we need to make the modifications in the same datasets
    self.train_data = self.train_data.drop(columns=[variable_to_delete], axis=1)
    self.validation_data = self.validation_data.drop(columns=[variable_to_delete], axis=1)
    self.test_data = self.test_data.drop(columns=[variable_to_delete], axis=1)

    self.process(**kwargs)

    self.results = pd.DataFrame.from_dict(self.metrics_dict, 
                                          orient = 'index', 
                                          columns = ['1st test','2nd test'])

def single_test(self, data_file_name, **kwargs):
    """For the cases when you just want to run a single test(prototyping or whatever).

    """
    raw_data = self.load_data(data_file_name)
    processed_data = self.preprocess(raw_data)
    self.split_data(processed_data)
    self.create_metrics_variables()
    self.process(**kwargs)
    self.results = pd.DataFrame.from_dict(self.metrics_dict, 
                                          orient = 'index', 
                                          columns = ['1st test'])

So let's say I want to test whether the variable 'pax' would improve my model; I just need to code this:

a = HoldOutSample()
a.test_new_variable(data_file_name = 'data1.csv', variable_to_delete = 'pax')
a.results

And I will get this:

enter image description here

I have omitted the maximum length of 79 established by PEP8 because I am going to use this code just in Jupyter Notebook and I have sticked to the width of notebook. However, there are a few lines that are very large and I don't know how to split them without breaking the code (or if that is important at all).

This is the first time I use OOP, so I guess there's plenty of room for improvements. Also, I was wondering if the way I store and present the data is the better. Even I have doubts concerning if this is an appropriate case to use OOP instead of functional programming. Any help is appreciated.

Thank you very much

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.