<s> """<STR_LIT>""" <EOL> from contracts import contract <EOL> from neat . contracts_primitive import * <EOL> from neat . contracts_extra import * <EOL> from numpy import median <EOL> from scipy . optimize import leastsq <EOL> import numpy as np <EOL> import logging <EOL> log = logging . getLogger ( __name__ ) <EOL> @ contract <EOL> def loess_factory ( time_step , migration_time , params ) : <EOL> """<STR_LIT>""" <EOL> migration_time_normalized = float ( migration_time ) / time_step <EOL> return lambda utilization , state = None : ( loess ( params [ '<STR_LIT>' ] , <EOL> params [ '<STR_LIT>' ] , <EOL> params [ '<STR_LIT>' ] , <EOL> migration_time_normalized , <EOL> utilization ) , <EOL> { } ) <EOL> @ contract <EOL> def loess_robust_factory ( time_step , migration_time , params ) : <EOL> """<STR_LIT>""" <EOL> migration_time_normalized = float ( migration_time ) / time_step <EOL> return lambda utilization , state = None : ( loess_robust ( params [ '<STR_LIT>' ] , <EOL> params [ '<STR_LIT>' ] , <EOL> params [ '<STR_LIT>' ] , <EOL> migration_time_normalized , <EOL> utilization ) , <EOL> { } ) <EOL> @ contract <EOL> def mad_threshold_factory ( time_step , migration_time , params ) : <EOL> """<STR_LIT>""" <EOL> return lambda utilization , state = None : ( mad_threshold ( params [ '<STR_LIT>' ] , <EOL> params [ '<STR_LIT>' ] , <EOL> utilization ) , <EOL> { } ) <EOL> @ contract <EOL> def iqr_threshold_factory ( time_step , migration_time , params ) : <EOL> """<STR_LIT>""" <EOL> return lambda utilization , state = None : ( iqr_threshold ( params [ '<STR_LIT>' ] , <EOL> params [ '<STR_LIT>' ] , <EOL> utilization ) , <EOL> { } ) <EOL> @ contract <EOL> def loess ( threshold , param , length , migration_time , utilization ) : <EOL> """<STR_LIT>""" <EOL> return loess_abstract ( loess_parameter_estimates , <EOL> threshold , <EOL> param , <EOL> length , <EOL> migration_time , <EOL> utilization ) <EOL> @ contract <EOL> def loess_robust ( threshold , param , length , migration_time , utilization ) : <EOL> """<STR_LIT>""" <EOL> return loess_abstract ( loess_robust_parameter_estimates , <EOL> threshold , <EOL> param , <EOL> length , <EOL> migration_time , <EOL> utilization ) <EOL> @ contract <EOL> def loess_abstract ( estimator , threshold , param , length , migration_time , utilization ) : <EOL> """<STR_LIT>""" <EOL> if len ( utilization ) < length : <EOL> return False <EOL> estimates = estimator ( utilization [ - length : ] ) <EOL> prediction = ( estimates [ <NUM_LIT:0> ] + estimates [ <NUM_LIT:1> ] * ( length + migration_time ) ) <EOL> return param * prediction >= threshold <EOL> @ contract <EOL> def mad_threshold ( param , limit , utilization ) : <EOL> """<STR_LIT>""" <EOL> return utilization_threshold_abstract ( lambda x : <NUM_LIT:1> - param * mad ( x ) , <EOL> limit , <EOL> utilization ) <EOL> @ contract <EOL> def iqr_threshold ( param , limit , utilization ) : <EOL> """<STR_LIT>""" <EOL> return utilization_threshold_abstract ( lambda x : <NUM_LIT:1> - param * iqr ( x ) , <EOL> limit , <EOL> utilization ) <EOL> @ contract <EOL> def utilization_threshold_abstract ( f , limit , utilization ) : <EOL> """<STR_LIT>""" <EOL> if ( len ( utilization ) < limit ) : <EOL> return False <EOL> return f ( utilization ) <= utilization [ - <NUM_LIT:1> ] <EOL> @ contract <EOL> def mad ( data ) : <EOL> """<STR_LIT>""" <EOL> data_median = median ( data ) <EOL> return float ( median ( [ abs ( data_median - x ) for x in data ] ) ) <EOL> @ contract <EOL> def iqr ( data ) : <EOL> """<STR_LIT>""" <EOL> sorted_data = sorted ( data ) <EOL> n = len ( data ) + <NUM_LIT:1> <EOL> q1 = int ( round ( <NUM_LIT> * n ) ) - <NUM_LIT:1> <EOL> q3 = int ( round ( <NUM_LIT> * n ) ) - <NUM_LIT:1> <EOL> return float ( sorted_data [ q3 ] - sorted_data [ q1 ] ) <EOL> @ contract <EOL> def loess_parameter_estimates ( data ) : <EOL> """<STR_LIT>""" <EOL> def f ( p , x , y , weights ) : <EOL> return weights * ( y - ( p [ <NUM_LIT:0> ] + p [ <NUM_LIT:1> ] * x ) ) <EOL> n = len ( data ) <EOL> estimates , _ = leastsq ( f , [ <NUM_LIT:1.> , <NUM_LIT:1.> ] , args = ( <EOL> np . array ( range ( <NUM_LIT:1> , n + <NUM_LIT:1> ) ) , <EOL> np . array ( data ) , <EOL> np . array ( tricube_weights ( n ) ) ) ) <EOL> return estimates . tolist ( ) <EOL> @ contract <EOL> def loess_robust_parameter_estimates ( data ) : <EOL> """<STR_LIT>""" <EOL> def f ( p , x , y , weights ) : <EOL> return weights * ( y - ( p [ <NUM_LIT:0> ] + p [ <NUM_LIT:1> ] * x ) ) <EOL> n = len ( data ) <EOL> x = np . array ( range ( <NUM_LIT:1> , n + <NUM_LIT:1> ) ) <EOL> y = np . array ( data ) <EOL> weights = np . array ( tricube_weights ( n ) ) <EOL> estimates , _ = leastsq ( f , [ <NUM_LIT:1.> , <NUM_LIT:1.> ] , args = ( x , y , weights ) ) <EOL> p = estimates . tolist ( ) <EOL> residuals = ( y - ( p [ <NUM_LIT:0> ] + p [ <NUM_LIT:1> ] * x ) ) <EOL> weights2 = np . array ( tricube_bisquare_weights ( residuals . tolist ( ) ) ) <EOL> estimates2 , _ = leastsq ( f , [ <NUM_LIT:1.> , <NUM_LIT:1.> ] , args = ( x , y , weights2 ) ) <EOL> return estimates2 . tolist ( ) <EOL> @ contract <EOL> def tricube_weights ( n ) : <EOL> """<STR_LIT>""" <EOL> spread = top = float ( n - <NUM_LIT:1> ) <EOL> weights = [ ] <EOL> for i in range ( <NUM_LIT:2> , n ) : <EOL> weights . append ( ( <NUM_LIT:1> - ( ( top - i ) / spread ) ** <NUM_LIT:3> ) ** <NUM_LIT:3> ) <EOL> return [ weights [ <NUM_LIT:0> ] , weights [ <NUM_LIT:0> ] ] + weights <EOL> @ contract <EOL> def tricube_bisquare_weights ( data ) : <EOL> """<STR_LIT>""" <EOL> n = len ( data ) <EOL> s6 = <NUM_LIT:6> * median ( map ( abs , data ) ) <EOL> weights = tricube_weights ( n ) <EOL> weights2 = [ ] <EOL> for i in range ( <NUM_LIT:2> , n ) : <EOL> weights2 . append ( weights [ i ] * ( <NUM_LIT:1> - ( data [ i ] / s6 ) ** <NUM_LIT:2> ) ** <NUM_LIT:2> ) <EOL> return [ weights2 [ <NUM_LIT:0> ] , weights2 [ <NUM_LIT:0> ] ] + weights2 </s>
