遺伝的アルゴリズムを用いた指標パラメータの最適化

本ブログの約一年前の記事で、backtesting.pyライブラリを用いたバックテストを紹介しました。
今回はそれと類似の話題ですが、遺伝的アルゴリズム(GA)を用いたパラメータ最適化をとりあげます。
遺伝的アルゴリズムとは、「進化論の考え方に基づいたアルゴリズム」のことで、データを遺伝子のように変形、選択、交叉、突然変異などを繰り返すことで最適解を探索するアルゴリズムです。
今回のコードはPythonのdeapライブラリを用いたもので、Medium英文記事をベースとしていますが、オリジナルの記事では移動平均のゴールデンクロス戦略を扱っているのに対して、本記事ではコードの汎用性を確認するために、
「RSI（相対力指数。窓の長さa）が、定数bを下から上にクロスしたら買い、定数100-bを上から下にクロスしたら売り」
という戦略に変更し、aとbをパラメータとして、シャープレシオを最大化する値を探索することにしました。

まず、必要なライブラリをインポートします。

import pandas as pd
import numpy as np
import yfinance as yf
from datetime import date, datetime, timedelta
from deap import base, creator, tools, algorithms
import talib
import random
import warnings
warnings.simplefilter(action="ignore", category=FutureWarning)
warnings.simplefilter(action="ignore", category=RuntimeWarning)

次は、上で述べたRSIの戦略を関数として記述します。

#目的関数で用いられる戦略の定義（RSI逆張り）
def rsi_strategy(data, n_window, margin_pct):
    n_window = int(n_window)
    margin_pct = int(margin_pct)
#   エラー処理
    if not isinstance(n_window, int) or not isinstance(margin_pct, int):
        raise ValueError("Window lengths must be integers")
    if n_window <= 2 or margin_pct >= 48:
        raise ValueError("Parameter Value out of Range")
#   データフレームの初期設定
    data = data.copy()
#    RSIの計算 (talibを利用）
    data["RSI"] = talib.RSI(data["Close"], timeperiod=n_window)
    data = data.dropna()
#   エラー処理
    if data.empty:
        return np.nan
#   買いシグナルで1,売りシグナルで-1をとる列を設定
    data['positions'] = 0
#   買いシグナル
    data.loc[(data['RSI'].shift(1) <= margin_pct ) & (data['RSI'] > margin_pct), 'positions'] = 1
#   売りシグナル
    data.loc[(data['RSI'].shift(1) >= 100-margin_pct) & (data['RSI'] < 100-margin_pct), 'positions'] = -1
#   エラー処理
    if data.empty:
        return np.nan
#   リターンは、ポジションをとった時点から1時点後までの損益の累計
    data['returns'] = data['Close'].pct_change().shift(-1)
    data['strategy_returns'] = data['returns'] * data['positions'].shift(1)
    data = data.dropna(subset=['strategy_returns'])
#   エラー処理
    if data.empty:
        return np.nan
#   シャープレシオの計算
    mean_return = data['strategy_returns'].mean()
    std_return = data['strategy_returns'].std()
    if std_return == 0:
        return np.nan
    sharpe_ratio = mean_return / std_return
    return sharpe_ratio

次に、deapライブラリで遺伝的アルゴリズムをセットアップする部分です。

# 遺伝的アルゴリズムのセットアップ
def setup_ga(data):
    #目的関数の定義。必ずreturnの後に,をつける
    def evaluate(individual):
        n_window, margin_pct = individual
        if (n_window<=2) or (margin_pct>=45):
            return -np.inf,
        try:
            sharpe = rsi_strategy(data, n_window, margin_pct)
            if np.isnan(sharpe):
                return -np.inf,
            return sharpe,
        except ValueError:
            return -np.inf,
    #最小化問題として設定(-1.0で最小化、1.0で最大化問題)
    creator.create("FitnessMax", base.Fitness, weights=(1.0,))
    #個体の定義（list型と指定、中身の遺伝子は後で入れる）
    creator.create("Individual", list, fitness=creator.FitnessMax)
    #各種関数の設定
    #交叉、選択、突然変異などには、DEAPのToolbox内にある関数を利用
    toolbox = base.Toolbox()
    #random.randintの別名をattr_n_window関数として設定。各個体の遺伝子の中身を決める関数(各遺伝子は7～17のランダムな値)
    toolbox.register("attr_n_window", random.randint, 7, 25) # パラメータの動く範囲を指定
    toolbox.register("attr_margin_pct", random.randint, 5, 45) # パラメータの動く範囲を指定
    #individualという関数を設定。それぞれの個体に含まれる1個の遺伝子をattr_n_windowとattr_margin_pctにより決めるよ、ということ。
    toolbox.register("individual", tools.initCycle, creator.Individual,
                     (toolbox.attr_n_window, toolbox.attr_margin_pct), n=1)
    #集団の個体数を設定するための関数を準備
    toolbox.register("population", tools.initRepeat, list, toolbox.individual)
    #交叉関数の設定。ブレンド交叉という手法を採用
    toolbox.register("mate", tools.cxBlend, alpha=0.5)
    #突然変異関数の設定。indpbは各遺伝子が突然変異を起こす確率。変異はlow[i]からup[i]までの整数で変異(i=0,1)
    toolbox.register("mutate", tools.mutUniformInt, low=[7, 5], up=[25, 45], indpb=0.2) # パラメータの動く範囲を指定
    #トーナメント方式で次世代に子を残す親を選択（tornsizeは各トーナメントに参加する個体の数）
    toolbox.register("select", tools.selTournament, tournsize=3)
    #評価したい関数の設定（目的関数のこと）
    toolbox.register("evaluate", evaluate)
    return toolbox

次は、市場価格データの獲得と遺伝的アルゴリズムの実行を関数として記述します。

# 市場価格データのダウンロード
def download_data(symbol, start_date, end_date):
    data = yf.download(symbol, start=start_date, end=end_date)
    data['Close'] = data['Adj Close']
    data = data.dropna()
    return data

# 遺伝的アルゴリズムの実行
def run_ga(toolbox, population_size=50, n_generations=20):
    population = toolbox.population(n=population_size)
    hall_of_fame = tools.HallOfFame(1)
    stats = tools.Statistics(lambda ind: ind.fitness.values)
    stats.register("avg", np.mean)
    stats.register("std", np.std)
    stats.register("min", np.min)
    stats.register("max", np.max)
    population, logbook = algorithms.eaSimple(population, toolbox, cxpb=0.5, mutpb=0.2,
                                              ngen=n_generations, stats=stats,
                                              halloffame=hall_of_fame, verbose=True)
    return hall_of_fame[0]

最後に、メイン関数とその実行です。

# メイン関数
def main(symbol, start_date, end_date):
    data = download_data(symbol, start_date, end_date)
    toolbox = setup_ga(data)
    best_individual = run_ga(toolbox)
    best_n_window, best_margin_pct = best_individual
    print(f"Best parameters: n_window={best_n_window}, margin_pct={best_margin_pct}")

# 全体の実行
if __name__ == "__main__":
    symbol = '9983.T'
    end_date = datetime(2024,7,20)
    start_date = end_date - timedelta(days=1095)
    main(symbol, start_date, end_date)