실험 3. 이전 종가의 등락폭을 가지고 미래 종가의 등락폭 예측하기 (2) 실험 코드

이번 포스팅에서는 지난 포스팅에서 설계한 실험을 수행하기 위한 코드를 소개한다.

사실 실험 1의 코드와 매우 유사하기에 코드를 작성하고 구동하는데 매우 적은 시간이 소요되었다.

그러면 곧바로 코드 소개로 넘어가자.

 

먼저, 필요한 모듈을 임포트한다. 이전 코드와 다르게 MAE를 임포트하였고, 회귀모델만 임포트하였다. 

import os
import pandas as pd
import numpy as np
from sklearn.metrics import mean_absolute_error as MAE
from sklearn.linear_model import LinearRegression as LR
import warnings

그리고 워닝을 무시하고, 데이터가 있는 경로로 설정한다.

# 주가 데이터가 있는 곳으로 경로 설정
os.chdir(r"C:\Users\Gilseung\Desktop\Jupyter\GILLAB\QUANT_DATA\일별주가")
warnings.filterwarnings("ignore")

이전과 마찬가지로 학습 데이터를 생성하는 함수를 작성한다.

보면 diff 함수를 사용하여 정의한 등락률을 사용했다는 점을 제외하면, 이전 실험 코드와 거의 같다.

def make_train_data(data, L):
    data['등락률'] = data['종가'].diff() / data['종가'] * 100
    train_data = df.copy()[['등락률']] # df를 copy한 뒤 종가 컬럼만 가져옴 (데이터프레임 형태로 가져오기 위해 괄호를 두개사용)

    # L 이전의 종가 데이터를 부착 후, shift하면서 발생하는 결측 제거 
    for l in range(1, L+1):
        train_data["{}_이전등락률".format(l)] = train_data['등락률'].shift(l)
    train_data.dropna(inplace = True)
    return train_data

이제 lag가 주어졌을 때 모델의 성능을 평가하는 함수를 작성한다. 역시 등락률을 사용했다는 점외엔 특별한게 없다 (이쯤되면 실험코드 1을 보고 오는게 낫겠다...)

def scoring_model(train_data, model_func):
    total_score = 0 # 점수 초기화

    # train data를 10등분하는데 사용 (맨 앞 인덱스는 0이므로 무시)
    train_fold_idx = np.linspace(0, len(train_data), 11).astype(int)

    for end_idx in train_fold_idx[1:-1]:
        # 데이터 분할
        train_df = train_data.iloc[:end_idx]
        test_df = train_data.iloc[end_idx:]
        
        train_X = train_df.drop(['등락률'], axis = 1)
        train_Y = train_df['등락률']
        test_X = test_df.drop(['등락률'], axis = 1)
        test_Y = test_df['등락률']
        
        # 모델 학습 및 평가
        model = model_func().fit(train_X, train_Y)
        pred_Y = model.predict(test_X)
        score = MAE(test_Y, pred_Y)
        total_score += score
    
    return total_score / 9

메인 코드에서 KOSPI 폴더에 있는 모든 데이터에 대해 모델을 학습하고 평가한다.

학습이 빠른 회귀 모델만 사용하다보니, 실험 1에 비해 매우 빠르게 실험이 종료되었다.  

다음 포스팅에서는 이 코드의 실험 결과를 정리하고 백테스트를 설계하도록 하자!

KOSPI_result = []

for data_name in os.listdir("KOSPI"):
    print(data_name)
    df = pd.read_csv("KOSPI/{}".format(data_name), encoding = "cp949").sort_values("날짜")
    if len(df) < 1000: # 길이가 1000 이상일때만 계속 수행
        continue
    best_score = 9999999 # 최고 점수 초기화
    for L in range(1, 11):
        train_data = make_train_data(df, L)
        model_score = scoring_model(train_data, LR)
        if model_score < best_score:
            best_score = model_score
            best_L = L

    KOSPI_result.append([data_name.split('.csv')[0], best_score, best_L])    

KOSPI_result = pd.DataFrame(KOSPI_result, columns = ["데이터", "평균MAE", "lag"])
KOSPI_result.to_csv("실험3_이전종가등락폭을가지고미래종가등락폭예측하기_KOSPI결과.csv", index = False, encoding = "cp949")

전체 실험 코드는 아래에서 받을 수 있다.

실험 3. 이전 종가의 등락폭을 가지고 미래 종가의 등락폭 예측하기.ipynb

0.10MB

 

수집하고 싶은 금융 데이터나 실험하고 싶은 퀀트 관련 아이디어가 있으면 댓글로 남겨주세요! 
관련 포스팅을 준비하도록 하겠습니다!