Dacon 손동작 분류 경진대회

이번에 데이콘 베이직 스터디를 시작하게 되었다. 스터디 시작 시점에 진행 중이던 손동작 분류 경진대회가 얼마 남지 않은 탓에 다음 대회부터 참여하기로 하고, 이번 주는 개인적으로 손동작 분류 경진대회에 대해 공부해보기로 했다.

대회 마감이 얼마 남지 않은 시점이라, 대회가 끝나고 수상자 분들께서 공유한 코드를 보고 공부해야겠다고 생각했다. 전복 나이 예측 경진대회는 열심히 공부하며 EDA부터 모델링까지 최선을 다해 해보겠다고 다짐! (근데 전복 나이를 예측하는 게 좀 귀엽..)

따라서 이번 포스팅에서는 해당 대회에서 Public/Private score 1위를 기록하신 분의 코드를 참고하며, 수상자 분들의 코드에서 공통적으로 쓰인 합성곱 신경망(CNN) 대해 알아보고자 한다.

👋 손동작 분류 경진대회

주어진 센서 데이터를 이용해 손동작을 분류하는 경진대회

데이터

1. train.csv : 학습 데이터

• id : 샘플 아이디
• sensor_1 ~ sensor_32 : 센서 데이터
• target : 손동작 class

2. test.csv : 테스트 데이터

• id : 샘플 아이디
• sensor_1 ~ sensor_32 : 센서 데이터

3. sample_submissoin.csv : 제출 양식

• id : 샘플 아이디
• target : 손동작 class

EDA

라이브러리 import

import pandas as pd
from sklearn.preprocessing import MinMaxScaler
import matplotlib.pyplot as plt
import seaborn as sns
import numpy as np
from typing import List, Tuple
from sklearn.preprocessing import OneHotEncoder
from sklearn.model_selection import StratifiedKFold
from sklearn.metrics import accuracy_score
import tensorflow as tf
from keras.layers import Conv2D, BatchNormalization, GlobalAveragePooling2D, Dropout, Dense, Flatten, MaxPooling2D, GlobalAveragePooling1D
from keras.models import Sequential, load_model
from keras.losses import CategoricalCrossentropy, SparseCategoricalCrossentropy
from keras.callbacks import EarlyStopping, ModelCheckpoint

데이터 불러오기

train = pd.read_csv('train.csv')
train.head()

train 데이터의 모양을 살펴보면 id, sensor_1~sensor_32, target까지 총 34개의 열이 존재한다. 총 32개의 센서 데이터를 이용하여 target 값을 예측하는 것이 이번 경진대회의 목표이다. 센서 데이터는 모두 수치형 데이터이다.

결측치 확인

def check_missing_col(dataframe):
    missing_col = []
    counted_missing_col = 0
    for i, col in enumerate(dataframe.columns):
        missing_values = sum(dataframe[col].isna())
        is_missing = True if missing_values >= 1 else False
        if is_missing:
            counted_missing_col += 1
            print(f'결측치가 있는 컬럼은: {col}입니다')
            print(f'해당 컬럼에 총 {missing_values}개의 결측치가 존재합니다.')
            missing_col.append([col, dataframe[col].dtype])
    if counted_missing_col == 0:
        print('결측치가 존재하지 않습니다')
    return missing_col

missing_col = check_missing_col(train)


결측치가 존재하지 않습니다

결측치가 존재하지 않으므로 별도의 결측치 처리는 필요하지 않아 보인다.

상관관계 분석

plt.figure(figsize=(25,25))

heat_table = train.corr()
mask = np.zeros_like(heat_table)
mask[np.triu_indices_from(mask)] = True
heatmap_ax = sns.heatmap(heat_table, annot=True, mask = mask, cmap='coolwarm')
heatmap_ax.set_xticklabels(heatmap_ax.get_xticklabels(), fontsize=10, rotation=45)
heatmap_ax.set_yticklabels(heatmap_ax.get_yticklabels(), fontsize=10)
plt.title('correlation between features', fontsize=30)
plt.show()

센서들 간의 상관관계를 보면, 4 간격으로 음의 상관관계를 보이는 것을 알 수 있다. 따라서, 4씩 차이 나는 센서들을 하나의 feature로 묶어주기 위해 각 데이터를 (8,4) 행렬로 만들어주어야 한다.

Preprocessing

전처리 함수 정의

def data_preprocess(data:pd.DataFrame, dim:List[int], is_train:bool):
    ohe = OneHotEncoder(sparse = False)
    if is_train:
        x = data.iloc[:,1:-1]
        y = ohe.fit_transform(data[['target']])

    else:
        x = data.iloc[:,1:]
        y = None
    x = np.array(x).reshape(dim)
    
    return x, y

data_preprocess로 정의된 전처리 함수에는 data, dimension, True/False(학습 데이터/테스트 데이터) 값을 입력해줘야 하고, 함수 내에서는 이를 기반으로 학습 데이터일 때와 테스트 데이터일 때의 전처리를 각각 수행한다.

먼저, 학습 데이터일 때는 x는 id와 target을 제외한 sensor_1~sensor_32 column들을 정의하고, y는 target 값을 원-핫 인코딩 해준 값으로 정의한다.

테스트 데이터라면, x는 id를 제외한 column들로 정의하고 y는 따로 정의해주지 않는다.

마지막으로 x를 입력된 차원(dimension)으로 재배열한 뒤, x와 y를 반환한다.

전처리

data_shape = [-1,8,4,1]
x1, y1 = data_preprocess(train,data_shape,True)
test_x1, _ = data_preprocess(test,data_shape, False)

전처리 함수의 매개변수 dim에 [-1, 8, 4, 1] 값을 넣어줌으로써 각 데이터를 (8,4) 행렬로 재배열 해주었고, 4씩 차이 나는 센서들을 하나의 feature로 묶어주었다.

Modeling

모델 함수 정의

def create_model(filters:int, kernels:List[Tuple], input_shape ,activation,max_pool_shapes, has_dense):
    model = Sequential([
        Conv2D(filters=filters, kernel_size=kernels[0],padding='same',activation=activation, input_shape = input_shape),
        BatchNormalization(),
        Dropout(0.2),
    ])

    for i in range(1, len(kernels)):
        model.add(Conv2D(filters=filters, kernel_size=kernels[i],padding='same',activation=activation))
        model.add(BatchNormalization())
        if max_pool_shapes:
            model.add(MaxPooling2D(max_pool_shapes[i - 1]))
        model.add(Dropout(0.2))

    model.add(GlobalAveragePooling2D())
    if has_dense:
        model.add(Dense(64, activation = activation))
    model.add(Dropout(0.2))
    model.add(Dense(4, activation = 'softmax'))

    return model

모델은 이미지 분류에 높은 성능을 보이는 CNN을 이용하였고, 배치 정규화를 이용해 학습 속도를 가속화시키는 동시에 가중치 초기화에 대한 민감도를 감소시켰다. 또한 어떤 특정한 Feature만을 과도하게 집중하여 학습함으로써 발생할 수 있는 과적합(Overfitting)을 방지하기 위해 dropout을 이용하였다.

# 교차 검증
skf = StratifiedKFold(n_splits = 5, random_state = 42, shuffle = True)

# Early Stopping
es = EarlyStopping(monitor = 'val_acc', patience = 250, mode = 'max', verbose = 0)

KFold 교차 검증과 Early Stopping 모두 과적합 방지에 도움이 된다.

acc1 = []
acc2 = []
acc3 = []
pred = np.zeros((test_x1.shape[0], 4))
filters = 32
activation = 'elu'

kernels1 = [(1,1), (8,1), (2,4), (2,4)]
max_pool_shape1 = []
input_shape1 = data_shape[1:]

kernels2 = [(1,1),(2,4), (4,2), (2,2)]
max_pool_shape2 = []
input_shape2 = data_shape[1:]

kernels3 = [(1,1),(4,1), (4,2), (2,4)]
max_pool_shape3 = []
input_shape3 = data_shape[1:]

데이터를 normalize하지 않았으니 첫번째 conv의 경우 (1,1)을 통해 각 feature들의 값을 batchnorm 해준다.

이때 activation을 None으로 주어 음수인 데이터를 그대로 사용하는 것 보다는 elu를 사용 하는 것이 score가 더 높게 나옴을 확인

그 다음 feature extration을 위한 conv로는 Nx1 conv를 이용하여 위에서 확인 했던 음의 상관관계를 가지는 센서들을 우선적으로 conv 연산을 해주었다. (kernel1, kernel2)

kernel3의 경우 첫 conv는 2x4지만 이후 4x2를 통과 시켜줌으로 다음 레이어에서 최대한 원본 데이터 column을 전부 커버 할 수 있게 만들었다.

for i, (train_idx, valid_idx) in enumerate(skf.split(x1, train.target)) :
    print(f'{i + 1} Fold Training.....')
    train_x1, train_y1 = x1[train_idx], y1[train_idx]
    valid_x1, valid_y1 = x1[valid_idx], y1[valid_idx]

    with tf.device('/device:GPU:0'):

        model1 = create_model(filters=filters, 
                            kernels=kernels1,
                            input_shape=input_shape1,
                            activation=activation, 
                            max_pool_shapes=max_pool_shape1,
                            has_dense = False)
        
        model2 = create_model(filters=filters, 
                            kernels=kernels2,
                            input_shape=input_shape1,
                            activation=activation, 
                            max_pool_shapes=max_pool_shape2,
                            has_dense = True)
        
        # model2의 경우 dense layer를 한번 더 통과
        
        model3 = create_model(filters=filters, 
                            kernels=kernels3,
                            input_shape=input_shape3,
                            activation=activation, 
                            max_pool_shapes=max_pool_shape3,
                            has_dense = False)
        
        # ModelCheckPoint Fold마다 갱신        
        mc1 = ModelCheckpoint(f'model1_{i + 1}.h5', save_best_only = True, monitor = 'val_acc', mode = 'max', verbose = 0)
        mc2 = ModelCheckpoint(f'model2_{i + 1}.h5', save_best_only = True, monitor = 'val_acc', mode = 'max', verbose = 0)
        mc3 = ModelCheckpoint(f'model3_{i + 1}.h5', save_best_only = True, monitor = 'val_acc', mode = 'max', verbose = 0)

        cce = CategoricalCrossentropy(label_smoothing = 0.3) # 일반화를 위한 label smoothing
        
        # 모델 compile
        model1.compile(optimizer = tf.keras.optimizers.Adam(learning_rate = 0.001), loss = cce, metrics = ['acc'])
        model2.compile(optimizer = tf.keras.optimizers.Adam(learning_rate = 0.001), loss = cce, metrics = ['acc'])
        model3.compile(optimizer = tf.keras.optimizers.Adam(learning_rate = 0.001), loss = cce, metrics = ['acc'])

        model1.fit(train_x1, train_y1, validation_data = (valid_x1, valid_y1), epochs = 1000, batch_size = 512, callbacks = [es, mc1], verbose = 0)
        model2.fit(train_x1, train_y1, validation_data = (valid_x1, valid_y1), epochs = 1000, batch_size = 512, callbacks = [es, mc2], verbose = 0)
        model3.fit(train_x1, train_y1, validation_data = (valid_x1, valid_y1), epochs = 1000, batch_size = 512, callbacks = [es, mc3], verbose = 0)

	# 최고 성능 기록 모델 Load
    best1 = load_model(f'model1_{i + 1}.h5')
    best2 = load_model(f'model2_{i + 1}.h5')
    best3 = load_model(f'model3_{i + 1}.h5')

    valid_pred1 = best1.predict(valid_x1)
    valid_pred2 = best2.predict(valid_x1)
    valid_pred3 = best3.predict(valid_x1)

    ensem_pred = np.argmax(valid_pred1 + valid_pred2 + valid_pred3, axis = 1)

    valid_pred1 = np.argmax(valid_pred1, axis = 1)
    valid_pred2 = np.argmax(valid_pred2, axis = 1)
    valid_pred3 = np.argmax(valid_pred3, axis = 1)

	# Fold별 정확도 산출
    fold_model1_acc = accuracy_score(np.argmax(valid_y1, axis = 1), valid_pred1)
    fold_model2_acc = accuracy_score(np.argmax(valid_y1, axis = 1), valid_pred2)
    fold_model3_acc = accuracy_score(np.argmax(valid_y1, axis = 1), valid_pred3)
	
    # 앙상블 정확도 산출
    ensem_acc = accuracy_score(np.argmax(valid_y1, axis = 1), ensem_pred)

    print(f'{i + 1} model1 ACC = {fold_model1_acc:.4f}')
    print(f'{i + 1} model2 ACC = {fold_model2_acc:.4f}')
    print(f'{i + 1} model3 ACC = {fold_model3_acc:.4f}')
    print(f'{i + 1} ensemble ACC = {ensem_acc:.4f}\n')

    acc1.append(fold_model1_acc)
    acc2.append(fold_model2_acc)
    acc3.append(fold_model3_acc)

    fold_pred = best1.predict(test_x1) + best2.predict(test_x1)+ best3.predict(test_x1)
    
    # Fold별 test 데이터에 대한 예측값 생성 및 앙상블
    total_div = 5 * skf.n_splits
    fold_pred /= total_div

    pred += fold_pred

print(np.mean(acc1))
print(np.mean(acc2))
print(np.mean(acc3))
print(np.mean(acc1 + acc2 + acc3))

마지막으로 CNN 모델 3개를 앙상블(Ensemble)하여 일반화 성능을 높였다.

💡 CNN

여기서 사용된 합성곱 신경망(CNN) 은 딥러닝의 대표적인 방법으로 주로 이미지 인식에 많이 사용된다. 일반적인 인공신경망 모델의 경우에는 1차원의 입력 형태로 한정된다. 따라서 3차원의 사진 데이터를 일반적인 인공신경망으로 학습시키려면 1차원으로 평면화해야 하므로 이미지 공간 정보 유실 문제가 발생한다. 이는 정보 손실 문제로 인해 올바른 학습을 불가능하게 만든다. 이러한 단점을 보완하고, 이미지 공간 정보를 유지한 상태로 학습이 가능하게 만든 모델이 바로 CNN이다.

CNN은 위와 같이 이미지 특징을 추출하는 부분과 클래스를 분류하는 부분으로 나뉜다. 특징 추출 영역은 합성곱 층(Convolutional layer)과 풀링 층(Pooling layer)을 쌓는 형태로 구성된다.

Convolutional Layer

• Filter
  합성곱 층은 필터(커널)를 이용하여 이미지의 특징들을 추출해내고, 이를 바탕으로 특성 맵(feature map)을 구성한다.
  

• stride
  필터의 이동량(step size)를 의미

• bias
  활성화 함수를 거쳐 최종적으로 출력되는 값을 조절하는 역할로, CNN에서는 커널들을 계산한 특성 맵의 모든 원소에 더해줌으로써 편향을 계산한다. 하지만 CNN에서는 주로 bias와 같은 역할을 하는 term을 포함하는 batch norm을 이용함으로써 bias의 역할을 대신한다. 위에서 정의한 모델 함수에서도 batch norm을 이용하였다.

• Padding
  입력 이미지에 합성곱을 수행하면 출력 이미지의 크기는 입력 이미지의 크기보다 작아지게 되고, 결국 가장자리 픽셀들의 정보 손실이 일어난다. 패딩(Padding)은 이러한 문제점을 보완하기 위한 방법으로, 입력 이미지의 가장자리에 특정 값으로 설정된 픽셀들을 추가함으로써 입력 이미지와 출력 이미지의 크기를 같거나 비슷하게 만드는 역할을 수행한다. CNN에서는 주로 가장자리에 0의 값을 갖는 픽셀을 추가하는 zero-padding이 이용된다.
  

Pooling Layer

이미지의 크기를 유지한 채로 클래스를 분류하는 파트로 넘어가게 되면 연산량이 기하급수적으로 늘어나므로, 크기를 줄이면서 특정 feature를 강조할 수 있게 만든다. 풀링 레이어는 범위 내의 픽셀 중 대표값을 추출하는 방식으로 특징을 추출한다. 대표적으로 다음 세가지 방법이 있고, 가장 많이 사용되는 방법은 Max Pooling이다.

• Max Pooling
• Average Pooling
• Min Pooling

Fully Connected Layer

이미지를 분류하는 인공신경망

• Flatten layer: 데이터 타입을 Fully connected 네트워크 형태로 변경
• Softmax layer: 분류(classification) 수행