<s> import numpy as np <EOL> from . . util . warping_functions import * <EOL> from . . core import GP <EOL> from . . import likelihoods <EOL> from GPy . util . warping_functions import TanhWarpingFunction_d <EOL> from GPy import kern <EOL> class WarpedGP ( GP ) : <EOL> def __init__ ( self , X , Y , kernel = None , warping_function = None , warping_terms = <NUM_LIT:3> ) : <EOL> if kernel is None : <EOL> kernel = kern . RBF ( X . shape [ <NUM_LIT:1> ] ) <EOL> if warping_function == None : <EOL> self . warping_function = TanhWarpingFunction_d ( warping_terms ) <EOL> self . warping_params = ( np . random . randn ( self . warping_function . n_terms * <NUM_LIT:3> + <NUM_LIT:1> ) * <NUM_LIT:1> ) <EOL> else : <EOL> self . warping_function = warping_function <EOL> self . scale_data = False <EOL> if self . scale_data : <EOL> Y = self . _scale_data ( Y ) <EOL> self . Y_untransformed = Y . copy ( ) <EOL> self . predict_in_warped_space = True <EOL> likelihood = likelihoods . Gaussian ( ) <EOL> GP . __init__ ( self , X , self . transform_data ( ) , likelihood = likelihood , kernel = kernel ) <EOL> self . link_parameter ( self . warping_function ) <EOL> def _scale_data ( self , Y ) : <EOL> self . _Ymax = Y . max ( ) <EOL> self . _Ymin = Y . min ( ) <EOL> return ( Y - self . _Ymin ) / ( self . _Ymax - self . _Ymin ) - <NUM_LIT:0.5> <EOL> def _unscale_data ( self , Y ) : <EOL> return ( Y + <NUM_LIT:0.5> ) * ( self . _Ymax - self . _Ymin ) + self . _Ymin <EOL> def parameters_changed ( self ) : <EOL> self . Y [ : ] = self . transform_data ( ) <EOL> super ( WarpedGP , self ) . parameters_changed ( ) <EOL> Kiy = self . posterior . woodbury_vector . flatten ( ) <EOL> grad_y = self . warping_function . fgrad_y ( self . Y_untransformed ) <EOL> grad_y_psi , grad_psi = self . warping_function . fgrad_y_psi ( self . Y_untransformed , <EOL> return_covar_chain = True ) <EOL> djac_dpsi = ( ( <NUM_LIT:1.0> / grad_y [ : , : , None , None ] ) * grad_y_psi ) . sum ( axis = <NUM_LIT:0> ) . sum ( axis = <NUM_LIT:0> ) <EOL> dquad_dpsi = ( Kiy [ : , None , None , None ] * grad_psi ) . sum ( axis = <NUM_LIT:0> ) . sum ( axis = <NUM_LIT:0> ) <EOL> warping_grads = - dquad_dpsi + djac_dpsi <EOL> self . warping_function . psi . gradient [ : ] = warping_grads [ : , : - <NUM_LIT:1> ] <EOL> self . warping_function . d . gradient [ : ] = warping_grads [ <NUM_LIT:0> , - <NUM_LIT:1> ] <EOL> def transform_data ( self ) : <EOL> Y = self . warping_function . f ( self . Y_untransformed . copy ( ) ) . copy ( ) <EOL> return Y <EOL> def log_likelihood ( self ) : <EOL> ll = GP . log_likelihood ( self ) <EOL> jacobian = self . warping_function . fgrad_y ( self . Y_untransformed ) <EOL> return ll + np . log ( jacobian ) . sum ( ) <EOL> def plot_warping ( self ) : <EOL> self . warping_function . plot ( self . Y_untransformed . min ( ) , self . Y_untransformed . max ( ) ) <EOL> def _get_warped_term ( self , mean , std , gh_samples , pred_init = None ) : <EOL> arg1 = gh_samples . dot ( std . T ) * np . sqrt ( <NUM_LIT:2> ) <EOL> arg2 = np . ones ( shape = gh_samples . shape ) . dot ( mean . T ) <EOL> return self . warping_function . f_inv ( arg1 + arg2 , y = pred_init ) <EOL> def _get_warped_mean ( self , mean , std , pred_init = None , deg_gauss_hermite = <NUM_LIT:100> ) : <EOL> """<STR_LIT>""" <EOL> gh_samples , gh_weights = np . polynomial . hermite . hermgauss ( deg_gauss_hermite ) <EOL> gh_samples = gh_samples [ : , None ] <EOL> gh_weights = gh_weights [ None , : ] <EOL> return gh_weights . dot ( self . _get_warped_term ( mean , std , gh_samples ) ) / np . sqrt ( np . pi ) <EOL> def _get_warped_variance ( self , mean , std , pred_init = None , deg_gauss_hermite = <NUM_LIT:100> ) : <EOL> """<STR_LIT>""" <EOL> gh_samples , gh_weights = np . polynomial . hermite . hermgauss ( deg_gauss_hermite ) <EOL> gh_samples = gh_samples [ : , None ] <EOL> gh_weights = gh_weights [ None , : ] <EOL> arg1 = gh_weights . dot ( self . _get_warped_term ( mean , std , gh_samples , <EOL> pred_init = pred_init ) ** <NUM_LIT:2> ) / np . sqrt ( np . pi ) <EOL> arg2 = self . _get_warped_mean ( mean , std , pred_init = pred_init , <EOL> deg_gauss_hermite = deg_gauss_hermite ) <EOL> return arg1 - ( arg2 ** <NUM_LIT:2> ) <EOL> def predict ( self , Xnew , which_parts = '<STR_LIT:all>' , pred_init = None , full_cov = False , Y_metadata = None , <EOL> median = False , deg_gauss_hermite = <NUM_LIT:100> ) : <EOL> mu , var = GP . _raw_predict ( self , Xnew ) <EOL> mean , var = self . likelihood . predictive_values ( mu , var ) <EOL> if self . predict_in_warped_space : <EOL> std = np . sqrt ( var ) <EOL> if median : <EOL> wmean = self . warping_function . f_inv ( mean , y = pred_init ) <EOL> else : <EOL> wmean = self . _get_warped_mean ( mean , std , pred_init = pred_init , <EOL> deg_gauss_hermite = deg_gauss_hermite ) . T <EOL> wvar = self . _get_warped_variance ( mean , std , pred_init = pred_init , <EOL> deg_gauss_hermite = deg_gauss_hermite ) . T <EOL> else : <EOL> wmean = mean <EOL> wvar = var <EOL> if self . scale_data : <EOL> pred = self . _unscale_data ( pred ) <EOL> return wmean , wvar <EOL> def predict_quantiles ( self , X , quantiles = ( <NUM_LIT> , <NUM_LIT> ) , Y_metadata = None ) : <EOL> """<STR_LIT>""" <EOL> m , v = self . _raw_predict ( X , full_cov = False ) <EOL> if self . normalizer is not None : <EOL> m , v = self . normalizer . inverse_mean ( m ) , self . normalizer . inverse_variance ( v ) <EOL> a , b = self . likelihood . predictive_quantiles ( m , v , quantiles , Y_metadata ) <EOL> if not self . predict_in_warped_space : <EOL> return [ a , b ] <EOL> new_a = self . warping_function . f_inv ( a ) <EOL> new_b = self . warping_function . f_inv ( b ) <EOL> return [ new_a , new_b ] <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> X = np . random . randn ( <NUM_LIT:100> , <NUM_LIT:1> ) <EOL> Y = np . sin ( X ) + np . random . randn ( <NUM_LIT:100> , <NUM_LIT:1> ) * <NUM_LIT> <EOL> m = WarpedGP ( X , Y ) </s>
