볼린저밴드-찾기

볼린저밴드 백테스트에 수수료와 슬리피지 추가

mode = 1 : 상승시 구매

mode = 2 : 하락시 구매

로 백테스트

 

""" 시간대별 데이터 불러오기 """
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import pyupbit

data = pyupbit.get_ohlcv("KRW-DOGE", interval = "minute5", count=12*24*7*4)

# window_b : 볼린저밴드 윈도우 
# window_f : 현금흐름지표 윈도우
# buy_pb : 매수조건 퍼센트볼린저 하한
# buy_mfi : 매수조건 현금흐름지표 하한
# sell_pb : 매도조건 퍼센트볼린저 상한
# sell_mfi : 매도조건 현금흐름지표 상한
# 0 < pb < 1
# 0 < mfi < 100
# slippage 거래당 0.5% (도지코인 실제 거래시 200원짜리 오차가 1원 씩 발생)
# fee 수수료 거래당 0.05%
def Bollenger_backtest(data=data, mode=1, fee=0.0005, slippage=0.005, window_b = 20, window_f=10, window_i=21, buy_pb = 0.8, buy_mfi = 80, sell_pb = 0.2, sell_mfi=20, chart=False):
    df = data.copy()
    df['MA20'] = df['close'].rolling(window=window_b).mean()
    df['stddev'] = df['close'].rolling(window=window_b).std()
    df['upper'] = df['MA20'] + df['stddev']*2
    df['lower'] = df['MA20'] - df['stddev']*2
    df['PB'] = (df['close'] - df['lower']) / (df['upper'] - df['lower']) 
    df['bandwidth'] = (df['upper'] - df['lower']) / df['MA20'] * 100
    df['TP'] = (df['high'] + df['low'] + df['close']) / 3
    df['PMF'] = 0 
    df['NMF'] = 0
    for i in range(len(df.close)-1):
        if df.TP.values[i] < df.TP.values[i+1]: # TP값이 전날보다 증가한다면, 
            df.PMF.values[i+1] = df.TP.values[i+1] * df.volume.values[i+1]
            df.NMF.values[i+1] = 0
        else:
            df.PMF.values[i+1] = 0
            df.NMF.values[i+1] = df.TP.values[i+1] * df.volume.values[i+1]
    df['MFR'] = df.PMF.rolling(window=window_f).sum() / df.NMF.rolling(window=window_f).sum()
    df['MFI'] = 100 - 100 / (1+df['MFR'])
    df['II'] = (2*df['close']-df['high']-df['low']) / (df['high']-df['low']) * df['volume']
    df['IIP'] = df['II'].rolling(window=window_i).sum() / df['volume'].rolling(window=window_i).sum() * 100  
    
    if mode == 1:
        cond_buy = (df['PB'] > buy_pb) & (df['MFI'] > buy_mfi)
        cond_sell = (df['PB'] < sell_pb) & (df['MFI'] < sell_mfi)
    elif mode == 2:
        cond_buy = (df['PB'] < buy_pb) & (df['MFI'] < buy_mfi)
        cond_sell = (df['PB'] > sell_pb) & (df['MFI'] > sell_mfi)
        
    df.loc[cond_buy, 'position'] = 1
    df.loc[cond_sell, 'position'] = 0
    
    df['position'] = df['position'].fillna(method = "ffill") # 마지막은 매도한 값인 0으로
#    df.iloc[-2]['position'] = 0
#    df.iloc[-1]['position'] = 0
    df['position2'] = np.where(df.position != df.position.shift(1), df.position, np.NaN) 
    
    df['return'] = np.log(df['close'].shift(-1)/df['close']) # 로그수익률 : 당일 종가에 매매시 다음날 종가와 비교하여..   
    df['return_trade'] = np.log(df['close'].shift(-1)/df['close']) # 로그수익률 : 당일 종가에 매매시 다음날 종가와 비교하여..   
    
    cond_trade = df['position2'].notnull() # 거래가 이뤄지는 조건
    df.loc[cond_trade,'return_trade'] = np.log(np.exp(df.loc[cond_trade,'return_trade'])-fee-slippage) # 수수료와 slippage 고려
        
    df['strategy'] = df['position'] * df['return_trade']
    df['cumret'] = df['strategy'].cumsum().apply(np.exp)
    df['cummax'] = df['cumret'].cummax()
    
    #df = df[max(window_b, window_f):]
      
    drawdown = df['cummax'] - df['cumret']
    mdd = drawdown.max()
    result = df[['return','strategy']].sum().apply(np.exp)
    strategy = result['strategy']
    print("backtest results : ")
    print(data.index[0] , " ~ ", data.index[-1], data.shape[0])
    print("buy : sell = " + str(df.loc[cond_buy].shape[0]) + " : " + str(df.loc[cond_sell].shape[0]))
    print("return : ", np.round(result['return']*100,1),"%")
    print("strategy : ", np.round(result['strategy']*100,1),"%")
    print("mdd : △", np.round(mdd*100,0),"%")
       
    if chart == True:
        fig, (ax1, ax2, ax3) = plt.subplots(nrows=3, figsize=(20,8))
        ax1.plot(df.index, df['close'], color='#0000ff', label='close')
        ax1.plot(df.index, df['upper'], 'r--', label='upper band')
        ax1.plot(df.index, df['lower'], 'c--', label='lower band')
        ax1.plot(df.index, df['MA20'], 'k--', label=f'MA{window_b}')
        ax1.fill_between(df.index, df['upper'], df['lower'], color='0.9')
        for i in range(len(df.close)):
            if mode == 1:
                if df.PB.values[i] > buy_pb and df.MFI.values[i] > buy_mfi:
                    ax1.plot(df.index.values[i], df.close.values[i], 'r^')
                elif df.PB.values[i] < sell_pb and df.MFI.values[i] < sell_mfi:
                    ax1.plot(df.index.values[i], df.close.values[i], 'bv')
            elif mode == 2:
                if df.PB.values[i] < buy_pb and df.MFI.values[i] < buy_mfi:
                    ax1.plot(df.index.values[i], df.close.values[i], 'r^')
                elif df.PB.values[i] > sell_pb and df.MFI.values[i] > sell_mfi:
                    ax1.plot(df.index.values[i], df.close.values[i], 'bv')
        ax1.legend(loc='best')
        ax2.plot(df.index, df['PB']*100, 'b', label= '%B x 100')
        ax2.plot(df.index, df['MFI'], 'g--', label=f'MFI({window_f} day)')
        ax2.set_yticks([-20, 0, 20, 40, 60, 80, 100, 120])
        for i in range(len(df.close)):
            if mode == 1:
                if df.PB.values[i] > buy_pb and df.MFI.values[i] > buy_mfi:
                    ax2.plot(df.index.values[i], 0, 'r^')
                elif df.PB.values[i] < sell_pb and df.MFI.values[i] < sell_mfi:
                    ax2.plot(df.index.values[i], 0, 'gv')
            elif mode == 2:
                if df.PB.values[i] < buy_pb and df.MFI.values[i] < buy_mfi:
                    ax2.plot(df.index.values[i], 0, 'r^')
                elif df.PB.values[i] > sell_pb and df.MFI.values[i] > sell_mfi:
                    ax2.plot(df.index.values[i], 0, 'gv')
        #ax2.grid(True)
        ax2.legend(loc='best')
        df[['return','strategy']].cumsum().apply(np.exp).plot(ax=ax3)
    return df

# window_b : 볼린저밴드 윈도우 
# window_f : 현금흐름지표 윈도우
# buy_pb : 매수조건 퍼센트볼린저 하한
# buy_mfi : 매수조건 현금흐름지표 하한
# sell_pb : 매도조건 퍼센트볼린저 상한
# sell_mfi : 매도조건 현금흐름지표 상한
# 0 < pb < 1
# 0 < mfi < 100
# slippage 거래당 0.5% (도지코인 실제 거래시 200원짜리 오차가 1원 씩 발생)
# fee 수수료 거래당 0.05%
def Bollenger_backtest_find(data=data, mode=1, fee=0.0005, slippage=0.005, window_b = 20, window_f=10, window_i=21, buy_pb = 0.8, buy_mfi = 80, sell_pb = 0.2, sell_mfi=20, chart=False, pprint=True):
    df = data.copy()
    df['MA20'] = df['close'].rolling(window=window_b).mean()
    df['stddev'] = df['close'].rolling(window=window_b).std()
    df['upper'] = df['MA20'] + df['stddev']*2
    df['lower'] = df['MA20'] - df['stddev']*2
    df['PB'] = (df['close'] - df['lower']) / (df['upper'] - df['lower']) 
    df['bandwidth'] = (df['upper'] - df['lower']) / df['MA20'] * 100
    df['TP'] = (df['high'] + df['low'] + df['close']) / 3
    df['PMF'] = 0 
    df['NMF'] = 0
    for i in range(len(df.close)-1):
        if df.TP.values[i] < df.TP.values[i+1]: # TP값이 전날보다 증가한다면, 
            df.PMF.values[i+1] = df.TP.values[i+1] * df.volume.values[i+1]
            df.NMF.values[i+1] = 0
        else:
            df.PMF.values[i+1] = 0
            df.NMF.values[i+1] = df.TP.values[i+1] * df.volume.values[i+1]
    df['MFR'] = df.PMF.rolling(window=window_f).sum() / df.NMF.rolling(window=window_f).sum()
    df['MFI'] = 100 - 100 / (1+df['MFR'])
    df['II'] = (2*df['close']-df['high']-df['low']) / (df['high']-df['low']) * df['volume']
    df['IIP'] = df['II'].rolling(window=window_i).sum() / df['volume'].rolling(window=window_i).sum() * 100  
    
    if mode == 1:
        cond_buy = (df['PB'] > buy_pb) & (df['MFI'] > buy_mfi)
        cond_sell = (df['PB'] < sell_pb) & (df['MFI'] < sell_mfi)
    elif mode == 2:
        cond_buy = (df['PB'] < buy_pb) & (df['MFI'] < buy_mfi)
        cond_sell = (df['PB'] > sell_pb) & (df['MFI'] > sell_mfi)
        
    df.loc[cond_buy, 'position'] = 1
    df.loc[cond_sell, 'position'] = 0
    
    df['position'] = df['position'].fillna(method = "ffill") # 마지막은 매도한 값인 0으로
#    df.iloc[-2]['position'] = 0
#    df.iloc[-1]['position'] = 0
    df['position2'] = np.where(df.position != df.position.shift(1), df.position, np.NaN) 
    
    df['return'] = np.log(df['close'].shift(-1)/df['close']) # 로그수익률 : 당일 종가에 매매시 다음날 종가와 비교하여..   
    df['return_trade'] = np.log(df['close'].shift(-1)/df['close']) # 로그수익률 : 당일 종가에 매매시 다음날 종가와 비교하여..   
    
    cond_trade = df['position2'].notnull() # 거래가 이뤄지는 조건
    df.loc[cond_trade,'return_trade'] = np.log(np.exp(df.loc[cond_trade,'return_trade'])-fee-slippage) # 수수료와 slippage 고려
        
    df['strategy'] = df['position'] * df['return_trade']
    df['cumret'] = df['strategy'].cumsum().apply(np.exp)
    df['cummax'] = df['cumret'].cummax()
    
    #df = df[max(window_b, window_f):]
      
    drawdown = df['cummax'] - df['cumret']
    mdd = drawdown.max()
    result = df[['return','strategy']].sum().apply(np.exp)
    strategy = result['strategy']
    if pprint == True :
        print("window_b=",window_b, "window_f=" , window_f, "buy_pb=", buy_pb, "buy_mfi=", buy_mfi, "sell_pb=", sell_pb, "sell_mfi=", sell_mfi, strategy, mdd)
    return strategy, mdd

---

○ 슬리피지 포함 mode = 1 

a = Bollenger_backtest(data=data, mode =1, fee=0.005, slippage=0.05, window_b = 20, window_f=10, window_i=21, buy_pb = 0.8, buy_mfi = 80, sell_pb = 0.2, sell_mfi=20, chart=True)

○ 슬리피지 포함 mode = 2

a = Bollenger_backtest(data=data, fee=0.0005, slippage=0.005, mode =2, window_b = 50, window_f=5,buy_pb = 0.2, buy_mfi = 20, sell_pb = 0.8, sell_mfi=80, chart=True)

> 안하는 것만 못하다

 

---

○ 파라미터 최적화해보기

# mode = 2 는 낮을때 사고 높을 때 파는 것
results = pd.DataFrame(columns = ['window_b', 'window_f','buy_pb','buy_mfi','sell_pb','sell_mfi','strategy', 'mdd'])
mode = 2
i=0
for j in range(1000):
    b = np.random.randint(5, 50)
    f = np.random.randint(3, 30)
    b_mfi = np.random.randint(1, 99) #1~99
    s_mfi = np.random.randint(1, 100) #1~99
    b_pb = np.random.rand(1)[0] # 0~1
    s_pb = np.random.rand(1)[0] # 0~1
    
    if mode == 2:
        if (b_mfi < s_mfi) & (b_pb < s_pb):    
            try:
                strategy, mdd = Bollenger_backtest_find(data=data, mode=mode, window_b = b, window_f = f,  buy_pb = b_pb, buy_mfi = b_mfi, sell_pb = s_pb, sell_mfi=s_mfi, pprint=True)
                results.loc[i, 'window_b'] = b
                results.loc[i, 'window_f'] = f
                results.loc[i, 'buy_pb'] = b_pb
                results.loc[i, 'buy_mfi'] = b_mfi
                results.loc[i, 'sell_pb'] = s_pb
                results.loc[i, 'sell_mfi'] = s_mfi
                results.loc[i, 'strategy'] = strategy
                results.loc[i, 'mdd'] = mdd 
                i=i+1
            except:
                print('error')
    pass

a = Bollenger_backtest(data=data, mode=2, fee=0.0005, slippage=0.005, window_b = 26, window_f=5, window_i=21, 
                       buy_pb = 0.1, buy_mfi = 50, sell_pb = 0.88, sell_mfi=66, chart=True)

▷ 도지코인은 가장 좋았던 난수 시나리오를 사용하면, 그냥 지니고 있는 것 보다는 낫다.

 

---

○ 비트코인으로 테스트

▷ + 가 나는 시나리오가 없다고 봐도 무방 

 

---

○ NEAR 코인

mdd가 높은편이지만 수익률도 꽤 높다. 하지만..

▷ 보유수익률만도 못하다. 

mode=1도 마찬가지. 

 

(결론)

볼린저밴드만 가지고 실전에서 테스트하는 것은 무리인 것 같다. 

다른 지표들을 검색해보고, 다른 알고리즘을 추가적으로 넣어봐야할 것 같다. 

---