[변동성 돌파 전략 3] 변동성 돌파 전략 백테스트 k 최적화 (베이지안 최적화)

저번 포스팅에 이어 이번에는 베이지안 최적화를 통해 BTC/USDT 1일봉 기준 변동성 돌파전략에서의 최적의 k값을 찾고자 한다.

 

---

 

저장된 데이터를 받아오고, 이에 대한 변동성 돌파 백테스트 함수를 선언하는 것은 저번 포스팅과 동일하다.

import pandas as pd
import numpy as np
import random as rand
from bayes_opt import BayesianOptimization
import matplotlib.pyplot as plt

################ 백테스트 ################

# 백테스트 기본 셋팅

symbol = "BTCUSDT" # 다른 티커도 가능
interval = '1d' # 1m, 3m, 5m, 15m, 30m, 1h, 2h, 4h, 6h, 8h, 12h, 1d, 3d, 1w, 1M
year = 'All' # 2017~2022 중 선택 가능.

# 데이터 읽어오기: 전체 데이터
df = pd.read_csv (f'C:.\\Data\\{symbol[:-4]}-USDT_Data\\{symbol[:-4]}-USDT_Data_{interval}\\{symbol[:-4].upper()}_USDT_{interval}_{year}.csv')
# 전체: _All
# 기간: 연도 불러오면 됨
df = pd.DataFrame(df)
df = df.dropna(how='any')

# 변동성 돌파 전략 위한 로직 설정

def VBS(df, config_data):
    # k값 넣기
    k = config_data['k']

    # 목표가 구하기
    df['range'] = df['High'] - df['Low']  # 고저 변동폭
    df['target'] = df['Open'] + df['range'].shift(1) * k  # 목표가 한칸 내려주고 (shift), 이후 k값을 곱한 목표가 계산
    # df = df.iloc[1:-2]

    # 매수 시뮬레이션
    df['ror'] = np.where(df['High'] > df['target'], df['Close'] / df['target'],
                         1)  # high > target인 지점에서 close / target

    # 최종 누적 산출
    df['total'] = df['ror'].cumprod()
    final_cum_ror = (df['total'].iloc[-1].astype(float)) * 100 -100

    # 연간 수익률 (기간 수익률)
    N = ((df.index[-1] - df.index[0])) / 365
    CAGR = (final_cum_ror ** (1 / N))

    # dd값 기록 및 mdd 계산
    array_v = np.array(df['total'])
    dd_list = -(np.maximum.accumulate(array_v) - array_v) / np.maximum.accumulate(array_v)
    peak_lower = np.argmax(np.maximum.accumulate(array_v) - array_v)
        # np.maximum.accumulate(array_v)는 최고값을 계속 갱신한 array
        # 그 array에서 원래 array를 뺀 값들의 array를 새로 생성
        # 뺀 값만 남은 array 중 가장 큰 값의 index를 추출
    peak_upper = np.argmax(array_v[:peak_lower])
        # peak_lower의 index값까지의 array_v값 추출
        # 그 중에 가장 큰 놈 (MDD 계산을 위한 최고값) 의 index 추출
    mdd = round((array_v[peak_lower] - array_v[peak_upper]) / array_v[peak_upper] * 100, 3)

    return final_cum_ror, CAGR, mdd, dd_list,df['total'] # 최종 수익률, 연간 수익률, 최대손실폭 (mdd), 손실 목록, 누적 이익 목록

 

---

 

이후 beyes optimization 모듈을 활용하여 최대수익률을 기록하는 k값을 도출하면 된다.

 

### 베이지안 최적화

# 베이지안 최적화 위한 함수
def black_box_function (k):
    config_data = {
        'k': k,
    }
    revenue = VBS(df,config_data)
    return revenue[0]

# parameter k의 범위
pbounds = {
    'k': (0.20,1.0)
}

# 베이지안 최적화 시행
optimizer = BayesianOptimization(f=black_box_function,pbounds=pbounds,random_state=1)
optimizer.maximize(init_points=5, n_iter=100)

# 결과값 저장
target_list = []
i=0

for res in optimizer.res: # res는 최적화 결과값을 리스트로 저장한 것.
    print(res)
    target_list.append([res["target"], i+1,res['params']]) # target값을 append
    i=i+1

target_list.sort(reverse=True)

target_list[0][-1].get('k')
# 값 추출

max_beyes_k = target_list[0][-1].get('k')

 

총 105번의 시도를 통해 k값을 최적화한다. 

 

 

max_beyes_k를 구하였으면 이를 변동성 돌파 전략 함수에 집어넣어서 결과를 도출하면 된다. 다만 config_data라고 유지하면 결과가 도출되지 않으므로 k값을 따로 선언하여야 하는 VBS_opt 함수를 새로 만들어 이를 활용하여 결과를 도출한다.

 

def VBS_opt(df, k):

    # 목표가 구하기
    df['range'] = df['High'] - df['Low']  # 고저 변동폭
    df['target'] = df['Open'] + df['range'].shift(1) * k  # 목표가 한칸 내려주고 (shift), 이후 k값을 곱한 목표가 계산
    # df = df.iloc[1:-2]

    # 매수 시뮬레이션
    df['ror'] = np.where(df['High'] > df['target'], df['Close'] / df['target'],
                         1)  # high > target인 지점에서 close / target

    # 최종 누적 산출
    df['total'] = df['ror'].cumprod()
    final_cum_ror = (df['total'].iloc[-1].astype(float)) * 100 -100

    # 연간 수익률 (기간 수익률)
    N = ((df.index[-1] - df.index[0])) / 365
    CAGR = (final_cum_ror ** (1 / N))

    # dd값 기록 및 mdd 계산
    array_v = np.array(df['total'])
    dd_list = -(np.maximum.accumulate(array_v) - array_v) / np.maximum.accumulate(array_v)
    peak_lower = np.argmax(np.maximum.accumulate(array_v) - array_v)
        # np.maximum.accumulate(array_v)는 최고값을 계속 갱신한 array
        # 그 array에서 원래 array를 뺀 값들의 array를 새로 생성
        # 뺀 값만 남은 array 중 가장 큰 값의 index를 추출
    peak_upper = np.argmax(array_v[:peak_lower])
        # peak_lower의 index값까지의 array_v값 추출
        # 그 중에 가장 큰 놈 (MDD 계산을 위한 최고값) 의 index 추출
    mdd = round((array_v[peak_lower] - array_v[peak_upper]) / array_v[peak_upper] * 100, 3)

    return final_cum_ror, CAGR, mdd, dd_list,df['total'] # 최종 수익률, 연간 수익률, 최대손실폭 (mdd), 손실 목록, 누적 이익 목록

final_result = VBS_opt(df,max_beyes_k)

print("최종 누적 수익률: ", final_result[0] ,"%") # 최종 누적 수익률 도출
print("MDD: ", final_result[2], "%") # mdd 도출
print("최선의 K값: ",max_beyes_k)

 

 

파이썬 파일을 실행하면 이런 식으로 최적화가 진행되며 

 

그 결과 최적의 k값이 나온다. (중간에 target, params 가 나오는 것은 res를 출력했기 때문이다.) 최적의 k값은 그리드 서치를 통해 나온 결과와 상당히 유사하였다.

 

 

다음 포스팅에서는 이러한 결과를 시각화해보고자 한다.

 

전체 코드:

 

import pandas as pd
import numpy as np
import random as rand
from bayes_opt import BayesianOptimization
import matplotlib.pyplot as plt

################ 백테스트 ################

# 백테스트 기본 셋팅

symbol = "BTCUSDT" # 다른 티커도 가능
interval = '1d' # 1m, 3m, 5m, 15m, 30m, 1h, 2h, 4h, 6h, 8h, 12h, 1d, 3d, 1w, 1M
year = 'All' # 2017~2022 중 선택 가능.

# 데이터 읽어오기: 전체 데이터
df = pd.read_csv (f'C:.\\Data\\{symbol[:-4]}-USDT_Data\\{symbol[:-4]}-USDT_Data_{interval}\\{symbol[:-4].upper()}_USDT_{interval}_{year}.csv')
# 전체: _All
# 기간: 연도 불러오면 됨
df = pd.DataFrame(df)
df = df.dropna(how='any')

# 변동성 돌파 전략 위한 로직 설정

def VBS(df, config_data):
    # k값 넣기
    k = config_data['k']

    # 목표가 구하기
    df['range'] = df['High'] - df['Low']  # 고저 변동폭
    df['target'] = df['Open'] + df['range'].shift(1) * k  # 목표가 한칸 내려주고 (shift), 이후 k값을 곱한 목표가 계산
    # df = df.iloc[1:-2]

    # 매수 시뮬레이션
    df['ror'] = np.where(df['High'] > df['target'], df['Close'] / df['target'],
                         1)  # high > target인 지점에서 close / target

    # 최종 누적 산출
    df['total'] = df['ror'].cumprod()
    final_cum_ror = (df['total'].iloc[-1].astype(float)) * 100 -100

    # 연간 수익률 (기간 수익률)
    N = ((df.index[-1] - df.index[0])) / 365
    CAGR = (final_cum_ror ** (1 / N))

    # dd값 기록 및 mdd 계산
    array_v = np.array(df['total'])
    dd_list = -(np.maximum.accumulate(array_v) - array_v) / np.maximum.accumulate(array_v)
    peak_lower = np.argmax(np.maximum.accumulate(array_v) - array_v)
        # np.maximum.accumulate(array_v)는 최고값을 계속 갱신한 array
        # 그 array에서 원래 array를 뺀 값들의 array를 새로 생성
        # 뺀 값만 남은 array 중 가장 큰 값의 index를 추출
    peak_upper = np.argmax(array_v[:peak_lower])
        # peak_lower의 index값까지의 array_v값 추출
        # 그 중에 가장 큰 놈 (MDD 계산을 위한 최고값) 의 index 추출
    mdd = round((array_v[peak_lower] - array_v[peak_upper]) / array_v[peak_upper] * 100, 3)

    return final_cum_ror, CAGR, mdd, dd_list,df['total'] # 최종 수익률, 연간 수익률, 최대손실폭 (mdd), 손실 목록, 누적 이익 목록

################ 최적화 ################

### 베이지안 최적화

# 베이지안 최적화 위한 함수
def black_box_function (k):
    config_data = {
        'k': k,
    }
    revenue = VBS(df,config_data)
    return revenue[0]

# parameter k의 범위
pbounds = {
    'k': (0.20,1.0)
}

# 베이지안 최적화 시행
optimizer = BayesianOptimization(f=black_box_function,pbounds=pbounds,random_state=1)
optimizer.maximize(init_points=5, n_iter=100)

# 결과값 저장
target_list = []
i=0

for res in optimizer.res: # res는 최적화 결과값을 리스트로 저장한 것.
    print(res)
    target_list.append([res["target"], i+1,res['params']]) # target값을 append
    i=i+1

target_list.sort(reverse=True)

target_list[0][-1].get('k')
# 값 추출

max_beyes_k = target_list[0][-1].get('k')

def VBS_opt(df, k):

    # 목표가 구하기
    df['range'] = df['High'] - df['Low']  # 고저 변동폭
    df['target'] = df['Open'] + df['range'].shift(1) * k  # 목표가 한칸 내려주고 (shift), 이후 k값을 곱한 목표가 계산
    # df = df.iloc[1:-2]

    # 매수 시뮬레이션
    df['ror'] = np.where(df['High'] > df['target'], df['Close'] / df['target'],
                         1)  # high > target인 지점에서 close / target

    # 최종 누적 산출
    df['total'] = df['ror'].cumprod()
    final_cum_ror = (df['total'].iloc[-1].astype(float)) * 100 -100

    # 연간 수익률 (기간 수익률)
    N = ((df.index[-1] - df.index[0])) / 365
    CAGR = (final_cum_ror ** (1 / N))

    # dd값 기록 및 mdd 계산
    array_v = np.array(df['total'])
    dd_list = -(np.maximum.accumulate(array_v) - array_v) / np.maximum.accumulate(array_v)
    peak_lower = np.argmax(np.maximum.accumulate(array_v) - array_v)
        # np.maximum.accumulate(array_v)는 최고값을 계속 갱신한 array
        # 그 array에서 원래 array를 뺀 값들의 array를 새로 생성
        # 뺀 값만 남은 array 중 가장 큰 값의 index를 추출
    peak_upper = np.argmax(array_v[:peak_lower])
        # peak_lower의 index값까지의 array_v값 추출
        # 그 중에 가장 큰 놈 (MDD 계산을 위한 최고값) 의 index 추출
    mdd = round((array_v[peak_lower] - array_v[peak_upper]) / array_v[peak_upper] * 100, 3)

    return final_cum_ror, CAGR, mdd, dd_list,df['total'] # 최종 수익률, 연간 수익률, 최대손실폭 (mdd), 손실 목록, 누적 이익 목록

final_result = VBS_opt(df,max_beyes_k)

print("최종 누적 수익률: ", final_result[0] ,"%") # 최종 누적 수익률 도출
print("MDD: ", final_result[2], "%") # mdd 도출
print("최선의 K값: ",max_beyes_k)