<s> __author__ = '<STR_LIT>' <EOL> import numpy as np <EOL> from sklearn import cross_validation , metrics <EOL> from sklearn . decomposition import PCA <EOL> import multiprocessing <EOL> import copy <EOL> import matplotlib . pyplot as plt <EOL> class SupervisedPCABase ( object ) : <EOL> def __init__ ( self , regressor , max_components = None , n_components = <NUM_LIT:1> , whiten = True ) : <EOL> """<STR_LIT>""" <EOL> self . regressor = regressor <EOL> self . max_components = max_components <EOL> self . pca_object = PCA ( n_components = n_components , whiten = whiten ) <EOL> self . n_components = n_components <EOL> self . whiten = whiten <EOL> self . n_reduced = <NUM_LIT:0> <EOL> self . sort_idx = np . zeros ( <NUM_LIT:1> ) <EOL> def _compute_stnd_coefs ( self , X , y ) : <EOL> """<STR_LIT>""" <EOL> p = X . shape [ <NUM_LIT:1> ] <EOL> scoefs = np . zeros ( p ) <EOL> for j in xrange ( p ) : <EOL> thisX = X [ : , j ] <EOL> self . regressor . fit ( thisX [ : , np . newaxis ] , y ) <EOL> scoefs [ j ] = self . regressor . coef_score_ <EOL> return scoefs <EOL> def _get_reduced_features ( self , X , coefs , pmax ) : <EOL> """<STR_LIT>""" <EOL> sort_idx = np . argsort ( coefs ) [ : : - <NUM_LIT:1> ] <EOL> sort_idx = sort_idx [ : pmax ] <EOL> self . pca_object . fit ( X [ : , sort_idx ] ) <EOL> X_reduced = self . pca_object . transform ( X [ : , sort_idx ] ) <EOL> return X_reduced , sort_idx <EOL> def fit ( self , X , y , n_reduced ) : <EOL> """<STR_LIT>""" <EOL> scoefs = self . _compute_stnd_coefs ( X , y ) <EOL> X_reduced , sort_idx = self . _get_reduced_features ( X , scoefs , n_reduced ) <EOL> self . sort_idx = sort_idx <EOL> self . regressor . fit ( X_reduced , y ) <EOL> def predict ( self , X ) : <EOL> """<STR_LIT>""" <EOL> X_reduced = self . pca_object . transform ( X [ : , self . sort_idx ] ) <EOL> y_predict = self . regressor . predict ( X_reduced ) <EOL> return y_predict <EOL> def launch_coef_scores ( args ) : <EOL> """<STR_LIT>""" <EOL> spca , X , y = args <EOL> scoefs = spca . _compute_stnd_coefs ( X , y ) <EOL> return scoefs <EOL> def compute_cv_prediction ( args ) : <EOL> """<STR_LIT>""" <EOL> spca , X_train , y_train , X_test , n_reduced , scoef = args <EOL> SPCA = SupervisedPCABase ( copy . deepcopy ( spca . regressor ) , spca . max_components , spca . n_components , spca . whiten ) <EOL> X_reduced , sort_idx = SPCA . _get_reduced_features ( X_train , scoef , n_reduced ) <EOL> SPCA . regressor . fit ( X_reduced , y_train ) <EOL> X_test_reduced = SPCA . pca_object . transform ( X_test [ : , sort_idx ] ) <EOL> y_predict = SPCA . regressor . predict ( X_test_reduced ) <EOL> return y_predict <EOL> class SupervisedPCA ( SupervisedPCABase ) : <EOL> def __init__ ( self , regressor , max_components = None , n_components = <NUM_LIT:1> , whiten = True , n_jobs = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> super ( SupervisedPCA , self ) . __init__ ( regressor , max_components , n_components , whiten ) <EOL> if n_jobs < <NUM_LIT:0> : <EOL> n_jobs = multiprocessing . cpu_count ( ) <EOL> self . n_jobs = n_jobs <EOL> def _compute_cv_prediction ( self , args ) : <EOL> """<STR_LIT>""" <EOL> X_train , y_train , X_test , n_reduced , scoef = args <EOL> SPCA = SupervisedPCABase ( copy . deepcopy ( self . regressor ) , self . max_components , self . n_components , self . whiten ) <EOL> X_reduced , sort_idx = SPCA . _get_reduced_features ( X_train , scoef , n_reduced ) <EOL> SPCA . regressor . fit ( X_reduced , y_train ) <EOL> X_test_reduced = SPCA . pca_object . transform ( X_test [ : , sort_idx ] ) <EOL> y_predict = SPCA . regressor . predict ( X_test_reduced ) <EOL> return y_predict <EOL> def _launch_coef_scores ( self , args ) : <EOL> """<STR_LIT>""" <EOL> X , y = args <EOL> scoefs = self . _compute_stnd_coefs ( X , y ) <EOL> return scoefs <EOL> def choose_nreduced ( self , X , y , lossfunc = None , cv = None , verbose = False , cvplot = False ) : <EOL> """<STR_LIT>""" <EOL> if self . n_jobs > <NUM_LIT:1> : <EOL> pool = multiprocessing . Pool ( self . n_jobs ) <EOL> pool . map ( int , range ( self . n_jobs ) ) <EOL> if cv is None : <EOL> K_folds = <NUM_LIT:8> <EOL> if isinstance ( cv , int ) : <EOL> K_folds = cv <EOL> cv = cross_validation . KFold ( y . size , n_folds = K_folds ) <EOL> if lossfunc is None : <EOL> lossfunc = metrics . mean_squared_error <EOL> if self . max_components is None : <EOL> self . max_components = X . shape [ <NUM_LIT:1> ] <EOL> if verbose : <EOL> print '<STR_LIT>' , self . max_components , '<STR_LIT>' <EOL> print '<STR_LIT>' <EOL> sargs = [ ] <EOL> for train_idx , test_idx in cv : <EOL> if self . n_jobs == <NUM_LIT:1> : <EOL> sargs . append ( ( X [ train_idx , : ] , y [ train_idx ] ) ) <EOL> else : <EOL> sargs . append ( ( self , X [ train_idx , : ] , y [ train_idx ] ) ) <EOL> if self . n_jobs == <NUM_LIT:1> : <EOL> scoefs = map ( self . _launch_coef_scores , sargs ) <EOL> else : <EOL> scoefs = pool . map ( launch_coef_scores , sargs ) <EOL> cverrors = np . zeros ( self . max_components ) <EOL> if verbose : <EOL> print '<STR_LIT>' , self . max_components , '<STR_LIT>' , '<STR_LIT>' <EOL> for k in xrange ( self . max_components ) : <EOL> cverror_args = [ ] <EOL> ytest = [ ] <EOL> fold_idx = <NUM_LIT:0> <EOL> for train_idx , test_idx in cv : <EOL> if self . n_jobs == <NUM_LIT:1> : <EOL> cverror_args . append ( ( X [ train_idx , : ] , y [ train_idx ] , X [ test_idx , : ] , k + <NUM_LIT:1> , scoefs [ fold_idx ] ) ) <EOL> else : <EOL> cverror_args . append ( ( self , X [ train_idx , : ] , y [ train_idx ] , X [ test_idx , : ] , k + <NUM_LIT:1> , scoefs [ fold_idx ] ) ) <EOL> ytest . append ( y [ test_idx ] ) <EOL> fold_idx += <NUM_LIT:1> <EOL> if self . n_jobs == <NUM_LIT:1> : <EOL> ypredictions = map ( self . _compute_cv_prediction , cverror_args ) <EOL> else : <EOL> ypredictions = pool . map ( compute_cv_prediction , cverror_args ) <EOL> cverror_k = <NUM_LIT:0.0> <EOL> for yt , yp in zip ( ytest , ypredictions ) : <EOL> cverror_k += lossfunc ( yt , yp ) / K_folds <EOL> cverrors [ k ] = cverror_k <EOL> if cvplot : <EOL> plt . plot ( np . arange ( <NUM_LIT:1> , self . max_components + <NUM_LIT:1> ) , cverrors ) <EOL> plt . xlabel ( '<STR_LIT>' ) <EOL> plt . ylabel ( '<STR_LIT>' ) <EOL> plt . show ( ) <EOL> n_reduced = cverrors . argmin ( ) + <NUM_LIT:1> <EOL> if verbose : <EOL> print '<STR_LIT>' , n_reduced , '<STR_LIT>' <EOL> return n_reduced </s>
