數(shù)學(xué)推導(dǎo)+純Python實(shí)現(xiàn)機(jī)器學(xué)習(xí)算法15：GBDT

class TreeNode():    def __init__(self, feature_i=None, threshold=None,                 value=None, true_branch=None, false_branch=None):         pass

class Tree(object):    def __init__(self, min_samples_split=2, min_impurity=1e-7,                 max_depth=float("inf"), loss=None):        self.root = None  # Root node in dec. tree        # Minimum n of samples to justify split        self.min_samples_split = min_samples_split        # The minimum impurity to justify split        self.min_impurity = min_impurity        # The maximum depth to grow the tree to        self.max_depth = max_depth        # Function to calculate impurity (classif.=>info gain, regr=>variance reduct.)        # 切割樹(shù)的方法，gini，方差等        self._impurity_calculation = None        # Function to determine prediction of y at leaf        # 樹(shù)節(jié)點(diǎn)取值的方法，分類樹(shù)：選取出現(xiàn)最多次數(shù)的值，回歸樹(shù)：取所有值的平均值        self._leaf_value_calculation = None        # If y is one-hot encoded (multi-dim) or not (one-dim)        self.one_dim = None        # If Gradient Boost        self.loss = loss
    def fit(self, X, y, loss=None):        """ Build decision tree """        pass
    def _build_tree(self, X, y, current_depth=0):        """ Recursive method which builds out the decision tree and splits X and respective y        pass
    def predict_value(self, x, tree=None):        """ Do a recursive search down the tree and make a prediction of the data sample by the            value of the leaf that we end up at """        pass
    def predict(self, X):        """ Classify samples one by one and return the set of labels """        pass
    def print_tree(self, tree=None, indent=" "):        pass

class RegressionTree(Tree):    # 使用方差法進(jìn)行樹(shù)分割    def _calculate_variance_reduction(self, y, y1, y2):        var_tot = calculate_variance(y)        var_1 = calculate_variance(y1)        var_2 = calculate_variance(y2)        frac_1 = len(y1) / len(y)        frac_2 = len(y2) / len(y)        # Calculate the variance reduction        variance_reduction = var_tot - (frac_1 * var_1 + frac_2 * var_2)        return sum(variance_reduction)            # 使用均值法取葉子結(jié)點(diǎn)值    def _mean_of_y(self, y):        value = np.mean(y, axis=0)        return value if len(value) > 1 else value[0]            # 回歸樹(shù)擬合    def fit(self, X, y):        self._impurity_calculation = self._calculate_variance_reduction        self._leaf_value_calculation = self._mean_of_y        super(RegressionTree, self).fit(X, y)

class Loss(object):    def loss(self, y_true, y_pred):        return NotImplementedError()    def gradient(self, y, y_pred):        raise NotImplementedError()    def acc(self, y, y_pred):        return 0        class SquareLoss(Loss):    def __init__(self): pass    def loss(self, y, y_pred):        return 0.5 * np.power((y - y_pred), 2)    def gradient(self, y, y_pred):        return -(y - y_pred)

class GBDT(object):    def __init__(self, n_estimators, learning_rate, min_samples_split,                 min_impurity, max_depth, regression):        # 基本參數(shù)        self.n_estimators = n_estimators        self.learning_rate = learning_rate        self.min_samples_split = min_samples_split        self.min_impurity = min_impurity        self.max_depth = max_depth        self.regression = regression        self.loss = SquareLoss()        if not self.regression:            self.loss = SotfMaxLoss()        # 分類問(wèn)題也可以使用回歸樹(shù)，利用殘差去學(xué)習(xí)概率        self.estimators = []        for i in range(self.n_estimators):            self.estimators.append(RegressionTree(min_samples_split=self.min_samples_split,                                             min_impurity=self.min_impurity,                                             max_depth=self.max_depth))    # 擬合方法    def fit(self, X, y):        # 讓第一棵樹(shù)去擬合模型        self.estimators[0].fit(X, y)        y_pred = self.estimators[0].predict(X)        for i in range(1, self.n_estimators):            gradient = self.loss.gradient(y, y_pred)            self.estimators[i].fit(X, gradient)            y_pred -= np.multiply(self.learning_rate, self.estimators[i].predict(X))    # 預(yù)測(cè)方法    def predict(self, X):        y_pred = self.estimators[0].predict(X)        for i in range(1, self.n_estimators):            y_pred -= np.multiply(self.learning_rate, self.estimators[i].predict(X))        if not self.regression:            # Turn into probability distribution            y_pred = np.exp(y_pred) / np.expand_dims(np.sum(np.exp(y_pred), axis=1), axis=1)            # Set label to the value that maximizes probability            y_pred = np.argmax(y_pred, axis=1)        return y_pred

# regression treeclass GBDTRegressor(GBDT):      def __init__(self, n_estimators=200, learning_rate=0.5, min_samples_split=2,                 min_var_red=1e-7, max_depth=4, debug=False):        super(GBDTRegressor, self).__init__(n_estimators=n_estimators,                                            learning_rate=learning_rate,                                            min_samples_split=min_samples_split,                                            min_impurity=min_var_red,                                            max_depth=max_depth,                                            regression=True)# classification treeclass GBDTClassifier(GBDT):      def __init__(self, n_estimators=200, learning_rate=.5, min_samples_split=2,                 min_info_gain=1e-7, max_depth=2, debug=False):         super(GBDTClassifier, self).__init__(n_estimators=n_estimators,                                             learning_rate=learning_rate,                                             min_samples_split=min_samples_split,                                             min_impurity=min_info_gain,                                             max_depth=max_depth,                                             regression=False)      def fit(self, X, y):        y = to_categorical(y)        super(GBDTClassifier, self).fit(X, y)

from sklearn import datasetsboston = datasets.load_boston()X, y = shuffle_data(boston.data, boston.target, seed=13)X = X.astype(np.float32)offset = int(X.shape[0] * 0.9)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3)print(X_train.shape, y_train.shape, X_test.shape, y_test.shape)
model = GBDTRegressor()model.fit(X_train, y_train)y_pred = model.predict(X_test)# Color mapcmap = plt.get_cmap('viridis')mse = mean_squared_error(y_test, y_pred)print ("Mean Squared Error:", mse)
# Plot the resultsm1 = plt.scatter(range(X_test.shape[0]), y_test, color=cmap(0.5), s=10)m2 = plt.scatter(range(X_test.shape[0]), y_pred, color='black', s=10)plt.suptitle("Regression Tree")plt.title("MSE: %.2f" % mse, fontsize=10)plt.xlabel('sample')plt.ylabel('house price')plt.legend((m1, m2), ("Test data", "Prediction"), loc='lower right')plt.show();