[금융 AI] 금융 투자 영역에서의 AI(3) - 머신러닝을 이용한 투자 전략
금융 시계열 데이터에 대한 교차 검증 방법
금융 시계열 데이터는 시간에 따라 연속적인 특성을 가지고 있기 때문에, 전통적인 k-fold 교차 검증 방법은 적합하지 않다. k-fold 교차 검증은 데이터의 순서를 무시하고 무작위로 분할하는데, 이는 시계열 데이터의 시간적 순서를 무시하는 것으로, 미래 데이터가 모델 학습에 포함될 위험이 있다. 금융 시장에서는 미래 데이터를 알 수 없기 때문에, 이러한 방식은 비현실적이다.
금융 시계열 데이터에는 walk-forward 교차 검증과 blocking walk-forward 교차 검증 방법이 적합하다. 이 방법들은 데이터를 시간 순서대로 분할하고, 각 단계에서 모델을 학습시킨 후 바로 다음 시점의 데이터를 사용해 모델을 검증한다. 이는 시간의 순서를 유지하면서 모델이 미래 데이터를 예측하는 능력을 평가할 수 있다. 또한 TimeSeriesSplit 라이브러리를 사용하면 시계열 데이터에 대한 교차 검증을 수행할 수 있다. 이 라이브러리는 sklearn의 일부로, 학습 데이터를 점진적으로 확장하면서 검증 데이터셋을 생성해 모델의 성능을 평가한다. 이 방식은 금융 시계열 데이터의 시간적 연속성을 존중하며, 모델이 시간에 따라 변화하는 패턴을 얼마나 잘 포착하는지를 평가하는데 유용하다.
엠바고와 퍼징
엠바고와 퍼징은 금융 시계열 데이터를 다룰 때 데이터 누설을 방지하는 필수 기법이다. 특히 정보가 빠르게 확산되는 고빈도 거래 시장에서 이 기법들의 중요성은 더욱 강조된다.
엠바고 방법은 퍼징 방법의 확장판으로 볼 수 있다. 이 방법에서는 테스트 세트 뒤에 더욱 넓은 마진을 추가한다. 이 '엠바고 기간' 동안의 데이터는 훈련이나 테스트에 전혀 사용되지 않는다. 특히 정보가 빠르게 퍼지는 시장에서, 이 기간 동안 발생할 수 있는 모든 정보 누설을 방지함으로써, 훈련 세트가 테스트 세트에 영향을 미치는 것을 막는다.
퍼징 방법은 테스트 세트와 훈련 세트 사이에 일정한 마진을 설정해 서로 간의 정보 누설을 방지한다. 예를 들어 어느 특정 기간의 데이터를 훈련 세트로 사용할 때, 바로 그 이후 기간의 데이터는 테스트 세트로 사용하지 않고 그 다음 기간의 데이터를 사용함으로써, 훈련 데이터가 테스트 데이터에 영향을 주는 것을 차단한다.
mlfinlab 라이브러리는 퍼징과 엠바고 기법을 쉽게 구현할 수 있도록 돕는 파이썬 라이브러리다. 이 라이브러리는 금융 머신러닝을 위한 다양한 도구와 알고리즘을 제공하며, 고빈도 거래 데이터를 안전하게 다루기 위한 기능을 포함하고 있다. mlfinlab을 활용하면 훈련 세트와 테스트 세트를 안전하게 분리할 수 있고, 데이터 누설을 방지하여 모델이 실제 시장 상황에서 더욱 정확한 예측을 하도록 돕는다.
잡음을 줄이는 데이터 처리 방법
금융 데이터에서 잡음을 줄이는 것은 중요하며, 이를 위해 다양한 기법이 사용된다. 각 기법은 데이터의 특성과 사용자의 목적에 따라 다르게 적용될 수 있다.
- 이동평균 : 주가의 단기적인 변동을 완화시키기 위해 널리 사용된다. 일정 기간 동안의 평균을 계산하고, 이를 일정한 간격으로 그래프에 표시함으로써 노이즈와 격차를 최소화하고, 주가의 전반적인 트렌드를 더 명확히 파악할 수 있다. 이는 단순이동평균과 지수이동평균으로 나뉘는데, 지수이동평균은 최근 데이터에 더 많은 가중치를 부여하여 변화에 민감하게 반응한다.
- 웨이블릿 변환 : 시간-주파수 도메인에서 노이즈를 제거하는 복잡한 방법이다. 이는 주가 데이터의 고주파 노이즈를 제거하고, 기본적인 추세와 패턴을 더 명확하게 드러낼 수 있게 도와준다. 웨이블릿 변환은 로컬라이즈된 데이터의 특성을 보존하는 능력을 가지고 있으므로, 잡음이 많은 금융 데이터에 특히 유용하다.
- 주성분 분석 : 데이터의 차원을 축소하고 주요 패턴을 추출하는데 사용된다. 이 기법은 노이즈를 제거하고 데이터의 주요 특성을 보존하는데 효과적이다. 주성분 분석은 데이터셋 내 변수들 간의 상관관계를 분석하여, 가장 중요한 정보를 나타내는 새로운 변수들을 생성한다.
- 오토인코더 : 딥러닝 기법 중 하나로, 데이터의 중요한 특징을 학습하고 노이즈를 제거하는데 사용된다. 이는 원본 데이터를 압축한 후 다시 복원하는 과정을 통해 데이터셋에서 중요한 정보만을 추출하고 불필요한 부분을 제거한다.
ETFs를 활용한 주가 방향 예측 모델 개발
이제는 상징지수펀드(ETFs) 데이터와 거시경제 지표 데이터를 활용하여 트리 기반 머신러닝 알고리즘을 이용한 투자 전략을 구현해보고자 한다. 단순히 알고리즘을 적용한다고 해서 항상 우수한 결과를 얻을 수 있는 것은 아니고, 실습을 통해 금융데이터에 머신러닝 알고리즘이 적용되는 사례를 살펴보는 목적임을 참고하자.
1. 필요 라이브러리 호출
import numpy as np
import pandas as pd
import os
for dirname, _, filenames in os.walk('/kaggle/input'):
for filename in filenames:
print(os.path.join(dirname, filename))
import warnings
import glob
import os
import datetime
import matplotlib.pyplot as plt
import pandas as pd
import numpy as np
from sklearn import preprocessing
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from sklearn.model_selection import cross_validate
from sklearn.model_selection import TimeSeriesSplit
from sklearn.model_selection import GridSearchCV
from sklearn.linear_model import LinearRegression
from sklearn.ensemble import RandomForestClassifier
from xgboost import XGBClassifier
from xgboost import plot_importance
from sklearn.metrics import f1_score
from sklearn.metrics import mean_squared_error, r2_score
from sklearn.metrics import accuracy_score
from sklearn.metrics import confusion_matrix, accuracy_score, precision_score, recall_score, f1_score, roc_auc_score
from sklearn import svm
import seaborn as sns; sns.set()
2. 데이터 불러오기, 기술적 지표 생성
df = pd.read_csv('/kaggle/input/etfs-main/ETFs_main.csv')
# 기술적 지표 만들기
def moving_average(df, n):
MA = pd.Series(df['CLOSE_SPY'].rolling(n, min_periods=n).mean(), name='MA_' + str(n))
df = df.join(MA)
return df
def volume_moving_average(df, n):
VMA = pd.Series(df['VOLUME'].rolling(n, min_periods=n).mean(), name='VMA_' + str(n))
df = df.join(VMA)
return df
def relative_strength_index(df, n):
delta = df['CLOSE_SPY'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=n).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=n).mean()
RS = gain / loss
RSI = 100 - (100 / (1 + RS))
RSI.name = 'RSI_' + str(n)
df = df.join(RSI)
return df
# 기술 지표 적용
df = moving_average(df, 45)
df = volume_moving_average(df, 45)
df = relative_strength_index(df, 14)
# 'Dates' 열을 인덱스로 설정
df = df.set_index('Dates')
df = df.dropna()
print(len(df))
금융 데이터를 불러오고 이번 실습에서는 이동평균, 거래량 이동평균, 시장 강도 지수라는 세 가지 기술적 지표를 사용할 것이므로 세 가지의 함수를 생성한다.
앞선 포스팅에서 말한 TA-Lib 라이브러리를 활용하면 더욱 간결하게 기술적 지표를 계산할 수 있다.
import talib
df['MA_' + str(n)] = talib.SMA(df['CLOSE_SPY'], timeperiod=n)
df['VMA_' + str(n)] = talib.SMA(df['VOLUME'], timeperiod=n)
df['RSI_' + str(n)] = talib.RSI(df['CLOSE_SPY'], timeperiod=n)
3. 타겟 변수 생성 및 변수 전처리
# 타겟 변수 생성 (pct_change)
df['pct_change'] = df['CLOSE_SPY'].pct_change()
# 모델링을 위한 이진 분류 값 생성
df['target'] = np.where(df['pct_change'] > 0, 1, 0)
df = df.dropna(subset=['target']) # 결측값 제거
# 정수형 변환
df['target'] = df['target'].astype(np.int64)
print(df['target'].value_counts())
# 다음날 예측을 위해 타겟 변수를 shift
df['target'] = df['target'].shift(-1)
df = df.dropna()
print(len(df))
4. 변수 처리 및 데이터셋 분할
# 설명 변수와 타겟 변수 분리
y_var = df['target']
x_var = df.drop(['target', 'OPEN', 'HIGH', 'LOW', 'VOLUME', 'CLOSE_SPY', 'pct_change'], axis=1)
# 상승과 하락 비율 확인
up = df[df['target'] == 1].target.count()
total = df.target.count()
print('up/down ratio: {0:.2f}'.format(up / total))
up/down ratio: 0.54
# 훈련셋과 테스트셋 분할
X_train, X_test, y_train, y_test = train_test_split(x_var, y_var, test_size=0.3, shuffle=False, random_state=3)
# 훈련셋과 테스트셋의 양성 샘플 비율 확인
train_count = y_train.count()
test_count = y_test.count()
print('train set label ratio')
print(y_train.value_counts() / train_count)
print('test set label ratio')
print(y_test.value_counts() / test_count)
5. 모델 학습 및 성능 평가
#혼동 행렬 및 성능 평가 함수
def get_confusion_matrix(y_test, pred):
confusion = confusion_matrix(y_test, pred)
accuracy = accuracy_score(y_test, pred)
precision = precision_score(y_test, pred)
recall = recall_score(y_test, pred)
f1 = f1_score(y_test, pred)
roc_score = roc_auc_score(y_test, pred)
print('confusion matrix')
print(confusion)
print('accuracy: {0:.4f}, precision: {1:.4f}, recall: {2:.4f}, F1: {3:.4f}, ROC AUC score: {4:.4f}'.format(
accuracy, precision, recall, f1, roc_score))
#모델 학습 및 평가
# XGBoost 모델 학습 및 예측
xgb_dis = XGBClassifier(n_estimators=400, learning_rate=0.1, max_depth=3)
xgb_dis.fit(X_train, y_train)
xgb_pred = xgb_dis.predict(X_test)
# 훈련 정확도 확인
print(xgb_dis.score(X_train, y_train))
# 성능 평가
get_confusion_matrix(y_test, xgb_pred)
XGBoost 분류기 모델을 활용해 모델을 학습하고, 정확도, ROC AUC 점수 등 분류 모델의 성능 평가 지표를 계산하는 함수를 설정하여 성능을 계산한다. 위의 코드를 통한 성능 평가 결과 훈련 데이터에서 84.7%의 정확도를 보여주었다. 이 모델은 훈련 데이터로 내일 주식의 등락을 84.7% 확률로 맞힌다는 것이다. 하지만 훈련 데이터의 정확도가 높다고 해서 항상 좋은 모델은 아니다. 우리는 실제로 내일 주가의 등락을 예측하는 것이 목표이기 때문에, 모델이 과적합되었을 가능성이 있다.
get_confusion_matrix() 함수를 활용해 테스트셋에서의 성능 지표도 계산해보았는데, 결과는 50%의 정확도가 나왔다. 이것이 바로 머신러닝 기반의 투자 전략을 만들 때 우리가 가장 신경써야 할 과적합 문제다. 만약 max_depth 파라미터를 더 크게 만들면 과적합이 더 심해져 훈련 덩확도를 100% 달성할 수도 있다. 하지만 그렇게 된다해도 테스트셋의 정확도는 여전히 낮게 머물 것이다.
과적합 문제를 해결하기 위해서는 더 많은 데이터를 수집하거나, 모델의 복잡도를 줄이는 방법을 고려해야 한다. 그렇다면 이번에는 랜덤 포레스트 알고리즘을 사용하는데, 임의의 파라미터를 지정하는 것이 아니라 지난 포스팅들에서 자주 등장한 GridSearchCV() 함수를 사용해보자.
# 랜덤 포레스트 매개변수 설정
n_estimators = range(10, 200, 10)
params = {
'bootstrap': [True],
'n_estimators': n_estimators,
'max_depth': [4, 6, 8, 10, 12],
'min_samples_leaf': [2, 3, 4, 5],
'min_samples_split': [2, 4, 6, 8, 10],
'max_features': [4]
}
# 교차 검증 설정
my_cv = TimeSeriesSplit(n_splits=5).split(X_train)
# GridSearchCV를 사용한 모델 학습
clf = GridSearchCV(RandomForestClassifier(), params, cv=my_cv, n_jobs=-1)
clf.fit(X_train, y_train)
# 최적의 매개변수와 정확도 출력
print('best parameter:\n', clf.best_params_)
print('best prediction: {0:.4f}'.format(clf.best_score_))
# 테스트셋에서의 성능 확인
pred_con = clf.predict(X_test)
accuracy_con = accuracy_score(y_test, pred_con)
print('accuracy: {0:.4f}'.format(accuracy_con))
get_confusion_matrix(y_test, pred_con)
# 타겟 변수 정의 변경 (0.0005% 이상의 수익률)
df['target'] = np.where(df['pct_change'] > 0.0005, 1, -1)
df['target'].value_counts()
# 타겟 변수를 한 행 앞으로 이동
df['target'] = df['target'].shift(-1)
df = df.dropna()
# 타겟 변수를 1과 0으로 변환
df['target'] = df['target'].replace(-1, 0)
df['target'].value_counts() # 변환된 결과 확인
# 설명 변수와 타겟 변수 분리
y_var = df['target']
x_var = df.drop(['target', 'OPEN', 'HIGH', 'LOW', 'VOLUME', 'CLOSE_SPY', 'pct_change'], axis=1)
# 훈련셋과 테스트셋 분할
X_train, X_test, y_train, y_test = train_test_split(x_var, y_var, test_size=0.3, shuffle=False, random_state=3)
# 랜덤 포레스트 매개변수 설정
n_estimators = range(10, 200, 10)
params = {
'bootstrap': [True],
'n_estimators': n_estimators,
'max_depth': [4, 6, 8, 10, 12],
'min_samples_leaf': [2, 3, 4, 5],
'min_samples_split': [2, 4, 6, 8, 10],
'max_features': [4]
}
# 교차 검증 설정
my_cv = TimeSeriesSplit(n_splits=5).split(X_train)
# GridSearchCV를 사용한 모델 학습
clf = GridSearchCV(RandomForestClassifier(), params, cv=my_cv, n_jobs=-1)
clf.fit(X_train, y_train)
# 최적의 매개변수와 정확도 출력
print('best parameter:\n', clf.best_params_)
print('best prediction: {0:.4f}'.format(clf.best_score_))
# 테스트셋에서의 성능 확인
pred_con = clf.predict(X_test)
accuracy_con = accuracy_score(y_test, pred_con)
print('accuracy: {0:.4f}'.format(accuracy_con))
get_confusion_matrix(y_test, pred_con)
위의 코드를 실행한 결과, 테스트 정확도는 53.55%이고, ROC AUC 점수는 53.26%이다. 훈련셋과 테스트셋에서 주가 트렌드가 균일하기 때문에 마냥 오른다고 찍는다고 해서 높은 정확도를 얻는 것도 아닌 상황임을 고려하면, 정확도가 매우 높지는 않지만 이전 모델과 비교했을 때 더 안정적이며 과적합 문제도 보이지 않는다.
물론 이 모델을 실무에 사용하려면 여러 번의 검증을 거쳐야 한다. 예측값을 바탕으로 백테스트 수익률을 계산해보고, 해당 수익률을 기반으로 MDD나 수익률의 변동성도 살펴봐야 한다. 무엇보다 더 오랜 기간 테스트해보며 안정적인 정확도를 유지하거나 수익률을 가져올 수 있는지도 검토해야 한다. 보수적이라면 56% 이상의 정확도나 AUC-ROC 점수를 확보하는 것이 좋다.