Decision Tree (구현)

결정 트리(Decision Tree)는 스무고개 게임과 유사하여 룰 기반의 프로그램에 적용되는 `if`, `else`를 자동으로 찾아내(분할 규칙) 예측을 위한 알고리즘

결국 결정트리를 생성하는 것은 주어진 특성공간을 분할 규칙에 따라 분할하는 것과 같음

 

학습 데이터 $D=\{(x_i,y_i)|1\le i \le m\}$의 특성벡터 $x_i\, (1\le i \le m)$를 포함하는 특성공간  $\mathcal X$를 어떤 <span style="color:blue"> 분할 규칙(splitting rule)</span>에 따라 겹치지 않는 작은 영역 $\mathcal R_i$로 나눔
$$\mathcal X = \mathcal R_1 \cup \mathcal R_2 \cup \cdots \cup \mathcal R_N$$

 

회귀문제인지 분류문제인지에 따라, 임의의 샘플벡터 $x$에 대해 다음과 같이 예측 

Regression

샘플 $x$가 속하는 작은 영역 $\mathcal R_i$에 대해, 이 영역에 속하는 훈련샘플 $x_j$의 $y_j$값의 평균으로 예측 
$$\hat y = \dfrac 1 {r_i} \sum_{x_j \in \mathcal R_i}y_j, \quad (r_i=\bigl|\{(x_j,y_j)\in D|x_j\in \mathcal R_i\}\bigr|)$$

 

Classification

샘플 $x$가 속하는 작은 영역 $\mathcal R_i$에 대해, $\mathcal R_i$에 속하는 훈련샘플에 대한 레이블 중 가장 많이 나타나는 레이블  

1. 트리구조를 구현하기 위해 트리노드 클래스를 구현

class TNode:
    def __init__(self,depth, X, y):

        self.depth = depth                      # 트리 max depth
        self.X = X                              # train_X (Feature)
        self.y = y                              # train_y (Label)
        self.xi = None                          # 분할 인덱스
        self.left = None                        # 왼쪽 자식 노드
        self.right = None                       # 오른쪽 자식 노드
        self.predictor = None                   # 예측 함수

    def CalculateLoss(self):
        if len(self.y)==0:
            return 0
        else:
            ####### Empty Module.1 #######
            return np.sum((self.y - np.mean(self.y)) ** 2)     # SSE Loss를 직접 구현
            ##############################

 

2. 결정트리(Decision Tree) 생성 함수

def DataSplit(X, y, j, xi):
    ####### Empty Module.2 #######
    ids = X[:, j] <= xi                # xi보다 작거나 같은 X들의 인덱스 추출
    Xt = X[ids]                        # ids가 True인 X들만 샘플링
    Xf = X[~ids]                       # ids가 False인 X들만 샘플링
    yt = y[ids]                        # ids가 True인 y들만 샘플링
    yf = y[~ids]                       # ids가 False인 y들만 샘플링
    ##############################
    return Xt, yt, Xf, yf

def CalculateOptimalSplit(node):
    X = node.X
    y = node.y
    best_feature = 0
    bext_xi = X[0, best_feature]
    best_split_val = node.CalculateLoss()
    
    m,n = X.shape
    
    for j in range(0,n):
        for i in range(0,m):
            xi = X[i,j]
            Xt, yt, Xf, yf = DataSplit(X,y,j,xi)
            tmpt = TNode(0, Xt, yt)
            tmpf = TNode(0, Xf, yf)
            loss_t = tmpt.CalculateLoss()
            loss_f = tmpf.CalculateLoss()
            curr_val = loss_t + loss_f
            ####### Empty Module.3 #######
            if curr_val < best_split_val: 
                best_split_val = curr_val       # loss 업데이트
                best_feature = j                # best_feature 업데이트
                best_xi = xi                    # best_xi 업데이트
            ##############################

    return best_feature, best_xi

 

3. 결정트리(Decision Tree) 생성 함수

def Construct_Subtree(node, max_depth):
    if (node.depth == max_depth or len(node.y) == 1): # node의 깊이가 max_depth에 도달했거나 리프 노드일 때
        ####### Empty Module.4 #######
        node.predictor = np.mean(node.y)            # node 내부에 있는 y값들의 평균을 활용하여 예측 수행
        ##############################
    else:
        j, xi = CalculateOptimalSplit(node)                
        node.j = j
        node.xi = xi
        Xt, yt, Xf, yf = DataSplit(node.X, node.y, j, xi)  

        if (len(yt)>0): 
            ####### Empty Module.5 #######
            node.left = TNode(node.depth + 1, Xt, yt) # TNode를 활용하여 새로운 왼쪽 자식 노드 구축
            Construct_Subtree(node.left, max_depth)   # Construct_Subtree를 활용하여 왼쪽 자식 노드에 대한 Subtree 구축
            ##############################
        if (len(yf)>0): 
            ####### Empty Module.6 #######
            node.right = TNode(node.depth + 1, Xf, yf) # TNode를 활용하여 새로운 오른쪽 자식 노드 구축
            Construct_Subtree(node.right, max_depth)  # Construct_Subtree를 활용하여 오른쪽 자식 노드에 대한 Subtree 구축
            ##############################

    return node

 

4.결정트리(Decision Tree) 예측기 구현

def Predict(X, node):
    ####### Empty Module.7 #######
    if node.right is None and node.left is not None:
        return Predict(X, node.left)                 # 재귀적으로 왼쪽 node에 대해서 Predict 함수 호출
    
    if node.right is not None and node.left is None:
        return Predict(X, node.right)                # 재귀적으로 오른쪽 node에 대해서 Predict 함수 호출
    
    if node.right is None and node.left is None:
        return node.predictor 
    else:
        if X[node.j] <= node.xi:
            return Predict(X, node.left)             # 재귀적으로 왼쪽 node에 대해서 Predict 함수 호출
        else:
            return Predict(X, node.right)            # 재귀적으로 오른쪽 node에 대해서 Predict 함수 호출
    ##############################

 

5. 실습 데이터셋 생성 및 결정트리 실습

def makedata():
    n_samples = 1000
    X, y = make_friedman1(n_samples = n_samples, n_features = 5, noise=1.0, random_state=100)
    return train_test_split(X, y, test_size=0.5, random_state=3)

 

6. sklearn.tree 모듈의 DecisionTreeRegressor와 비교

X_train, X_test, y_train, y_test = makedata()

# 결정트리 깊이 설정
max_depth = 10

# depth 0에서 루트노드 생성 
treeRoot = TNode(0, X_train, y_train)

# 결정트리 학습 
Construct_Subtree(treeRoot, max_depth)

# 예측 
y_hat = np.zeros(len(X_test))

for i in range(len(X_test)):
    y_hat[i] = Predict(X_test[i], treeRoot)

regTree = DecisionTreeRegressor(max_depth = 10, random_state=0)

regTree.fit(X_train, y_train)
y_hat2 = regTree.predict(X_test)

MSE_scratch = np.mean(np.power(y_hat-y_test,2))
MSE_scikit = np.mean(np.power(y_hat2-y_test, 2))

print("사이킷런 결정트리:loss= {:.3f}".format(MSE_scikit))
print("직접구현 결정트리:loss= {:.3f}".format(MSE_scratch))