주가 예측 (1) - xgboost reg:squarederror, python, lag data

xgboost를 이용한 회귀 모델을 만들어 보았습니다. 여기에서는 단순히 다음날의 주가를 예측 하기 위한 내용은 아닙니다. 

간단하지만 앞의 몇일의 주가 평균을 구해서 해당 시점에 주가를 사도 되는지 정보를 얻을 수 있도록 구현해보았습니다. 이것은 단순한 예제일 뿐입니다. 실제 투자는 적절치 않습니다.

DB 준비

주식 데이터를 준비해야합니다. 이 부분까지 설명하면 내용이 많아져서 이 부분은 빼도록 하겠습니다. 다만 구조는 아래와 같습니다.

table 이름은 '주식명_DAY' 입니다. 여기에서는 005930 으로 하였습니다.

필드이름은 Time, Price, Vol, SPrice, HPrice, LPrice 입니다.

각각의 의미는 날짜, 현재가, 거래량, 시작가, 고가, 저가 입니다.

feature engineering

predict_feature_reg.py

import config
import platform
import sqlite3
import pandas as pd

CSV_PREDICT_FNAME = "predict.csv"
CSV_TEST_FNAME = "test.csv"
CSV_TRAIN_FNAME = "train.csv"

print(platform.architecture())

if platform.architecture()[0]!='64bit' :
	print("Please use 64bit python")
	exit(-1)


"""
DB
날짜(N-0) 가격(N-0) 거래량(N-0)
날짜(N-1) 가격(N-1) 거래량(N-1)
날짜(N-2) 가격(N-2) 거래량(N-2)
날짜(N-3) 가격(N-3) 거래량(N-3)
날짜(N-4) 가격(N-4) 거래량(N-4)
날짜(N-5) 가격(N-5) 거래량(N-5)
...

갯수는 X_M_PREDAYES : 예측할때 앞의 몇일을 가지고 학습 데이터를 만들지 결정하는 값
price(N-0),vol(N-0),price(N-1),vol(N-1),price(N-2),vol(N-2),price(N-3),vol(N-3)...N-M
가격(N-0) 거래량(N-0)|가격(N-1) 거래량(N-1)|가격(N-2) 거래량(N-2)|가격(N-3) 거래량(N-3)...N-M

갯수는 Y_O_POSTDAYES : 예측할때 뒤의 몇일을 가지고 결과를 만들지 결정하는 값
Y(N) : price(N+1),price(N+1),price(N+2),price(N+3)...N+O  이값중 price(N) 보다 목표% 보다 커지면 Y(N):1(구매해야함) 이 된다.

"""
def make_input_data(stockname,type):
	query = cur.execute("SELECT * FROM "+GetDBRealTableName(stockname,type)+" ORDER BY Time DESC")
	cols = [column[0] for column in query.description]
	df = pd.DataFrame.from_records(data=query.fetchall(), columns=cols)
	return df

def GetDBRealTableName(stockname,type):
	if type=="MIN" :
		return '"'+stockname+"_MIN"+'"'
	if type=="DAY" :
		return '"'+stockname+"_DAY"+'"'
	return ""

def date_to_int(datestr):
	return int(datestr)

def make_lag_df(train_df,laglabel,step):
	train_df[laglabel+"_lag"+str(step)]=train_df[laglabel].shift(step*-1)
	return train_df

def join_with_prev(df,prev_df,how):
	df = df.merge(prev_df,how=how)
	return df

# 날짜 날짜-1 날짜-2 가격 ...
#갯수는 Y_O_POSTDAYES : 예측할때 뒤의 몇일을 가지고 결과를 만들지 결정하는 값
#Y(N) : price(N+1),price(N+1),price(N+2),price(N+3)...N+O  이값중 price(N) 보다 목표% 보다 커지면 Y(N):1(구매해야함) 이 된다.
#  O = 2
#  price(N+2),price(N+1),price(0),price(N-1),price(N-2),price(N-3)...
#  price(0),price(N-1),price(N-2),price(N-3),price(N-4),price(N-5)...
#      2        1
def resultx(x):
	pprice = x[4*config.Y_O_POSTDAYES+0]+x[4*config.Y_O_POSTDAYES+0] * config.P_SELLP/100.0# Price
	sum=0
	for i in range(config.Y_O_POSTDAYES-1,1-1-1,-1):
		sum = sum + x[4*i+0] #0 Price_lag
	return sum/config.Y_O_POSTDAYES


if __name__ == '__main__':
	
	con = sqlite3.connect(config.db_name)
	cur = con.cursor()
	for stockname in config.predict_stock_list_for_db :
		df = make_input_data(stockname,"DAY")
		df_base = df #df[['Time','Price','Vol']]
		#df_base["int_time"] = df_base["Time"].map(date_to_int).astype(int)
		
		lag_want_list = df.columns.tolist()
		lag_want_list.remove("Time")
		for step in range(0,config.X_M_PREDAYES+config.Y_O_POSTDAYES+1):
			for lagcolname in lag_want_list:
				df_base = make_lag_df(df_base,lagcolname, step)
			
		print(df_base)
		check_col = []
		for i in range(0,config.X_M_PREDAYES+config.Y_O_POSTDAYES+1):
			check_col.append("Price_lag"+str(i)) #0
			check_col.append("SPrice_lag"+str(i))#1
			check_col.append("HPrice_lag"+str(i))#2
			check_col.append("LPrice_lag"+str(i))#3
		df_base['Y'] = df_base[check_col].apply(resultx, axis=1)
		df_base['Y'] = df_base['Y'].shift(config.Y_O_POSTDAYES)
		
		#df_base = df_base.drop(['Time','Price','Vol','SPrice','HPrice','LPrice'],axis=1)
		
		#예측데이터
		predict = df_base.head(config.Y_O_POSTDAYES).drop(['Y'],axis=1)
		predict.to_csv(CSV_PREDICT_FNAME, encoding='utf-8', index=False)
		df_base.drop(df_base.head(config.Y_O_POSTDAYES).index, inplace=True)
		#검증데이터
		test = df_base.head(config.COUNT_TEST_DATA)
		test.to_csv(CSV_TEST_FNAME, encoding='utf-8', index=False)
		df_base.drop(df_base.head(config.COUNT_TEST_DATA).index, inplace=True)
		#훈련데이터
		df_base.to_csv(CSV_TRAIN_FNAME, encoding='utf-8', index=False)
	con.close()

config.py

# db이름
db_name = "data/stock.db"

# predict ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
predict_stock_list_for_db=[
"005930"
]

X_M_PREDAYES=5 # 예측할때 과거 몇일을 가지고 학습 데이터를 만들지 결정하는 값
#price(N-0),vol(N-0),price(N-1),vol(N-1),price(N-2),vol(N-2),price(N-3),vol(N-3)...N-M
#가격(N-0) 거래량(N-0)|가격(N-1) 거래량(N-1)|가격(N-2) 거래량(N-2)|가격(N-3) 거래량(N-3)...N-M

Y_O_POSTDAYES=5# 예측할때 앞으로의 몇일을 가지고 결과Y를 만들지 결정하는 값
#Y(N) : price(N+1),price(N+1),price(N+2),price(N+3)...N+O  이값중 price(N) 보다 목표% 보다 커지면 Y(N):1(구매해야함) 이 된다.

#이익이 났을때 판매 가격 %
P_SELLP=3
#손해가 났을태 판매 가격 %
M_SELLP=3

COUNT_TEST_DATA = 600 # 테스트 검증를 위한 데이터 갯수

예측을 하기 위한 데이터가 3가지 종류가 있습니다.

 train, test, predict 인데 데이터 중에 train은 훈련을 하기 위한 데이터이고 내부적으로 다시 나뉘어서 검정도 하게됩니다. test data는 실제 여러가지 테스트를 하기위한 데이터로 COUNT_TEST_DATA = 600 갯수만큼 따로 준비하였습니다. predict는 Y값이 없는 데이터입니다.  predict의 값은 

'Y_O_POSTDAYES=5# 예측할때 앞으로의 몇일을 가지고 결과Y를 만들지 결정하는 값'

위 값과 관계가 있습니다. 즉 과거의 데이터로부터 예측 데이터Y를 구성할때 미래의 Y_O_POSTDAYES 날짜만큼 보고 평균값을 예측하도록 구성하였습니다.

해당 부분은 dataframe의 apply함수를 이용하여 resultx 에서 구현되도록 되어있습니다.

미래를 예측하는것은 일반적인 모델에서는 과거 데이터를 많이 사용합니다. 그래서 컬럼중에 lag data를 만들게 되는데 해당 동작이 아래코드에 의해서 이루어 집니다.

		lag_want_list = df.columns.tolist()
		lag_want_list.remove("Time")
		for step in range(0,config.X_M_PREDAYES+config.Y_O_POSTDAYES+1):
			for lagcolname in lag_want_list:
				df_base = make_lag_df(df_base,lagcolname, step)

그 중에 Time 컬럼은 의미가 없어서 삭제하고 나머지 데이터에 대해서 모두 lag를 만들게 됩니다.

실제 데이터는 아래와 같고 Price와 Price_lag0 는 같은 값입니다.

train.csv 파일 일부

즉 20180220일 기준으로 Price는 47400이고 Price_lag0는 47400 , Price_lag1는 48380 이되는데 이값은 결국 20180219일자 Price로 부터 오게 됩니다.

Y값은 좀 더 복잡합니다. 아래 부분에 의해서 결정되는데 config.Y_O_POSTDAYES 값이 2라면 현재 기준으로 2일간의 lag_0 lag_1 두개의 값을 보고 평균값을 취해서 Y 로 기록하게 됩니다.

이렇게 되면 lag data는 과거 데이터이므로 즉 해당 Y는 2일전 데이터로본 Y값이므로 이 값은 shift하여 적절한 미래값을 예측하도록 선정합니다.

		check_col = []
		for i in range(0,config.X_M_PREDAYES+config.Y_O_POSTDAYES+1):
			check_col.append("Price_lag"+str(i)) #0
			check_col.append("SPrice_lag"+str(i))#1
			check_col.append("HPrice_lag"+str(i))#2
			check_col.append("LPrice_lag"+str(i))#3
		df_base['Y'] = df_base[check_col].apply(resultx, axis=1)
		df_base['Y'] = df_base['Y'].shift(config.Y_O_POSTDAYES)

이예제에서는 config.Y_O_POSTDAYES 값이 5이므로 아래그림에서  빨간색 Y 값은 미래의 Price 5일치의 평균값을 가지도록 구현이 되어있습니다.

train.csv 파일 일부

predict_feature_reg.py 파일 실행시키면 아래 3개의 파일이 생성됩니다.

CSV_PREDICT_FNAME = "predict.csv"

CSV_TEST_FNAME = "test.csv"

CSV_TRAIN_FNAME = "train.csv"

Predict

이제 본격적으로 예측을 할 차례입니다.

predict_xgboost_reg.py

#predict_xgboost
STD_ALL = False
NOR_ALL = False
PREFIX = __file__[:-3]
BASE_PATH = "./"

SELECT_NEED = False
Y1_SELECT_LEVEL = 0
Y2_SELECT_LEVEL = 0

ADD_TEST_DATA = False
USE_TEST_DATA = True

param={
 'objective':'reg:squarederror',
 'eta':0.9
}

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
import time
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split 
from sklearn import preprocessing
import xgboost as xgb
from sklearn.metrics import mean_squared_error
import pickle


class writeLog():
 def write(self, fileName, text):
  print(text)
  f=open(fileName,'a')
  f.write(text)
  f.write("\n")
  f.close()
 def writeWithoutCR(self, fileName, text):
  f=open(fileName,'a')
  f.write(text)
  f.close()

log = writeLog()
log.write(PREFIX+"log.txt","start")

np.random.seed(1000)

train_df = pd.read_csv(BASE_PATH+'train.csv', header=0, encoding='utf8')
if ADD_TEST_DATA==True:
	test_df = pd.read_csv(BASE_PATH+'test.csv', header=0, encoding='utf8')
predict_df = pd.read_csv(BASE_PATH+'predict.csv', header=0, encoding='utf8')


# Y 없음
# TRAIN data
# TRAIN data
# TRAIN data
# TRAIN data
# concat 
# TEST data
# TEST data
# TEST data

if ADD_TEST_DATA==True:
	allX = pd.concat([train_df, test_df], axis=0)
	allX.reset_index(drop=True,inplace=True)
else:
	allX = train_df

#allX = pd.get_dummies(allX)
#allX = allX.fillna(value=0)

#drop Y
Y1train = allX.pop('Y')

if NOR_ALL == True:
 names = allX.columns
 scaler = preprocessing.MinMaxScaler()
 allX = scaler.fit_transform(allX)
 allX = pd.DataFrame(allX, columns=names)

if STD_ALL == True:
 names = allX.columns
 scaler = preprocessing.StandardScaler()
 allX = scaler.fit_transform(allX)
 allX = pd.DataFrame(allX, columns=names)

#분리
trainX = allX.copy() #allX[0:int(train_size)]
trainX = trainX.drop(['Time','Price','Vol','SPrice','HPrice','LPrice'],axis=1)

predictX = predict_df #allX[int(train_size):int(allX.shape[0])]
predictX = predictX.drop(['Time','Price','Vol','SPrice','HPrice','LPrice'],axis=1)

del (allX)

def OutputData(filename,Y1):
	test_csv = pd.read_csv(BASE_PATH+'predict.csv', header=0, encoding='utf8')
	#test_csv.pop('Y')
	predAll = pd.concat([test_csv,pd.DataFrame(Y1, columns=['PY'])], axis=1)
	predAll.to_csv(path_or_buf=PREFIX+filename, index=False)
	del predAll

def OutputData2(filename,y_v,p_y_v):
	df = pd.DataFrame(y_v, columns=['Y'])
	df = pd.concat([df,pd.DataFrame(p_y_v,columns=['PY'])], axis=1)
	df = pd.concat([df,pd.DataFrame(y_v-p_y_v, columns=['Y-PY'])], axis=1)
	df.to_csv(path_or_buf=PREFIX+filename, index=False)

def Train(trainX, trainY, predictX, Ystr):
	features = trainX.columns.tolist()

	trainXnp = trainX.values
	trainYnp = trainY.values
	
	X_train, X_validate, y_train, y_validate = train_test_split(trainXnp, trainYnp, test_size=0.33)

	dtrn = xgb.DMatrix(X_train, label=y_train)
	dvld = xgb.DMatrix(X_validate, label=y_validate)

	watch_list = [(dvld,'eval'),(dtrn,'train')]
	model = xgb.train(param, dtrn, num_boost_round=10000, evals=watch_list, early_stopping_rounds=20)
	y_pred = model.predict(dvld, ntree_limit=model.best_ntree_limit)
	rmse = mean_squared_error(y_validate, y_pred)
	
	OutputData2("Result_validate.csv",y_validate,y_pred)

	"""
	log.write(PREFIX+"log.txt","rmse %f"%(rmse))
	log.write(PREFIX+"log.txt","%s Feature importance:split"%(Ystr))
	for kv in sorted([(k,v) for k, v in model.get_fscore().items()], key=lambda kv: kv [1], reverse=True):
		log.writeWithoutCR(PREFIX+"log.txt",str(kv))
		log.writeWithoutCR(PREFIX+"log.txt",",\n")
	log.writeWithoutCR(PREFIX+"log.txt",",\n\n")
	"""
   
	predictY = model.predict(xgb.DMatrix(predictX.values), ntree_limit=model.best_ntree_limit)
	
	if USE_TEST_DATA==True:
		test_df = pd.read_csv(BASE_PATH+'test.csv', header=0, encoding='utf8')
		test_copydf=test_df.copy()
		test_copydf = test_copydf.drop(['Time','Price','Vol','SPrice','HPrice','LPrice','Y'],axis=1)
		test_predictY = model.predict(xgb.DMatrix(test_copydf.values), ntree_limit=model.best_ntree_limit)
		test_df['PreY']=test_predictY
		test_df.to_csv(path_or_buf=PREFIX+'test_prey.csv', index=False)
	
	pickle.dump(model, open('model.sav', 'wb'))

	return (predictY, rmse)

y1_pred, rmse1 = Train(trainX, Y1train, predictX, "Y")
log.write(PREFIX+"log.txt","Total rmse %f"%(rmse1))
OutputData("Result.csv",y1_pred)

위 파일을 실행시키면 아래 내용에 의해

	if USE_TEST_DATA==True:
		test_df = pd.read_csv(BASE_PATH+'test.csv', header=0, encoding='utf8')
		test_copydf=test_df.copy()
		test_copydf = test_copydf.drop(['Time','Price','Vol','SPrice','HPrice','LPrice','Y'],axis=1)
		test_predictY = model.predict(xgb.DMatrix(test_copydf.values), ntree_limit=model.best_ntree_limit)
		test_df['PreY']=test_predictY
		test_df.to_csv(path_or_buf=PREFIX+'test_prey.csv', index=False)

대표적으로 predict_xgboost_regtest_prev.csv 파일이 생성됩니다.

test.csv 파일과 거의 비슷하나 마지막에 예측된 PreY 컬럼이 더 추가됩니다.

Y, PreY 두개를 비교해보면 적절히 예측했는지 알 수 있습니다.

그렇지만 위 데이터에서 어느 시점에 사야 하느냐? 라는 질문을 던질 수 있습니다. 이러한 문제를 해결하기위해서 사야하는 시점을 계산해보겠습니다.

predict_test_reg.py

import pandas as pd
import config

BASE_PATH = "./"

def resultx(x):
	pprice = x[0]+x[0] * config.P_SELLP/100.0# Price
	if pprice <= x[2] : return 1
	return 0


test_df = pd.read_csv(BASE_PATH+'predict_xgboost_regtest_prey.csv', header=0, encoding='utf8')

test_df['Buy'] = test_df[['Price','Y','PreY']].apply(resultx, axis=1)

test_df.to_csv("buy.csv", encoding='utf-8', index=False)

'predict_xgboost_regtest_prey.csv' 파일을 읽어들여 

#이익이 났을때 판매 가격 %
P_SELLP=3

PreY 예측 평균가가 목표값보다 높으면 Buy에 1을 저장하도록 합니다.

buy.csv 파일에 결과를 추가합니다.