【Python + 仮想通貨】ボリンジャーバンドを活用した売買アルゴリズムをOptunaで最適化する

import pandas as pd

# data準備
file = '1min_20230806.csv'
df = pd.read_csv(file)
df['date'] = pd.to_datetime(df['date'])
df.set_index('date', inplace=True)

# 5分足に変更
th = '5T'
df_resampled = pd.DataFrame()
df_resampled['open'] = df['open'].resample(th).first()
df_resampled['high'] = df['high'].resample(th).max()
df_resampled['low'] = df['low'].resample(th).min()
df_resampled['close'] = df['close'].resample(th).last()
df_resampled['volume'] = df['volume'].resample(th).sum()

df = df_resampled

window = 20
df['MA'] = df['close'].rolling(window).mean()
df['std'] = df['close'].rolling(window).std()

# パラメータ設定
c = 2           # 買い判断に使う標準偏差の係数
sell = 1.015    # 買付価格の101.5%で利確売り
loss_cut = 0.99 # 買付価格の99%でロスカット

# 買い判断のデータインデックスを取得
df['buy_th'] = df['MA'] - df['std'] * c
df['diff'] = df['close'] - df['buy_th']
df['buy_signal'] = df['diff'].apply(lambda x: True if x < 0 else False)
buy_signal_idx = df[df['buy_signal']].index.tolist()

profits = []
loss_cuts = []
time_out_count = 0

for idx in buy_signal_idx:
    # 買付価格から利確、ロスカット価格を計算
    buy_price = df.loc[idx, 'close']
    sell_price = buy_price * sell
    loss_cut_price = buy_price * loss_cut
    
    # 買い判断以降の行を取得
    data = df.loc[idx:,]
    data = data.reset_index()
    close_idx = data.columns.get_loc('close')

    # 利確、ロスカット判断のbool形式カラムを追加
    data['profit'] = data['high'].apply(lambda x: True if x > sell_price else False)
    data['loss_cut'] = data['close'].apply(lambda x: True if x < loss_cut_price else False)

    # 利確、ロスカット判断がいずれかにTrueがある場合
    if any(data['loss_cut']) or any(data['profit']):

        # 最初に利確判断/ロスカット判断になるインデックス値を取得
        try:
            first_profit = data[data['profit']].index.tolist()[0]
        except IndexError: # すべてFalseだとIndexErrorになるので対策
            first_profit = len(data)
        try:
            first_loss_cut = data[data['loss_cut']].index.tolist()[0]
        except IndexError:
            first_loss_cut = len(data)

    else:
        time_out_count += 1
        continue
   
    # 利確/ロスカットの早い方で損益をリストに格納
    if first_profit <= first_loss_cut:
        profits.append(sell_price - buy_price)
        profit_after_buy.append(first_profit)
    
    elif first_loss_cut < first_profit:
        loss_cut_price = data.iloc[first_loss_cut, close_idx] # 終値でロスカット成立と仮定
        loss_cuts.append(loss_cut_price - buy_price)
        loss_cut_after_buy.append(first_loss_cut)

# 損益を通算
profit = sum(profits)
loss_cut = sum(loss_cuts)
total_profit = profit + loss_cut

# 結果
print(f"buy signal rate: {len(df[df['buy_signal']]) / len(df)}")
print(f"time outs: {time_out_count}/{len(df[df['buy_signal']])}")
print(f'profit: {profit}, {len(profits)}')
print(f'loss_cut: {loss_cut}, {len(loss_cuts)}')
print(f'total_profit: {total_profit}')

>> buy signal rate: 0.053835021707670044
>> time outs: 29/186
>> profit: 4025765.429999973, 65
>> loss_cut: -4175854.0, 92
>> total_profit: -150088.57000002684

def objective(trial):

    # data準備
    file = './data/5min_20230806.csv' # ５分足で移動平均、標準偏差の計算まで処理済み
    df = pd.read_csv(file)

    # best parameterで
    c = trial.suggest_uniform('c', 0.7, 3)
    sell = trial.suggest_uniform('sell', 1.00005, 1.01) 
    loss_cut = trial.suggest_uniform('loss_cut', 0.99, 1)

    # 結果の一時保存
    profits = []
    loss_cuts = []

    # 買いシグナルのindexを取得
    df['buy_th'] = df['MA'] - df['std'] * c
    df['diff'] = df['close'] - df['buy_th']
    df['buy_signal'] = df['diff'].apply(lambda x: True if x < 0 else False)
    buy_signal_idx = df[df['buy_signal']].index.tolist()

    for idx in buy_signal_idx:
        buy_price = df.loc[idx, 'close']
        sell_price = buy_price * sell
        loss_cut_price = buy_price * loss_cut
        
        data = df.loc[idx:,]
        data = data.reset_index()

        data['profit'] = data['high'].apply(lambda x: True if x > sell_price else False)
        data['loss_cut'] = data['close'].apply(lambda x: True if x < loss_cut_price else False)

        if any(data['loss_cut']) or any(data['profit']):
            try:
                first_profit = data[data['profit']].index.tolist()[0]
            except IndexError:
                first_profit = len(data)

            try:
                first_loss_cut = data[data['loss_cut']].index.tolist()[0]
            except IndexError:
                first_loss_cut = len(data)
        
        else:
            continue

        if first_profit <= first_loss_cut:
            profits.append(sell_price - buy_price)
        
        elif first_loss_cut < first_profit:
            loss_cut_price = data.loc[first_loss_cut, 'close']
            loss_cuts.append(loss_cut_price - buy_price)

    # 損益を通算
    total_profit = sum(profits) + sum(loss_cuts)

    return total_profit

import optuna

study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=200)

print(study.best_params)
print(study.best_value)

>> {'c': 0.7447882343598757, 'sell': 1.0044986849434376, 'loss_cut': 0.9983094867526481}
>> 2250214.5933372173

import plotly.express as px

results = []
for each in study.trials:
    result_dict = each.params
    result_dict['profit'] = each.value
    results.append(result_dict)

result_df = pd.DataFrame(results)
result_df.to_csv('optimized_bollinger_5min_20230806.csv')

# 3D plot
fig = px.scatter_3d(result_df, x='c', y='sell', z='loss_cut', color='profit')
fig.show()