BitcoinBacktest / 파이썬비트코인백테스트

# -*- coding: utf-8 -*-
"""
Created on Thu Mar  5 21:02:35 2020

@author: grw19
"""

import pandas as pd
import talib
import time
"""
1. 데이터 불러오기
2. 지표 설정하기 (talib)
3. 매수 시점 결정하기 (조건 + 매수안한 상태일때만)
4. 매도 시점 결정하기 (조건 + 매수한 상태일때만)
5. 백테스트하기
6. 결과출력하기 (csv로 출력해서 엑셀로 분석)

"""

class CoinBacktest :
    def __init__(self):
        self.df = pd.read_csv("C:/coding/coin/data/BTCUSDT_1hour.csv")
        self.df['buy'] = 0
        self.df['sell'] = 0
        self.df['signal'] = 0
        self.indicator = pd.DataFrame()
        self.indicator['date'] = self.df['TotalTime']

    def cal_indicator(self) :
        print("지표를 계산합니다...")
        start_time = time.time()
        self.indicator['sma3'] = talib.SMA(self.df['priceClose'], timeperiod=3)
        self.indicator['3dis'] =  self.df['priceClose'] - self.indicator['sma3'] # 3시간 이격도

        print(time.time() - start_time , "소요, 지표 계산 완료...")

    def buy_signal(self) : # 조건 만족 시 매수열을 만들어서 0 -> 1로 설정
        print("매수 시점을 계산합니다...")
        start_time = time.time()

        for i in range(0,len(self.indicator)) :
            if self.indicator['3dis'][i] >=0 : # 이격도 위 + 미보유시 매수
                print(i, "매수")
                self.df['buy'][i] = 1 # 매수 조건 성립 시 1을 대입


        print(time.time() - start_time , "소요, 매수 시점 완료...")
    def sell_signal(self) : # 조건 만족 시  매도행을 만들어서 0 -> 1로, 매수행은 1 -> 0으로 설정
        print("매도 시점을 계산합니다...")
        start_time = time.time()

        for i in range(0,len(self.indicator)) :
            if self.indicator['3dis'][i] < 0  : # 매수중인상태로 이격도 아래에 있으면 매도
                print(i, "매도")
                self.df['sell'][i] = 1 # 매수 조건 성립 시 1을 대입


        print(time.time() - start_time, "소요, 매도 시점 완료...")

    def backtest(self,df) :
        print("백테스트를 시작합니다...")
        start_time = time.time()
        FEE = 0.1 / 100 # 수수료 0.06% + 슬리피 0.04%로 0.1%로 설정 (매수매도합 0.02%)
        money = 1000   # 초기 원금 1000달러로 설정
        slot = 0 # 미보유시 0, 보유시 1

        buy_date_list = []
        buy_price_list = []
        amount_list = []
        sell_date_list = []
        sell_price_list = []
        money_list = []
        result_list = []
        for i in range(0,len(self.df)-1) : # 한줄 한줄 해줌
            if self.df['buy'][i] == 1 and slot == 0 : # 매수시그널이 나왔고, 미 보유 중일 경우
                print("매수 진입...")
                slot = 1 # 보유 상태로 바꿔줌

                buy_date = df['TotalTime'][i+1]
                buy_price = df['priceOpen'][i+1] # 시그널이 뜬 다음 시가에 매수
                amount = (money / buy_price) / (1+FEE)  # 매수 수량 : 시가 / 보유수량에서 수수료 고려
                buy_date_list.append(buy_date) # 날짜
                buy_price_list.append(buy_price) # 가격
                amount_list.append(amount) # 수량

            if self.df['sell'][i] == 1 and slot == 1 : # 매도시그널이 나왔고, 보유 중일 경우
                print("매도 청산...")
                slot = 0 # 미보유 상태로 바꿔줌
                sell_price = df['priceOpen'][i+1] # 매도 시그널이 뜬 다음 시가에 매도
                sell_date = df['TotalTime'][i+1]
                result = (sell_price / buy_price)
                money = money * result / (1+FEE) # 수량 * 수익률 / 수수료
                sell_date_list.append(sell_date)
                sell_price_list.append(sell_price)
                money_list.append(money)
                result_list.append(result)
        if len(buy_date_list) > len(sell_date_list) : # 마지막 매수는 아직 보유 중임. 이건 그냥 잘라줌
            del buy_date_list[-1]
            del buy_price_list[-1]
            del amount_list[-1]


        result_data = {'buy_date': buy_date_list, 'buy_price': buy_price_list, 'amount': amount_list,'sell_date' : sell_date_list,
                         'sell_price' : sell_price_list,'money' : money_list,'result' :result_list }
        result_df = pd.DataFrame(result_data,
                          columns=['buy_date', 'buy_price', 'amount', 'sell_date', 'sell_price', 'money', 'result'])

        result_df.to_csv("C:/coding/coin/result/backtest.csv", header = True, index = False) # csv파일로 저장


if __name__ == "__main__":
    start_time = time.time()
    backtest = CoinBacktest()
    backtest.cal_indicator()
    backtest.buy_signal()
    backtest.sell_signal()
    backtest.backtest(backtest.df)
    print(time.time() - start_time , "소요, 백테스트 완료...")