투자지표 기반의 벡테스팅
2022.11.07 - [개발일지] - 2. 투자지표를 활용한 매매 시점 모니터링 - RSI, RMI
이전 글에서 RSI, RMI 지표를 파이썬으로 구현했다.
한 종목을 선택한 후 지표와 주가의 추이도 그래프로 확인했다.
그럼 이제 특정 기준으로 매매를 자동으로 진행했을 때 수익이 나는지, 수익금을 얼마인지 테스트를 해봐야 RSI지표가 가치 있는 지표인지 확인할 수 있기 때문에 벡테스팅을 하고자 한다.
우선 필요한 패키지를 import 하는 것으로 시작한다.
import pandas as pd
import numpy as np
import datetime
import time
import matplotlib.pyplot as plt
import FinanceDataReader as fdr
import copy
df_krx = fdr.StockListing('KRX')
kospi_data = df_krx[(df_krx["Market"]=="KOSPI") & (~df_krx["Sector"].isna())]
Kospi에 있는 종목으로 테스트를 할 예정이기 때문에 필터를 걸어 대상 데이터를 추려놓는다.
save_path = "trading_simulation/trading_simulation_test"
ks = df_krx[df_krx["Name"]=="삼성전자"]
target_stock = ks.Name.values[0]
price_data = fdr.DataReader(ks.Symbol.values[0],"2012-12-1")
price_data["stock_name"]=target_stock
RSI_df = rsi(price_data, "Close",14)
RMI_df = rmi(price_data, "Close",14)
price_data["RSI"] = RSI_df
price_data["RMI"] = RMI_df
price_data.tail()
save_path는 벡테스팅을 하는 동안 거래 기록을 저장하기 위한 경로이다.
벡테스팅은 삼성전자로 진행해보고자 한다. 데이터를 2012년 12월부터 수집한다. RSI계산을 위해 앞의 14일의 가격이 더 필요하기 때문에 12월 부터 테스트 기간을 2013년부터 2022년 현재까지로 약 10년을 기간으로 한다.
def save_txt(filename, lines, line_col):
lines.insert(0,0)
lines.append(0)
if line_col is not None:
line_col.insert(0, "Non")
line_col.append("Non")
with open(filename, 'a') as f:
if line_col is not None:
f.writelines(str(line_col))
f.write("\n")
f.writelines(str(lines))
f.write("\n")
meta_info = ["name","time","보유KRW", "보유주","price","총자산",
"TAG", "stock","money","tax","target_index"]
save_txt(save_path+".txt", meta_info, None)
log_info = ["name","time","price","target_index","buy","sell",
"current_stock","avg_price","yield","total_money"]
save_txt(save_path+"_log.txt", log_info, None)
log_info = ["time","money"]
log_info.append(target_stock)
log_info.append(target_stock+"_price")
save_txt(save_path+"_total_price_log.txt", log_info, None)
위의 코드로 거래 log와 일 별 주가 변동에 의한 현 자산의 기록을 위한 save파일의 Header를 저장한다.
header가 작 정의되어야 벡테스팅 후 시각화하는데 편리하다.
class set_init():
def __init__(self,stock_list,save_path,seed_money,set_base_buy):
self.money = seed_money
self.total_money = seed_money
self.stock_list = stock_list
self.stock_count = 0
self.now_rmi_list = 0
self.now_rsi_list = 0
self.stock_price = 0
self.my_asset_price = 0
self.v1 = 1
self.v2 = 1
self.save_path = save_path
self.set_stand_price = set_base_buy
def set_data(self,tmp,target_index):
self.buy_list = 0
self.sell_list = 0
self.stock_price = tmp.tail(1).Close.values[0]
self.now_index_list = tmp[target_index].values[0]
self.time_print = tmp.index.values[0]
self.time_check = tmp.ts.values[0]
self.total_money = self.money + np.round(np.dot(self.stock_count,self.stock_price),6)
def buy(self):
money2 = copy.copy(self.money)
trading_fee = 0
if self.money < self.stock_price:
buy_stock = 0
buy_money = 0
elif self.money <= self.set_stand_price :
buy_stock = money2 // self.stock_price
buy_money = buy_stock * self.stock_price
self.money = self.money - buy_money
else :
have_money = money2
if have_money>self.set_stand_price: have_money = self.set_stand_price
buy_stock = have_money // self.stock_price
buy_money = buy_stock * self.stock_price
self.money = self.money - buy_money
self.my_asset_price = self.my_asset_price + buy_money
self.stock_count = np.round(self.stock_count + buy_stock,8)
self.total_money = self.money + np.round(np.dot(self.stock_count,self.stock_price),6)
print("buy ",self.time_print,self.stock_list,np.round(self.money),self.stock_count,
buy_stock,buy_money,np.round(self.total_money))
save_list = [self.stock_list,self.time_check,self.money,self.stock_count,
self.stock_price,self.total_money,"buy",buy_stock,buy_money,tax,
self.now_index_list]
save_txt(self.save_path+".txt",save_list, None)
self.buy_list =1
# return self.money, self.stocks, buy_stock,buy_money, total_money,self.my_asset_price, tax
def sell(self,all_tag):
if self.stock_count == 0:
sell_stock = 0
sell_money = 0
trading_fee = 0
elif self.my_asset_price <= self.set_stand_price or all_tag ==1:
sell_stock = self.stock_count
self.my_asset_price = 0
sell_money = sell_stock * self.stock_price
tradig_fee = sell_money *0.0025
else :
rate_ = np.round(self.set_stand_price/self.my_asset_price,2)
sell_stock = int(self.stock_count * rate_)
self.my_asset_price = self.my_asset_price - self.my_asset_price * (sell_stock/self.stock_count)
sell_money = sell_stock * self.stock_price
tradig_fee = sell_money *0.0025
self.stock_count = np.round(self.stock_count - np.round(sell_stock,8),8)
sell_money = sell_money - tradig_fee
self.money = self.money + np.round(sell_money,2)
self.total_money = self.money + np.round(np.dot(self.stock_count,self.stock_price),6)
if all_tag ==1:
print("all_sell",self.time_print,self.stock_list,np.round(self.money),self.stock_count,
sell_stock,sell_money,np.round(self.total_money))
save_list = [self.stock_list,self.time_check,self.money,self.stock_count,
self.stock_price,self.total_money,"all_sell",sell_stock,sell_money,tradig_fee,
self.now_index_list]
else:
print("sell",self.time_print,self.stock_list,np.round(self.money),self.stock_count,
sell_stock,sell_money,np.round(self.total_money))
save_list = [self.stock_list,self.time_check,self.money,self.stock_count,
self.stock_price,self.total_money,"sell",sell_stock,sell_money,tradig_fee,
self.now_index_list]
save_txt(self.save_path+".txt", save_list, None)
self.sell_list =1
# return self.money, self.stocks, sell_stock,sell_money,total_money,self.my_asset_price,tradig_fee
def log_recording(self):
log_info = [self.stock_list,self.time_check,self.stock_price,
self.now_index_list,
self.buy_list,self.sell_list,
self.stock_count,
self.v1,self.v2,
self.total_money]
save_txt(self.save_path+"_log.txt", log_info, None)
log_info2 = [self.time_check,self.money]+[self.stock_count]+[self.stock_price]
save_txt(self.save_path+"_total_price_log.txt", log_info2, None)
def yield_table(self):
try:
self.v1 = self.my_asset_price / self.stock_count
self.v2 = self.stock_price * self.stock_count / self.my_asset_price
except:
None
벡테스팅을 하기 위한 class이다.
class 내부에 자산, 현재 가격, 투자 지표 등을 저장해놓고 사용하는 것이 편해서 위의 코드 방식으로 구현했다. 조금 복잡하지만, 하나하나 뜯어보면 간단한 로직으로 구현되어있다.
함수 하나씩 설명해보자면,
def __init__(self,stock_list,save_path,seed_money,set_base_buy):
self.money = seed_money # 최초 보유 현금
self.total_money = seed_money # 총 자산(현금 + 투자금액 / 최초는 보유현금과 동일)
self.stock_list = stock_list # 벡테스팅 기업 이름
self.stock_count = 0 # 보유 주식 수
self.now_index_list = 0 # 투자 지표
self.stock_price = 0 # 투자 종목 현재 주가
self.my_asset_price = 0 # 투자 종목 투자 금액(보유 주식 수 * 현재 주가)
self.v1 = 1 # 평단가
self.v2 = 1 # 평가손익률
self.save_path = save_path # 저장 경로
self.set_stand_price = set_base_buy # 매수/매도 기준 금액
__init__함수는 객체가 생성될 때 필요한 변수들을 선언해 준다. 각 내용은 주석으로 달아 놓겠다.
def set_data(self,tmp,target_index):
self.buy_list = 0
self.sell_list = 0
self.stock_price = tmp.tail(1).Close.values[0]
self.now_index_list = tmp[target_index].values[0]
self.time_print = tmp.index.values[0]
self.time_check = tmp.ts.values[0]
self.total_money = self.money + np.round(np.dot(self.stock_count,self.stock_price),6)
set_data 함수는 일 별 주가가 입력되는 함수이다. 가격, 투자지표, 날짜와 현재 기준 총자산을 입력받는다.
def buy(self):
money2 = copy.copy(self.money)
trading_fee = 0
if self.money < self.stock_price:
buy_stock = 0
buy_money = 0
elif self.money <= self.set_stand_price :
buy_stock = money2 // self.stock_price
buy_money = buy_stock * self.stock_price
self.money = self.money - buy_money
else :
have_money = money2
if have_money>self.set_stand_price: have_money = self.set_stand_price
buy_stock = have_money // self.stock_price
buy_money = buy_stock * self.stock_price
self.money = self.money - buy_money
self.my_asset_price = self.my_asset_price + buy_money
self.stock_count = np.round(self.stock_count + buy_stock,8)
self.total_money = self.money + np.round(np.dot(self.stock_count,self.stock_price),6)
print("buy ",self.time_print,self.stock_list,np.round(self.money),self.stock_count,
buy_stock,buy_money,np.round(self.total_money))
save_list = [self.stock_list,self.time_check,self.money,self.stock_count,
self.stock_price,self.total_money,"buy",buy_stock,buy_money,trading_fee,
self.now_index_list]
save_txt(self.save_path+".txt",save_list, None)
self.buy_list =1
매수 함수이다. 투자 전략에 따라 매수기준에 통과하면 매수 함수 내 로직이 실행된다.
주요하게 봐야 할 곳은 if문이다. 만약 보유 현금이 현재 주가보다 작으면 매수 없이 통과한다.
두 번째 조건으로 보유 현금이 1회 매매 기준 금액보다 작으면 보유 현금을 모두 해당 종목에 매수한다.
세 번째 조건으로 보유 현금이 1회 매매 기준 금액보다 높으면 1회 매매 기준 금액만큼 매수한다.
분할 매수를 위해 위와 같은 로직을 설계했다. 1회 매매 기준 금액을 100만 원으로 설정해두었다면, 투자 전략 시뮬레이션 중 매수 기준에 충족할 때마다 100만 원씩 구매하는 것이다.
if 문 아래의 코드는 매수 후 현재 매수 총금액, 보유 주식 수, 총 자산 등을 기록하고 저장한다.
def sell(self,all_tag):
if self.stock_count == 0:
sell_stock = 0
sell_money = 0
trading_fee = 0
elif self.my_asset_price <= self.set_stand_price or all_tag ==1:
sell_stock = self.stock_count
self.my_asset_price = 0
sell_money = sell_stock * self.stock_price
trading_fee = sell_money *0.0025
else :
rate_ = np.round(self.set_stand_price/self.my_asset_price,2)
sell_stock = int(self.stock_count * rate_)
self.my_asset_price = self.my_asset_price - self.my_asset_price * (sell_stock/self.stock_count)
sell_money = sell_stock * self.stock_price
trading_fee = sell_money *0.0025
self.stock_count = np.round(self.stock_count - np.round(sell_stock,8),8)
sell_money = sell_money - trading_fee
self.money = self.money + np.round(sell_money,2)
self.total_money = self.money + np.round(np.dot(self.stock_count,self.stock_price),6)
if all_tag ==1:
print("all_sell",self.time_print,self.stock_list,np.round(self.money),self.stock_count,
sell_stock,sell_money,np.round(self.total_money))
save_list = [self.stock_list,self.time_check,self.money,self.stock_count,
self.stock_price,self.total_money,"all_sell",sell_stock,sell_money,trading_fee,
self.now_index_list]
else:
print("sell",self.time_print,self.stock_list,np.round(self.money),self.stock_count,
sell_stock,sell_money,np.round(self.total_money))
save_list = [self.stock_list,self.time_check,self.money,self.stock_count,
self.stock_price,self.total_money,"sell",sell_stock,sell_money,trading_fee,
self.now_index_list]
save_txt(self.save_path+".txt", save_list, None)
self.sell_list =1
매도 함수도 비슷한 로직으로 구현하였지만, 세부적으로 몇 가지가 다르다.
1회 매매 기준 금액보다 작으면 전부 매도, 크면 1회 매매 기준 금액만큼 매도한다.
여기서 조건문에 all_tag 변수가 있다. 만약 전량 매도를 해야 할 경우 1회 매매 기준 금액과 상관없이 보유량 전부를 매도하는 조건을 붙였다.
또한 매매수수료를 계산해야 한다. trading_fee변수에 몇 %를 수수료로 낼 것인지 적용한다. 코드상에서는 0.025%를 적용했다.
마찬가지로 조건문 통과 후 보유 주식 수, 보유 금액, 총액 등을 계산 저장한다.
def log_recording(self):
log_info = [self.stock_list,self.time_check,self.stock_price,
self.now_index_list,
self.buy_list,self.sell_list,
self.stock_count,
self.v1,self.v2,
self.total_money]
save_txt(self.save_path+"_log.txt", log_info, None)
log_info2 = [self.time_check,self.money]+[self.stock_count]+[self.stock_price]
save_txt(self.save_path+"_total_price_log.txt", log_info2, None)
def yield_table(self):
try:
self.v1 = self.my_asset_price / self.stock_count
self.v2 = self.stock_price * self.stock_count / self.my_asset_price
except:
None마지막 두 함수중 log_recording함수는 매매 기록과 전체 거래 기록을 일 별로 남기기 위한 함수이다.
후에 시각화할 때 일 별 거래 및 총 자산 현황을 보기 위해 기록한다.
yield_table은 평균 단가, 손익률을 계산하는 함수로 매 일별 실행된다.
코드상에 불필요해 보이는 변수들이 존재할 수도 있다. 애초에 여러 종목을 동시에 모니터링하며 자동 매매를 할 수 있게 설계했기 때문이다. 즉, 실제 투자처럼 여러 종목을 후보에 올려두고 매매 신호가 나오면 현재 보유 금액에서 투자할 수 있도록 하기 위함이다. 이는 단일 종목 벡 테스팅과 시각화 자료 설명 이후 소개하도록 하겠다.
def trading_simulation(bitrade,price_data,target_index,lower_cut,upper_cut,stop_loss):
lower_list = False
upper_list = False
plot_value = []
for st in price_data.index:
all_tag = 0
dt64 = np.datetime64(st)
ts = (dt64 - np.datetime64('1970-01-01T00:00:00Z')) / np.timedelta64(1, 's')
tmp = price_data[price_data.index == st]
tmp["ts"]=ts
bitrade.set_data(tmp,target_index)
# bitrade.set_stand_price = int(bitrade.total_money *0.5)
bitrade.yield_table()
if bitrade.v2 *100-100 < stop_loss :
all_tag = 1
if bitrade.stock_count >0:
bitrade.sell(all_tag)
lower_list = False
upper_list = False
else:
if bitrade.now_index_list <lower_cut:
lower_list = True
elif bitrade.now_index_list >= lower_cut and lower_list == True:
if bitrade.money >100000:
bitrade.buy()
lower_list = False
elif bitrade.now_index_list >upper_cut:
upper_list = True
elif bitrade.now_index_list <= upper_cut and upper_list == True:
if bitrade.stock_count >0:
bitrade.sell(all_tag)
upper_list = False
bitrade.log_recording()
plot_value.append(int(bitrade.money + (bitrade.stock_count*bitrade.stock_price)))
return plot_value
시뮬레이션하기 위한 전략이 구현된 함수이다.
나는 RSI 기준으로 기본 전략을 구현했다. now_index_list에 RSI 값이 들어가는 것으로 보면 된다.
매수 매도 컷을 70, 30으로 잡았다.
stop_loss는 손익률이 -로 떨어질 때 손절선을 지정하기 위함이다.
예를 들어 -5프로 이상 떨어지면 전량 매도하는 것을 말한다.
전략을 간단히 설명하자면, RSI가 30 이하일 때 lower_list의 스위치가 True로 활성화된다. 이는 30 이상이 넘어가면 매수하기 위한 대기 상태라 보면 된다. RSI가 30이 넘어가는 순간 매수하고 lower_list가 False로 비활성화된다. 이제 30~70사이에서 왔다갔다해도 추가 매수를 하지 않는다.
70이 넘어가면 upper_list가 True로 활성화 된다. 역시 70 이하로 떨어지는 순간 매도를 하고, 대기한다.
여기까지가 벡테스팅을 하기 위한 필수 함수들을 설명했다.
%%time
lower_cut = 30
upper_cut = 70
seed_money = 10000000
set_base_buy = 5000000
stop_loss = -100
bitrade = set_init(target_stock,save_path,seed_money,set_base_buy)
plot_value = trading_simulation(bitrade,price_data,"RSI",lower_cut,upper_cut,stop_loss)
print("DONE")
이제 벡테스팅을 진행해보자.
위에서 수집 가공한 삼성전자의 2013년 데이터부터 사용을 했다.
시드 머니는 1,000만원, 1회 매매 기준 금액은 500만원으로 설정하고, 손절선은 -100%으로 적용했다. -100%란 뜻은 원금이 다 잃었을 때란 말인데 이는 즉 손절선을 적용하지 않겠다는 말과 같다.
set_init로 종목 이름, 저장 경로, 시드 머니, 1회 매매 기준 금액을 입력해 객체를 생성한다. bitrade란 객체가 생성이 되었다.
벡테스팅을 위해 trading_simulation함수를 실행시킨다. bitrade 객체와 주가 데이터, 사용할 투자 지표 이름, 매매 하한, 상한선, 손절선을 입력하면 2013년부터 시작해 자동으로 매매를 진행해 결과를 낸다.
매매가 진행될 때마다 로그를 출력하도록 했다.
매수, 매도 일시, 구매 금액 보유 주식 수 등을 볼 수 있다.
로그로만 보면
조금 복잡하기에 그래프로 나타내었다.
fig, ax1 = plt.subplots(figsize = (15,7))
ax1.set_ylabel('Price')
plt.plot(plot_value)
plt.show()
우선, 시드 머니 기준으로 현재 총자산. 10년간 1000만원이 약 1400만원이 되었다(..?)
file_path = "./trading_simulation/trading_simulation_test"
log1 = read_log(file_path+"_log")
log1["lower"] = lower_cut
log1["upper"] = upper_cut
log_plot(log1)
저장된 로그를 불러와 시각화.
파란색이 매수, 빨간색이 매도, 하늘색 그래프가 주가, 연두색 그래프가 RSI 지표, 회색 바가 총 자산 대비 해당 종목 보유 비율.
color_list = "moccasin"
color_list2 = "tab:orange"
profit_rate_plot(file_path,color_list,color_list2)
매수, 매도 시 손익률 수준.
매도 시점에 몇 %의 손익률이었는지 볼 수 있다.
log_total = total_price_df(file_path,target_stock,seed_money)
total_plot(log_total,target_stock,color_list,True)
보유 자산 대비 삼성전자 투자 비율과 손익률를 시각화.
# 연평균 수익률 조회
np.round((np.power((log_total.tail(1)["total_money"].values[0] / seed_money),1/10)-1)*100,2)연평균 수익률 : 2.78
# 연도별 거래내역 조회
for i in range(2013,2023):
log_plot(log1,str(i),str(i+1))
연도 별로 거래 기록을 시각화한다. 앞서 전체 기록을 봤는데 기간이 길어질수록 촘촘해져 보기가 어렵기 때문에 특정 기간 별로 조회할 수 있게 구현하였다.
대략 위와 같은 그림으로 출력이 된다.
결국 벡테스팅 결과가 의도한 바에 의해 혹은 어떤 과정으로 매매를 진행했는지 이해하고 있어야만 자동 매매 프로그램을 만들 수 있기에, 전략을 구현하기 이전에 시각화 부분에 공을 들였다.
글이 너무 길어져서 시각화된 결과와 코드는 다음 글에서 이어서 설명하겠다.
'개발일지' 카테고리의 다른 글
| 6. 투자지표를 활용한 매매 시점 모니터링 - 손절선 (0) | 2022.11.09 |
|---|---|
| 5. 투자지표를 활용한 매매 시점 모니터링 - Kospi 종목 벡테스팅 (0) | 2022.11.08 |
| 4. 투자지표를 활용한 매매 시점 모니터링 - 시각화 (0) | 2022.11.08 |
| 2. 투자지표를 활용한 매매 시점 모니터링 - RSI, RMI (0) | 2022.11.07 |
| 1. 투자지표를 활용한 매매 시점 모니터링 - 개요 (0) | 2022.11.01 |
댓글