<s> """<STR_LIT>""" <EOL> from __future__ import unicode_literals <EOL> from __future__ import division <EOL> from __future__ import absolute_import <EOL> from __future__ import print_function <EOL> import numpy as np <EOL> import pickle <EOL> class MLTools ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . last_training_pattern = [ ] <EOL> self . has_trained = False <EOL> self . cv_best_rmse = "<STR_LIT>" <EOL> def _local_train ( self , training_patterns , training_expected_targets , <EOL> params ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> def _local_test ( self , testing_patterns , testing_expected_targets , <EOL> predicting ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> def _ml_search_param ( self , database , dataprocess , path_filename , save , <EOL> cv , min_f ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> def _ml_print_parameters ( self ) : <EOL> """<STR_LIT>""" <EOL> return None <EOL> def _ml_predict ( self , horizon = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> if not self . has_trained : <EOL> print ( "<STR_LIT>" ) <EOL> return <EOL> new_pattern = np . hstack ( [ self . last_training_pattern [ <NUM_LIT:2> : ] , <EOL> self . last_training_pattern [ <NUM_LIT:0> ] ] ) <EOL> new_pattern = np . insert ( new_pattern , <NUM_LIT:0> , <NUM_LIT:1> ) . reshape ( <NUM_LIT:1> , - <NUM_LIT:1> ) <EOL> predicted_targets = np . zeros ( ( horizon , <NUM_LIT:1> ) ) <EOL> for t_counter in range ( horizon ) : <EOL> te_errors = self . test ( new_pattern , predicting = True ) <EOL> predicted_value = te_errors . predicted_targets <EOL> predicted_targets [ t_counter ] = predicted_value <EOL> new_pattern = np . hstack ( [ new_pattern [ <NUM_LIT:0> , <NUM_LIT:2> : ] , <EOL> np . squeeze ( predicted_value ) ] ) <EOL> new_pattern = np . insert ( new_pattern , <NUM_LIT:0> , <NUM_LIT:1> ) . reshape ( <NUM_LIT:1> , - <NUM_LIT:1> ) <EOL> return predicted_targets <EOL> def _ml_train ( self , training_matrix , params ) : <EOL> """<STR_LIT>""" <EOL> training_patterns = training_matrix [ : , <NUM_LIT:1> : ] <EOL> training_expected_targets = training_matrix [ : , <NUM_LIT:0> ] <EOL> training_predicted_targets = self . _local_train ( training_patterns , <EOL> training_expected_targets , <EOL> params ) <EOL> training_errors = Error ( training_expected_targets , <EOL> training_predicted_targets , <EOL> regressor_name = self . regressor_name ) <EOL> self . last_training_pattern = training_matrix [ - <NUM_LIT:1> , : ] <EOL> self . has_trained = True <EOL> return training_errors <EOL> def _ml_test ( self , testing_matrix , predicting = False ) : <EOL> """<STR_LIT>""" <EOL> testing_patterns = testing_matrix [ : , <NUM_LIT:1> : ] <EOL> testing_expected_targets = testing_matrix [ : , <NUM_LIT:0> ] . reshape ( - <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> testing_predicted_targets = self . _local_test ( testing_patterns , <EOL> testing_expected_targets , <EOL> predicting ) <EOL> testing_errors = Error ( testing_expected_targets , <EOL> testing_predicted_targets , <EOL> regressor_name = self . regressor_name ) <EOL> return testing_errors <EOL> def _ml_train_iterative ( self , database_matrix , params = [ ] , <EOL> sliding_window = <NUM_LIT> , k = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> p = database_matrix . shape [ <NUM_LIT:1> ] - <NUM_LIT:1> <EOL> number_iterations = database_matrix . shape [ <NUM_LIT:0> ] + p - k - sliding_window + <NUM_LIT:1> <EOL> print ( "<STR_LIT>" , number_iterations ) <EOL> tr_size = sliding_window - p - <NUM_LIT:1> <EOL> z = database_matrix [ <NUM_LIT:0> : - k , <NUM_LIT:1> ] . reshape ( - <NUM_LIT:1> , <NUM_LIT:1> ) * np . ones ( ( <NUM_LIT:1> , p ) ) <EOL> database_matrix [ k : , <NUM_LIT:1> : ] = database_matrix [ k : , <NUM_LIT:1> : ] - z <EOL> pr_target = [ ] <EOL> ex_target = [ ] <EOL> for i in range ( number_iterations ) : <EOL> self . _ml_train ( database_matrix [ k + i : k + i + tr_size - <NUM_LIT:1> , : ] , params ) <EOL> pr_t = self . _ml_predict ( horizon = <NUM_LIT:1> ) <EOL> pr_t = pr_t [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] + z [ i , <NUM_LIT:0> ] <EOL> pr_target . append ( pr_t ) <EOL> ex_target . append ( database_matrix [ k + i + tr_size , <NUM_LIT:0> ] ) <EOL> pr_result = Error ( expected = ex_target , predicted = pr_target ) <EOL> return pr_result <EOL> def save_regressor ( self , file_name ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> file = file_name <EOL> with open ( file , '<STR_LIT:wb>' ) as f : <EOL> pickle . dump ( self , f , protocol = pickle . HIGHEST_PROTOCOL ) <EOL> except : <EOL> print ( "<STR_LIT>" , file_name ) <EOL> return <EOL> else : <EOL> print ( "<STR_LIT>" , file_name ) <EOL> def load_regressor ( self , file_name ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> file = file_name <EOL> with open ( file , '<STR_LIT:rb>' ) as f : <EOL> self = pickle . load ( f ) <EOL> except : <EOL> print ( "<STR_LIT>" , file_name ) <EOL> return <EOL> return self <EOL> class Error ( object ) : <EOL> """<STR_LIT>""" <EOL> available_error_metrics = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> def __init__ ( self , expected , predicted , regressor_name = "<STR_LIT>" ) : <EOL> if type ( expected ) is list : <EOL> expected = np . array ( expected ) <EOL> if type ( predicted ) is list : <EOL> predicted = np . array ( predicted ) <EOL> expected = expected . flatten ( ) <EOL> predicted = predicted . flatten ( ) <EOL> self . regressor_name = regressor_name <EOL> self . expected_targets = expected <EOL> self . predicted_targets = predicted <EOL> self . dict_errors = { } <EOL> for error in self . available_error_metrics : <EOL> self . dict_errors [ error ] = "<STR_LIT>" <EOL> def _calc ( self , name , expected , predicted ) : <EOL> """<STR_LIT>""" <EOL> if self . dict_errors [ name ] == "<STR_LIT>" : <EOL> if name == "<STR_LIT>" : <EOL> error = expected - predicted <EOL> self . dict_errors [ name ] = np . mean ( np . fabs ( error ) ) <EOL> elif name == "<STR_LIT>" : <EOL> error = expected - predicted <EOL> self . dict_errors [ name ] = error . mean ( ) <EOL> elif name == "<STR_LIT>" : <EOL> error = expected - predicted <EOL> self . dict_errors [ name ] = ( error ** <NUM_LIT:2> ) . mean ( ) <EOL> elif name == "<STR_LIT>" : <EOL> error = expected - predicted <EOL> self . dict_errors [ name ] = np . sqrt ( ( error ** <NUM_LIT:2> ) . mean ( ) ) <EOL> elif name == "<STR_LIT>" : <EOL> if np . count_nonzero ( expected != <NUM_LIT:0> ) == <NUM_LIT:0> : <EOL> self . dict_errors [ name ] = np . nan <EOL> else : <EOL> find_zero = expected != <NUM_LIT:0> <EOL> _et = np . extract ( find_zero , expected ) <EOL> _pt = np . extract ( find_zero , predicted ) <EOL> relative_error = ( _et - _pt ) / _et <EOL> self . dict_errors [ name ] = <NUM_LIT:100> * relative_error . mean ( ) <EOL> elif name == "<STR_LIT>" : <EOL> if np . count_nonzero ( expected != <NUM_LIT:0> ) == <NUM_LIT:0> : <EOL> self . dict_errors [ name ] = np . nan <EOL> else : <EOL> find_zero = expected != <NUM_LIT:0> <EOL> _et = np . extract ( find_zero , expected ) <EOL> _pt = np . extract ( find_zero , predicted ) <EOL> relative_error = ( _et - _pt ) / _et <EOL> self . dict_errors [ name ] = <NUM_LIT:100> * np . fabs ( relative_error ) . mean ( ) <EOL> elif name == "<STR_LIT>" : <EOL> error = expected - predicted <EOL> self . dict_errors [ name ] = np . std ( error ) <EOL> elif name == "<STR_LIT>" : <EOL> _c = expected * predicted <EOL> if np . count_nonzero ( _c != <NUM_LIT:0> ) == <NUM_LIT:0> : <EOL> self . dict_errors [ name ] = np . nan <EOL> else : <EOL> self . dict_errors [ name ] = np . count_nonzero ( _c > <NUM_LIT:0> ) / np . count_nonzero ( _c != <NUM_LIT:0> ) <EOL> elif name == "<STR_LIT>" : <EOL> _a = expected <EOL> _b = predicted <EOL> if np . count_nonzero ( _b > <NUM_LIT:0> ) == <NUM_LIT:0> : <EOL> self . dict_errors [ name ] = np . nan <EOL> else : <EOL> self . dict_errors [ name ] = np . count_nonzero ( ( _a > <NUM_LIT:0> ) * ( _b > <NUM_LIT:0> ) ) / np . count_nonzero ( _b > <NUM_LIT:0> ) <EOL> elif name == "<STR_LIT>" : <EOL> _a = expected <EOL> _b = predicted <EOL> if np . count_nonzero ( _b < <NUM_LIT:0> ) == <NUM_LIT:0> : <EOL> self . dict_errors [ name ] = np . nan <EOL> else : <EOL> self . dict_errors [ name ] = np . count_nonzero ( ( _a < <NUM_LIT:0> ) * ( _b < <NUM_LIT:0> ) ) / np . count_nonzero ( _b < <NUM_LIT:0> ) <EOL> elif name == "<STR_LIT>" : <EOL> _a = expected . astype ( int ) <EOL> _b = np . round ( predicted ) . astype ( int ) <EOL> self . dict_errors [ name ] = np . count_nonzero ( _a == _b ) / _b . size <EOL> else : <EOL> print ( "<STR_LIT>" , name , <EOL> "<STR_LIT>" ) <EOL> return <EOL> def calc_metrics ( self ) : <EOL> """<STR_LIT>""" <EOL> for error in sorted ( self . dict_errors . keys ( ) ) : <EOL> self . _calc ( error , self . expected_targets , self . predicted_targets ) <EOL> def print_errors ( self ) : <EOL> """<STR_LIT>""" <EOL> self . calc_metrics ( ) <EOL> try : <EOL> from prettytable import PrettyTable <EOL> table = PrettyTable ( [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> table . align [ "<STR_LIT>" ] = "<STR_LIT:l>" <EOL> table . align [ "<STR_LIT>" ] = "<STR_LIT:l>" <EOL> for error in sorted ( self . dict_errors . keys ( ) ) : <EOL> table . add_row ( [ error , np . around ( self . dict_errors [ error ] , decimals = <NUM_LIT:8> ) ] ) <EOL> print ( ) <EOL> print ( table . get_string ( sortby = "<STR_LIT>" ) ) <EOL> print ( ) <EOL> except ImportError : <EOL> print ( "<STR_LIT>" ) <EOL> print ( ) <EOL> for error in sorted ( self . dict_errors . keys ( ) ) : <EOL> print ( error , np . around ( self . dict_errors [ error ] , decimals = <NUM_LIT:8> ) ) <EOL> print ( ) <EOL> def print_values ( self ) : <EOL> """<STR_LIT>""" <EOL> print ( "<STR_LIT>" , self . expected_targets . reshape ( <NUM_LIT:1> , - <NUM_LIT:1> ) , "<STR_LIT:\n>" , <EOL> "<STR_LIT>" , self . predicted_targets . reshape ( <NUM_LIT:1> , - <NUM_LIT:1> ) , "<STR_LIT:\n>" ) <EOL> def get ( self , error ) : <EOL> """<STR_LIT>""" <EOL> self . _calc ( error , self . expected_targets , self . predicted_targets ) <EOL> return self . dict_errors [ error ] <EOL> def get_std ( self ) : <EOL> self . _calc ( "<STR_LIT>" , self . expected_targets , self . predicted_targets ) <EOL> return self . dict_errors [ "<STR_LIT>" ] <EOL> def get_mae ( self ) : <EOL> self . _calc ( "<STR_LIT>" , self . expected_targets , self . predicted_targets ) <EOL> return self . dict_errors [ "<STR_LIT>" ] <EOL> def get_mse ( self ) : <EOL> self . _calc ( "<STR_LIT>" , self . expected_targets , self . predicted_targets ) <EOL> return self . dict_errors [ "<STR_LIT>" ] <EOL> def get_rmse ( self ) : <EOL> self . _calc ( "<STR_LIT>" , self . expected_targets , self . predicted_targets ) <EOL> return self . dict_errors [ "<STR_LIT>" ] <EOL> def get_mpe ( self ) : <EOL> self . _calc ( "<STR_LIT>" , self . expected_targets , self . predicted_targets ) <EOL> return self . dict_errors [ "<STR_LIT>" ] <EOL> def get_mape ( self ) : <EOL> self . _calc ( "<STR_LIT>" , self . expected_targets , self . predicted_targets ) <EOL> return self . dict_errors [ "<STR_LIT>" ] <EOL> def get_me ( self ) : <EOL> self . _calc ( "<STR_LIT>" , self . expected_targets , self . predicted_targets ) <EOL> return self . dict_errors [ "<STR_LIT>" ] <EOL> def get_hr ( self ) : <EOL> self . _calc ( "<STR_LIT>" , self . expected_targets , self . predicted_targets ) <EOL> return self . dict_errors [ "<STR_LIT>" ] <EOL> def get_hrm ( self ) : <EOL> self . _calc ( "<STR_LIT>" , self . expected_targets , self . predicted_targets ) <EOL> return self . dict_errors [ "<STR_LIT>" ] <EOL> def get_hrp ( self ) : <EOL> self . _calc ( "<STR_LIT>" , self . expected_targets , self . predicted_targets ) <EOL> return self . dict_errors [ "<STR_LIT>" ] <EOL> def get_accuracy ( self ) : <EOL> self . _calc ( "<STR_LIT>" , self . expected_targets , self . predicted_targets ) <EOL> return self . dict_errors [ "<STR_LIT>" ] <EOL> def get_error ( self ) : <EOL> return ( self . expected_targets - self . predicted_targets ) . flatten ( ) <EOL> def get_anderson ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> from scipy import stats <EOL> except ImportError : <EOL> raise ImportError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> error = ( self . expected_targets - self . predicted_targets ) . flatten ( ) <EOL> ad_statistic , ad_c , ad_s = stats . anderson ( error , "<STR_LIT>" ) <EOL> return ad_statistic , ad_c , ad_s <EOL> def get_shapiro ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> from scipy import stats <EOL> except ImportError : <EOL> raise ImportError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> error = ( self . expected_targets - self . predicted_targets ) . flatten ( ) <EOL> sw_statistic , sw_p_value = stats . shapiro ( error ) <EOL> return sw_statistic , sw_p_value <EOL> class CVError ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , fold_errors ) : <EOL> self . fold_errors = fold_errors <EOL> self . all_fold_errors = { } <EOL> self . all_fold_mean_errors = { } <EOL> for error in self . fold_errors [ <NUM_LIT:0> ] . available_error_metrics : <EOL> self . all_fold_errors [ error ] = [ ] <EOL> self . all_fold_mean_errors [ error ] = - <NUM_LIT> <EOL> self . calc_metrics ( ) <EOL> def calc_metrics ( self ) : <EOL> """<STR_LIT>""" <EOL> for fold in self . fold_errors : <EOL> for error in fold . dict_errors : <EOL> if fold . dict_errors [ error ] == "<STR_LIT>" : <EOL> fold . dict_errors [ error ] = fold . get ( error ) <EOL> self . all_fold_errors [ error ] . append ( fold . dict_errors [ error ] ) <EOL> for error in sorted ( self . all_fold_errors . keys ( ) ) : <EOL> self . all_fold_mean_errors [ error ] = np . mean ( self . all_fold_errors [ error ] ) <EOL> def print_errors ( self ) : <EOL> """<STR_LIT>""" <EOL> for error in sorted ( self . all_fold_errors . keys ( ) ) : <EOL> print ( error , "<STR_LIT>" , self . all_fold_mean_errors [ error ] ) <EOL> print ( self . all_fold_errors [ error ] , "<STR_LIT:\n>" ) <EOL> print ( ) <EOL> def get ( self , error ) : <EOL> return self . all_fold_mean_errors [ error ] <EOL> def get_rmse ( self ) : <EOL> return self . all_fold_mean_errors [ "<STR_LIT>" ] <EOL> def get_accuracy ( self ) : <EOL> return self . all_fold_mean_errors [ "<STR_LIT>" ] <EOL> def read ( file_name ) : <EOL> """<STR_LIT>""" <EOL> data = np . loadtxt ( file_name ) <EOL> return data <EOL> def write ( file_name , data ) : <EOL> """<STR_LIT>""" <EOL> np . savetxt ( file_name , data ) <EOL> def split_sets ( data , training_percent = None , n_test_samples = None , perm = False ) : <EOL> """<STR_LIT>""" <EOL> number_of_samples = data . shape [ <NUM_LIT:0> ] <EOL> if perm : <EOL> np . random . shuffle ( data ) <EOL> if n_test_samples is not None : <EOL> training_samples = number_of_samples - n_test_samples <EOL> elif training_percent is not None : <EOL> training_samples = round ( number_of_samples * training_percent ) <EOL> else : <EOL> raise Exception ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> training_matrix = data [ <NUM_LIT:0> : training_samples , : ] <EOL> testing_matrix = data [ training_samples : , : ] <EOL> return training_matrix , testing_matrix <EOL> def time_series_cross_validation ( ml , database , params , number_folds = <NUM_LIT:10> , <EOL> dataprocess = None ) : <EOL> """<STR_LIT>""" <EOL> if number_folds < <NUM_LIT:2> : <EOL> print ( "<STR_LIT>" ) <EOL> return <EOL> number_patterns = database . shape [ <NUM_LIT:0> ] <EOL> fold_size = round ( number_patterns / number_folds ) <EOL> folds = [ ] <EOL> for k in range ( number_folds ) : <EOL> folds . append ( database [ k * fold_size : ( k + <NUM_LIT:1> ) * fold_size , : ] ) <EOL> training_errors = [ ] <EOL> testing_errors = [ ] <EOL> training_matrix = folds [ <NUM_LIT:0> ] <EOL> testing_matrix = [ ] <EOL> for k in range ( number_folds - <NUM_LIT:1> ) : <EOL> if k > <NUM_LIT:0> : <EOL> training_matrix = np . concatenate ( ( training_matrix , testing_matrix ) , axis = <NUM_LIT:0> ) <EOL> testing_matrix = folds [ k + <NUM_LIT:1> ] <EOL> if dataprocess is not None : <EOL> training_matrix , testing_matrix = dataprocess . auto ( training_matrix , testing_matrix ) <EOL> tr_error = ml . train ( training_matrix , params ) <EOL> te_error = ml . test ( testing_matrix ) <EOL> training_errors . append ( tr_error ) <EOL> testing_errors . append ( te_error ) <EOL> cv_training_error = CVError ( training_errors ) <EOL> cv_testing_error = CVError ( testing_errors ) <EOL> return cv_training_error , cv_testing_error <EOL> def kfold_cross_validation ( ml , database , params , number_folds = <NUM_LIT:10> , <EOL> dataprocess = None ) : <EOL> """<STR_LIT>""" <EOL> if number_folds < <NUM_LIT:2> : <EOL> print ( "<STR_LIT>" ) <EOL> return <EOL> np . random . shuffle ( database ) <EOL> n_dim = database . shape [ <NUM_LIT:1> ] - <NUM_LIT:1> <EOL> number_patterns = database . shape [ <NUM_LIT:0> ] <EOL> fold_size = np . ceil ( number_patterns / number_folds ) <EOL> folds = [ ] <EOL> for k in range ( number_folds ) : <EOL> folds . append ( database [ k * fold_size : ( k + <NUM_LIT:1> ) * fold_size , : ] ) <EOL> training_errors = [ ] <EOL> testing_errors = [ ] <EOL> for k in range ( number_folds ) : <EOL> training_matrix = np . array ( folds [ : k ] + folds [ k + <NUM_LIT:1> : - <NUM_LIT:1> ] ) . reshape ( - <NUM_LIT:1> , n_dim + <NUM_LIT:1> ) <EOL> if k < number_folds - <NUM_LIT:1> : <EOL> training_matrix = np . vstack ( ( training_matrix , folds [ - <NUM_LIT:1> ] ) ) <EOL> testing_matrix = folds [ k ] <EOL> if dataprocess is not None : <EOL> training_matrix , testing_matrix = dataprocess . auto ( training_matrix , testing_matrix ) <EOL> training_errors . append ( ml . train ( training_matrix , params ) ) <EOL> testing_errors . append ( ml . test ( testing_matrix ) ) <EOL> cv_training_error = CVError ( training_errors ) <EOL> cv_testing_error = CVError ( testing_errors ) <EOL> return cv_training_error , cv_testing_error <EOL> def copy_doc_of ( fun ) : <EOL> def decorator ( f ) : <EOL> f . __doc__ = fun . __doc__ <EOL> return f <EOL> return decorator </s>
