실험 2. 종목 간 선후행 관계 파악 (2) 실험 코드

이번 포스팅에서는 이전 포스팅에서 설명한 이전 종가를 가지고 미래 종가를 예측하는 파이썬 코드를 설명한다.

실험 결과는 다음 포스팅에서 정리하도록 한다.

 

먼저, 실험에 필요한 모듈을 불러오고 워닝을 무시한다.

import os
import pandas as pd
import warnings
import itertools
from statsmodels.tsa.stattools import adfuller
from statsmodels.tsa.stattools import grangercausalitytests as GCT
os.chdir(r"C:\Users\Gilseung\Desktop\Jupyter\GILLAB\QUANT_DATA\201609~202108\주가")
warnings.filterwarnings("ignore")

그리고나서 KOSPI 폴더에 있는 모든 데이터를 all_df에 추가한다.

KOSPI 폴더에 있는 파일을 df로 불러오고, 날짜를 인덱스로 설정한 뒤, 종가만 가져오고, 종가를 종목명으로 변경한 뒤 all_df에 추가하는 방식이다. 

all_df = pd.DataFrame()

# 데이터 추가: 날짜 기준으로 종가만 합침
for file_name in os.listdir("KOSPI"):
    df = pd.read_csv("KOSPI/{}".format(file_name), encoding = "cp949")
    df.set_index("날짜", inplace = True) # 날짜 컬럼을 인덱스로 설정
    df = df[['종가']] # 종가만 남기고 나머지 변수 삭제
    df.rename({"종가":file_name.split(".csv")[0]}, axis = 1, inplace = True) # 종가를 파일명으로 변경    
    all_df = pd.concat([all_df, df], axis = 1) # 열 추가

all_df.head()

 

이제 각 구간에 대해 아래와 같은 코드로 실험을 진행한다. 

for period in [[20160901, 20210831], [20200901, 20210831]]:
    start, end = period
    period_df = all_df.loc[start:end]
    period_df.dropna(axis = 1, inplace = True) # 결측이 하나라도 있는 종목은 삭제

    # 1차 차분 수행
    period_df = period_df.diff().dropna()
      
    # adfuller 테스트 수행해서 정상성을 띈다고 판단된 컬럼만 사용
    ADF_passed_columns = []
    for col in period_df.columns:
        if adfuller(period_df[col])[1] < 0.05:
            ADF_passed_columns.append(col)
            
    result = []
    temp = [] # 계산한 컬럼 튜플을 저장해놓는 곳
    i = 0
    n = len(ADF_passed_columns) * len(ADF_passed_columns)
    for col1 in ADF_passed_columns:
        for col2 in ADF_passed_columns:
            i += 1
            print(i, "/", n)
            if (col1 == col2) or ((col1, col2) in temp) or ((col1, col2) in temp):
                pass
            else:
                # 선택 가능한 모든 컬럼 조합 선택
                temp.append((col1, col2))
                try:
                    GCT_result = GCT(period_df[[col1, col2]], maxlag = 11, verbose = False)
                    min_pvalue = 1
                    for lag in range(1, 11):
                        pvalue = GCT_result[lag][0]['ssr_chi2test'][1]
                        if pvalue < min_pvalue:
                            min_pvalue = pvalue
                            min_lag = lag

                    if min_pvalue < 0.01:
                        result.append([col2, col1, min_lag, min_pvalue])
                except:
                    pass
                
                
    result = pd.DataFrame(result, columns = ["선행종목", "후행종목", "lag", "pvalue"])
    result.to_csv("{}-{}_선후행관계분석결과.csv".format(start, end))

먼저, period는 전구간과 최신 구간을 순회하고, 각 구간에 대해 start와 end값을 설정한다.

설정한 start와 end값을 가지고 all_df에서 해당하는 구간의 데이터만 가져온 뒤, 1차 차분을 수행한다.

1차 차분을 수행한 데이터에 대해, adfuller 테스트를 진행하여 p-value가 0.05 미만인 경우에는 ADF_passed_columns라는 리스트에 추가한다. 

 

이제 ADF_passed_columns에 있는 모든 순서쌍에 대해 그랜저 테스트를 진행한다.

사용 함수는 grangercausalitytests(df, maxlag)로 df에 있는 두 번째 컬럼이 첫 번째 컬럼에 그랜져 인과인지를 (즉, 선행하는지를) 판단하는 함수이다. lag는 1부터 maxlag만큼 순회하면서 테스트가 진행된다. 이 함수의 출력은 lag가 key이고 lag별 결과가 value인 사전이고, lag별 결과는 다양한 통계 분석 결과로 구성된다. 본 실험에서는 ssr_chi2test의 p-value만 가져오려 코드를 구성했다. 그리고 lag별 결과에서 가장 작은 p-value를 min_pvalue에 저장하여 이 값이 0.01보다 작으면 result라는 데이터프레임에 추가하는 방식으로 실험을 진행한다. 

 

완성된 소스코드는 아래와 같다.

실험 2. 선후행 관계 파악.ipynb

0.18MB

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