파이썬으로 간단한 퀀트투자 백테스트 Tool(BackTester) 만들기

퀀트투자 백테스트를 여러가지로 하고 싶어서 백테스트 툴을 만들었다.

 

만들게 된 계기는 강환국씨와 닥터퀀트, systrader79, Quant 실전 퀀트투자(홍용찬, 이래미디어) 책을 읽고나서 왠지 간단하게 만들 수 있겠다는 생각이 들어서이다.

 

당연히 시중에 이렇게 백테스트를 하는 툴이 유료로 존재한다. 사실 사용해 본적은 없다. 돈이 없어서이다.(가난한 40대 가장의 현실이다.)

 

유로툴처럼 많은 기능이나 정확성 그리고 UI 디자인은 구현할 수 없지만, 나름 비슷하게는 만들 수 있다는 생각이 며칠전부터 들었다.

 

차차 업데이트를 할 목적으로 우선 간단하게 만들었다. 

 

계략적인 기능은 아래와 같다. 그리고 1년단위로 리밸런싱 하는 조건으로 만들었다. 리밸런싱 기간 조건도 선택을 할 수 있게 기능을 추가할 수 있지만,  귀찮다. 나는 그냥 1년단위로 투자를 할거다.

 

1. 백테스트 시작년도와 마지막년도를 입력하고, 백테스트를 할 월, 일을 입력한다.

 

2. 날짜는 영업일이 아니면 자동으로 근처 영업일로 바꾸어 주도록 함수를 작성했다.

 

3. 매수종목 개수를 설정할 수 있고, 거래대금, 시가총액을 조건에 설정할 수 있다.

 

4. 내가 가지고 있는 데이터 내에서 몇 가지 항목만 Factor로 만들었다. 추후 재무비율 등의 데이터를 확보하면 업데이트 할 계획도 있다.

 

5. Result 란에는 년도별 수익률과 누적수익률이 나온다.

 

6. BackData는 내가 Test한 종목의 일자, code, 종목명... 이런 정보들이 나타난다.

 

7. CAGR과 MDD는 PushButton 옆에 보이도록 했다.

 

8. Excel은 매번 필요한게 아니니 선택을 하도록 Radio Box로 만들었다.

 

9. 성격이 느긋하지 않아서 진행사항을 보기위해 Progress Bar도 추가했다.

 

10. BackData는 TextEdit으로 처음에 만들었다가 열과 행이 반듯하지 않아 보기 싫어서 Table Widget으로 바꿨다.

 

퇴근 후에 잠자는 시간을 쪼개서 만들다보니 예쁘게 만들기는 포기했다.

 

비전공에 비전문가 입장에서 혼자 독학으로 파이썬을 하는것은 절대 쉽지가 않았다.

 

하지만, 예상보다 작동이 잘 되고 활용성이 좋았다. 상당한 보람을 느낀다.

 

import sys
from PyQt5.QtWidgets import *
from PyQt5 import uic
import sqlite3
import pandas as pd
 
ui = uic.loadUiType('Q-Bot10.ui')[0]
 
class Window(QMainWindow, ui):
    def __init__(self):
        super().__init__()
        self.setupUi(self)
 
        # 시작년도 Combobox 만들기
        for sy in range(2003, 2023):
            self.cb_start.addItem(str(sy))
 
        # 백테스트의 마지막년도 Combobox 만들기
        for ey in range(2004, 2023):
            self.cb_end.addItem(str(ey))
 
        # 백테스트할 월을 Combobox로 만들기
        for m in range(1, 13):
            self.cb_month.addItem(str(m))
 
        # 백테스트할 날짜를 Combobox로 만들기
        for ㅇ in range(1, 32):
            self.cb_day.addItem(str(ㅇ))
 
        # 투자종목 개수를 Combobox로 만들기
        for c in range(10, 31):
            self.cb_count.addItem(str(c))
 
        # 거래대금을 Combobox로 만들기
        for t in range(0, 10000000000, 100000000):
            self.cb_transamount.addItem(f'{t:,}')
 
        # 거래대금 이상/이하를 Combobox로 만들기
        for tml in ['이상', '이하']:
            self.cb_transmoreless.addItem(tml)
 
        # 시가총액을 Combobox로 만들기
        for mc in range(0, 10000000000000, 50000000000):
            self.cb_cap.addItem(f'{mc:,}')
 
        # 시가총액 이상/이하를 Combobox로 만들기
        for cml in ['이상', '이하']:
            self.cb_cap_moreless.addItem(cml)
 
        # 백테스트 버튼과 연결 시그널
        self.btn_test.clicked.connect(self.back_test)
 
    def back_test(self):
        try:
            con = sqlite3.connect('krx_data_new.db')
 
            # DB에 있는 날짜인지 확인하여 없으면 근처일로 변경
            def near_business_day(b_start, b_month, b_day):
                df_date = pd.read_sql("SELECT 일자 FROM fund WHERE 일자 LIKE " + "'" + b_start + "%'", con)
                df_date.drop_duplicates(inplace=True)
                li_df = df_date['일자'].tolist()
 
                date = b_start + b_month + b_day
                if date in li_df:
                    b_date = date
                else:
                    if int(b_day) < 15: #날짜가 15미만이면 1씩 더해서 영업일 구하기
                        while 10: #10번 정도 더하기 반복해서 DB에 있는 날짜 구하기
                            b_day = int(b_day) + 1
                            b_day = str(b_day).zfill(2)
                            b_date = date[:6] + b_day
                            if b_date in li_df:
                                break
                    else: #날짜가 15이상이면 1씩 빼면서 영업일 구하기
                        while 10:
                            b_day = int(b_day) - 1
                            b_day = str(b_day).zfill(2)
                            b_date = date[:6] + b_day
                            if b_date in li_df:
                                break
                return b_date
 
            b_start = self.cb_start.currentText() #백테스트 시작년도
            b_end = self.cb_end.currentText() #백테스트 마지막 년도
            b_month = self.cb_month.currentText() #백테스트를 할 월
            b_month = b_month.zfill(2) #월이 한자리 수일때 앞에 0을 채우기
            b_day = self.cb_day.currentText() #백테스트 할 날짜
            b_day = b_day.zfill(2) #일이 한자리 수일때 앞에 0을 채우기
            count = int(self.cb_count.currentText()) #투자종목 개수
 
            #Group Box의 Radio Button, 백테스트 할 요소 선택하기
            if self.rb_pbr.isChecked():
                col1 = 'PBR'
            elif self.rb_dps.isChecked():
                col1 = 'DPS'
            elif self.rb_per.isChecked():
                col1 = 'PER'
            elif self.rb_hcapital.isChecked():
                col1 = '시가총액'
            elif self.rb_lcapital.isChecked():
                col1 = '시가총액'
            elif self.rb_heps.isChecked():
                col1 = 'EPS'
            elif self.rb_leps.isChecked():
                col1 = 'EPS'
            elif self.rb_hdiv.isChecked():
                col1 = 'DIV'
            elif self.rb_ldiv.isChecked():
                col1 = 'DIV'
            print(col1)
 
            if self.rb_pbr_2.isChecked():
                col2 = 'PBR'
            elif self.rb_dps_2.isChecked():
                col2 = 'DPS'
            elif self.rb_per_2.isChecked():
                col2 = 'PER'
            elif self.rb_hcapital_2.isChecked():
                col2 = '시가총액'
            elif self.rb_lcapital_2.isChecked():
                col2 = '시가총액'
            elif self.rb_heps_2.isChecked():
                col2 = 'EPS'
            elif self.rb_leps_2.isChecked():
                col2 = 'EPS'
            elif self.rb_hdiv_2.isChecked():
                col2 = 'DIV'
            elif self.rb_ldiv_2.isChecked():
                col2 = 'DIV'
            print(col2)
 
            trans_amount = self.cb_transamount.currentText() #거래대금을 변수로 지정
            trans_amount = int(trans_amount.replace(',', '')) #쉼표를 빼고 정수로 변경
            market_cap = self.cb_cap.currentText() #시가총액을 변수로 지정
            market_cap = int(market_cap.replace(',', '')) #쉼표를 빼고 정수로 변경
 
 
            li_result = [] #년도와 수익률을 담을 리스트
            i = 0 #첫번째 년도를 확인하기 위해 만든 임의 변수
            self.tw_backdata.clear() #Back Data용 Text Edit 기존내용 지우기
            self.te_return.clear() #Result에 기존내용 지우기
            for b_year in range(int(b_start), int(b_end)+1):
 
                b_date = near_business_day(str(b_year), b_month, b_day) #백테스트 기준일을 영업일로 변경
                print(b_date)
 
                if col1 == "시가총액" or col2 =="시가총액": #시가총액이 선택되었는지 확인
                    df = pd.read_sql("SELECT 일자, code, 종목명, 종가, 거래량, 거래대금, 상장주식수, " + col1 +
                                     ", " + col2 + " FROM Fund WHERE 일자= " + b_date, con)
                else:
                    df = pd.read_sql("SELECT 일자, code, 종목명, 종가, 거래량, 거래대금, 상장주식수, 시가총액, " + col1 +
                                     ", " + col2 + " FROM Fund WHERE 일자= " + b_date, con)
 
                df = df[(df['거래량'] > 0) & (df[col1] > 0) & (df[col2] > 0)]  # 거래량, col1, col2가 0이상인 종목만 가져오기
                print(df)
 
                # 일 거래대금 조건 적용하기
                if self.cb_transmoreless.currentText() == "이상":
                    df = df[df['거래대금'] >= trans_amount]
                else:
                    df = df[df['거래대금'] < trans_amount]
 
                # 시가총액 조건 적용하기
                if self.cb_cap_moreless.currentText() == "이상":
                    df = df[df['시가총액'] >= market_cap]
                else:
                    df = df[df['시가총액'] < market_cap]
 
                # col1 순위매기기
                if self.rb_hcapital.isChecked() or self.rb_heps.isChecked() or self.rb_hdiv.isChecked():
                    df[col1 + '_rank'] = df[col1].rank(ascending=False)  # 높은 순서대로 순위 매기기
                else:
                    df[col1 + '_rank'] = df[col1].rank()  # 낮은 순서대로 순위 매기기
 
                # col2 순위매기기
                if self.rb_hcapital_2.isChecked() or self.rb_heps_2.isChecked() or self.rb_hdiv_2.isChecked():
                    df[col2 + '_rank'] = df[col2].rank(ascending=False)  # 높은 순서대로 순위 매기기
                else:
                    df[col2 + '_rank'] = df[col2].rank()  # 낮은 순서대로 순위 매기기
 
                df[col1 + '_' + col2] = df[col1 + '_rank'] + df[col2 + '_rank']  # 순위 매긴것을 더하기
                df[col1 + '_' + col2 + '_' + 'rank'] = df[col1 + '_' + col2].rank()  # 최종순위 구하기
                df.sort_values(by=col1 + '_' + col2 + '_' + 'rank', inplace=True)  # 낮은 순서대로 정렬하기
 
                df = df.iloc[:count]  # 지정한 개수만 구하기
 
                # 1년후의 날짜 구하기
                date_later = near_business_day(str(b_year+1), b_month, b_day)
 
                # 1년후의 날짜로 데이터 가져오기
                df_later = pd.read_sql("SELECT 일자, code, 종가, 상장주식수 FROM fund WHERE 일자= " + date_later, con)
                df_later.columns = ['1년후의_일자', 'code', '1년후_종가', '1년후_상장주식수']
 
                df = pd.merge(df, df_later, on='code')
 
                df['수익률'] = ((df['1년후_종가'] - df['종가']) / df['종가']) * (df['1년후_상장주식수'] / df['상장주식수'])
                profit = df['수익률'].mean()
                li_result.append([b_date, profit])
 
                if i == 0:  # 첫번째 날짜는 만들어진 dataframe이 df_t로 저장되고, 나머지는 계속 concat로 붙여넣기
                    df_t = df
                else:
                    df_t = pd.concat([df_t, df])
 
                prb = i / (int(b_end) - int(b_start)) * 100  # Progress Bar
                self.prb.setValue(prb)  # Progress Bar
 
                i += 1
 
            print(df_t)
 
            # 수익률의 Back Data를 Table Widget에 나타내기
            self.tw_backdata.setRowCount(len(df_t))
            self.tw_backdata.setColumnCount(len(df_t.columns))
            self.tw_backdata.setHorizontalHeaderLabels(df_t.columns.tolist())
            for b_row in range(len(df_t)):
                for b_col in range(len(df_t.columns)):
                    self.tw_backdata.setItem(b_row, b_col, QTableWidgetItem(str(df_t.iloc[b_row, b_col])))
 
            df_result = pd.DataFrame(data=li_result, columns=['투자년도', '수익률'])  # 투자년도와 수익률로 데이터프레임 만들기
 
            li_acc = []
            for n, p in enumerate(df_result['수익률']):
                if n == 0:
                    li_acc.append(1 + p)
                else:
                    li_acc.append(li_acc[n - 1] * (1 + p))
 
            df_result['누적수익률'] = pd.DataFrame(data=li_acc, columns=['누적수익률'])
            cagr = (df_result['누적수익률'].iloc[-1]) ** (1 / len(df_result)) - 1
            cagr = str(round(cagr * 100, 2))
            self.lbl_cagr.setText(cagr) #CAGR 레이블에 나타내기
 
            MDD = df_result['수익률'].min()
 
            self.lbl_mdd.setText(f'{MDD*100:.2f}')
 
            #수익률을 textEdit TextEdit에 나타내기
            self.te_return.setText('투자시기     수익률     누적수익률')
            for row in range(len(df_result)):
                invest_y = str(df_result['투자년도'].iloc[row])
                return_rate = df_result['수익률'].iloc[row]
                acc_return = df_result['누적수익률'].iloc[row]
                self.te_return.append(f'{invest_y}      {return_rate:.2f}         {acc_return:.2f}')
 
            #Excel Export를 선택시 엑셀파일에 결과 저장
            if self.rb_excel.isChecked():
                trans_amount = self.cb_transamount.currentText()  # 거래대금을 변수로 지정
                market_cap = self.cb_cap.currentText()  # 시가총액을 변수로 지정
                transmoreless = self.cb_transmoreless.currentText()
                cap_moreless = self.cb_cap_moreless.currentText()
                # 하나의 엑셀파일에 2개의 시트를 저장하기 위해 ExcelWriter 함수를 사용
                writer = pd.ExcelWriter('백테스트결과(' + b_start + b_month + b_day + '-' + b_end +')_'
                                        + col1 + 'x' + col2 + '시가총액' + market_cap + transmoreless +
                                        '거래대금' + trans_amount + cap_moreless + '.xlsx')
                df_result.to_excel(writer, sheet_name='년도별수익률')
                df_t.to_excel(writer, sheet_name='BackData')
                writer.save()
        except:
            pass
 
if __name__=='__main__':
    app = QApplication(sys.argv)
    window = Window()
    window.show()
    app.exec_()
 
 

 

Q-Bot10.ui

0.02MB