<s> import numpy as np <EOL> from scipy import optimize , sparse <EOL> from . . base import BaseEstimator , RegressorMixin <EOL> from . base import LinearModel <EOL> from . . utils import check_X_y <EOL> from . . utils import check_consistent_length <EOL> from . . utils import axis0_safe_slice <EOL> from . . utils . extmath import safe_sparse_dot <EOL> def _huber_loss_and_gradient ( w , X , y , epsilon , alpha , sample_weight = None ) : <EOL> """<STR_LIT>""" <EOL> X_is_sparse = sparse . issparse ( X ) <EOL> _ , n_features = X . shape <EOL> fit_intercept = ( n_features + <NUM_LIT:2> == w . shape [ <NUM_LIT:0> ] ) <EOL> if fit_intercept : <EOL> intercept = w [ - <NUM_LIT:2> ] <EOL> sigma = w [ - <NUM_LIT:1> ] <EOL> w = w [ : n_features ] <EOL> n_samples = np . sum ( sample_weight ) <EOL> linear_loss = y - safe_sparse_dot ( X , w ) <EOL> if fit_intercept : <EOL> linear_loss -= intercept <EOL> abs_linear_loss = np . abs ( linear_loss ) <EOL> outliers_mask = abs_linear_loss > epsilon * sigma <EOL> outliers = abs_linear_loss [ outliers_mask ] <EOL> num_outliers = np . count_nonzero ( outliers_mask ) <EOL> n_non_outliers = X . shape [ <NUM_LIT:0> ] - num_outliers <EOL> outliers_sw = sample_weight [ outliers_mask ] <EOL> n_sw_outliers = np . sum ( outliers_sw ) <EOL> outlier_loss = ( <NUM_LIT> * epsilon * np . sum ( outliers_sw * outliers ) - <EOL> sigma * n_sw_outliers * epsilon ** <NUM_LIT:2> ) <EOL> non_outliers = linear_loss [ ~ outliers_mask ] <EOL> weighted_non_outliers = sample_weight [ ~ outliers_mask ] * non_outliers <EOL> weighted_loss = np . dot ( weighted_non_outliers . T , non_outliers ) <EOL> squared_loss = weighted_loss / sigma <EOL> if fit_intercept : <EOL> grad = np . zeros ( n_features + <NUM_LIT:2> ) <EOL> else : <EOL> grad = np . zeros ( n_features + <NUM_LIT:1> ) <EOL> X_non_outliers = - axis0_safe_slice ( X , ~ outliers_mask , n_non_outliers ) <EOL> grad [ : n_features ] = ( <EOL> <NUM_LIT> / sigma * safe_sparse_dot ( weighted_non_outliers , X_non_outliers ) ) <EOL> signed_outliers = np . ones_like ( outliers ) <EOL> signed_outliers_mask = linear_loss [ outliers_mask ] < <NUM_LIT:0> <EOL> signed_outliers [ signed_outliers_mask ] = - <NUM_LIT:1.0> <EOL> X_outliers = axis0_safe_slice ( X , outliers_mask , num_outliers ) <EOL> sw_outliers = sample_weight [ outliers_mask ] * signed_outliers <EOL> grad [ : n_features ] -= <NUM_LIT> * epsilon * ( <EOL> safe_sparse_dot ( sw_outliers , X_outliers ) ) <EOL> grad [ : n_features ] += alpha * <NUM_LIT> * w <EOL> grad [ - <NUM_LIT:1> ] = n_samples <EOL> grad [ - <NUM_LIT:1> ] -= n_sw_outliers * epsilon ** <NUM_LIT:2> <EOL> grad [ - <NUM_LIT:1> ] -= squared_loss / sigma <EOL> if fit_intercept : <EOL> grad [ - <NUM_LIT:2> ] = - <NUM_LIT> * np . sum ( weighted_non_outliers ) / sigma <EOL> grad [ - <NUM_LIT:2> ] -= <NUM_LIT> * epsilon * np . sum ( sw_outliers ) <EOL> loss = n_samples * sigma + squared_loss + outlier_loss <EOL> loss += alpha * np . dot ( w , w ) <EOL> return loss , grad <EOL> class HuberRegressor ( LinearModel , RegressorMixin , BaseEstimator ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , epsilon = <NUM_LIT> , max_iter = <NUM_LIT:100> , alpha = <NUM_LIT> , <EOL> warm_start = False , fit_intercept = True , tol = <NUM_LIT> ) : <EOL> self . epsilon = epsilon <EOL> self . max_iter = max_iter <EOL> self . alpha = alpha <EOL> self . warm_start = warm_start <EOL> self . fit_intercept = fit_intercept <EOL> self . tol = tol <EOL> def fit ( self , X , y , sample_weight = None ) : <EOL> """<STR_LIT>""" <EOL> X , y = check_X_y ( <EOL> X , y , copy = False , accept_sparse = [ '<STR_LIT>' ] , y_numeric = True ) <EOL> if sample_weight is not None : <EOL> sample_weight = np . array ( sample_weight ) <EOL> check_consistent_length ( y , sample_weight ) <EOL> else : <EOL> sample_weight = np . ones_like ( y ) <EOL> if self . epsilon < <NUM_LIT:1.0> : <EOL> raise ValueError ( <EOL> "<STR_LIT>" <EOL> % self . epsilon ) <EOL> if self . warm_start and hasattr ( self , '<STR_LIT>' ) : <EOL> parameters = np . concatenate ( <EOL> ( self . coef_ , [ self . intercept_ , self . scale_ ] ) ) <EOL> else : <EOL> if self . fit_intercept : <EOL> parameters = np . zeros ( X . shape [ <NUM_LIT:1> ] + <NUM_LIT:2> ) <EOL> else : <EOL> parameters = np . zeros ( X . shape [ <NUM_LIT:1> ] + <NUM_LIT:1> ) <EOL> bounds = np . tile ( [ - np . inf , np . inf ] , ( parameters . shape [ <NUM_LIT:0> ] , <NUM_LIT:1> ) ) <EOL> bounds [ - <NUM_LIT:1> ] [ <NUM_LIT:0> ] = <NUM_LIT> <EOL> try : <EOL> parameters , f , dict_ = optimize . fmin_l_bfgs_b ( <EOL> _huber_loss_and_gradient , parameters , <EOL> args = ( X , y , self . epsilon , self . alpha , sample_weight ) , <EOL> maxiter = self . max_iter , tol = self . tol , bounds = bounds , <EOL> iprint = <NUM_LIT:0> ) <EOL> except TypeError : <EOL> parameters , f , dict_ = optimize . fmin_l_bfgs_b ( <EOL> _huber_loss_and_gradient , parameters , <EOL> args = ( X , y , self . epsilon , self . alpha , sample_weight ) , <EOL> bounds = bounds ) <EOL> self . n_iter_ = dict_ . get ( '<STR_LIT>' , None ) <EOL> self . scale_ = parameters [ - <NUM_LIT:1> ] <EOL> if self . fit_intercept : <EOL> self . intercept_ = parameters [ - <NUM_LIT:2> ] <EOL> else : <EOL> self . intercept_ = <NUM_LIT:0.0> <EOL> self . coef_ = parameters [ : X . shape [ <NUM_LIT:1> ] ] <EOL> residual = np . abs ( <EOL> y - safe_sparse_dot ( X , self . coef_ ) - self . intercept_ ) <EOL> self . outliers_ = residual > self . scale_ * self . epsilon <EOL> return self </s>
