Dual Thrustの遺伝的最適化

コード WonderTrader

サマリー

この例では、Dual Thrust先物戦略のパラメータを探索する遺伝的アルゴリズムの最適化器を設定します。勝ちトレードと負けトレードそれぞれの平均結果から目的関数を定義し、固定パラメータと可変パラメータを割り当て、バックテスト環境と期間を設定して、複数のワーカーで探索を実行する方法を示します。別々の設定で、エントリー条件、損切り値、利益確定値、または損切りと利益確定の組み合わせを探索します。

この例では、戦略設定と集計結果を出力ファイルに書き込みます。目的関数は1取引あたりの平均利益と平均損失に基づく比率で、平均損失がゼロの場合には代替処理を用います。コードにはパラメータの範囲と1つの過去のテスト期間が示されていますが、得られた運用成績、ベースラインとの比較、アウトオブサンプル検証は報告されていません。そのため、遺伝的アルゴリズムの探索結果は、選択した評価指標、パラメータ範囲、市場データ、検証期間に左右されます。この例だけでは、最適化した設定が他の期間でも通用するとは言えません。

主なアイデア

遺伝的アルゴリズムは、候補の選択と変異を繰り返し、戦略パラメータの組み合わせを探索できます。
目的関数は、平均勝ちトレード結果と平均負けトレード結果の比率を使って候補を評価します。
固定、リスト制約付き、数値範囲の各パラメータで、異なる種類の戦略入力を設定できます。
エントリー条件、損切り、利益目標、またはリスク管理条件の組み合わせを個別に探索できます。
この例では過去のバックテスト期間を使いますが、アウトオブサンプル検証や最適化結果は報告していません。

# runGAOptimizer.py


```py
# -- coding: utf-8 --

from wtpy.apps.WtCtaGAOptimizer import WtCtaGAOptimizer

import sys
sys.path.append('../Strategies')
from DualThrust import StraDualThrust

# 适应度函数
def my_optimizing_target(summary: dict):  # 单目标优化
    """ 编写优化目标 """
    target_value = summary["单次盈利均值"] / abs(summary["单次亏损均值"]) if summary["单次亏损均值"] != 0 else summary["单次盈利均值"]
    return target_value,  # 必须返回tuple类型


def runBaseOptimizer():
    # 新建一个优化器，并设置工作进程数为2
    population_size = 100  # 种群数
    mu = 80  # 每代个体选取数
    ngen_size = 5  # 进化代数
    cx_prb = 0.9  # 交叉概率
    mut_prb = 0.005  # 变异概率
    optimizer = WtCtaGAOptimizer(worker_num=2,
                                 population_size=population_size,
                                 MU=mu,
                                 ngen_size=ngen_size,
                                 cx_prb=cx_prb,
                                 mut_prb=mut_prb)

    # 引入适应度函数
    target_name = "单次盈亏比率"  # 默认为None
    optimizer.set_optimizing_func(calculator=my_optimizing_target, target_name=target_name)

    # 设置要使用的策略，只需要传入策略类型即可，同时设置策略ID的前缀，用于区分每个策略的实例
    optimizer.set_strategy(StraDualThrust, "Dt_IF_")

    # 添加固定参数
    optimizer.add_fixed_param(name="barCnt", val=50)
    optimizer.add_fixed_param(name="period", val="m5")
    optimizer.add_fixed_param(name="days", val=30)
    optimizer.add_fixed_param(name="code", val="CFFEX.IF.HOT")

    # 添加预设范围的参数，即参数只能在预设列表中选择，适用于标的代码、周期等参数
    # optimizer.add_listed_param(name="code", val_list=["CFFEX.IF.HOT","CFFEX.IC.HOT"])

    # 添加可变参数，适用于一般数值类参数
    optimizer.add_mutable_param(name="k1", start_val=0.1, end_val=1.0, step_val=0.1, ndigits=1)
    optimizer.add_mutable_param(name="k2", start_val=0.1, end_val=1.0, step_val=0.1, ndigits=1)

    # 配置回测环境，主要是将直接回测的一些参数通过这种方式动态传递，优化器中会在每个子进程动态构造回测引擎
    optimizer.config_backtest_env(deps_dir='../common/', cfgfile='configbt.yaml', storage_type="csv",
                                  storage_path="../storage/")
    # optimizer.config_backtest_time(start_time=201909100930, end_time=202009251500)
    optimizer.config_backtest_time(start_time=201909260930, end_time=202010121500)

    # 启动优化器
    optimizer.go(out_marker_file="strategies.json", out_summary_file="total_summary.csv")


def runStopLossOptimizer():
    # 新建一个优化器，并设置最大工作进程数为8
    optimizer = WtCtaGAOptimizer(worker_num=2)

    # 设置要使用的策略，只需要传入策略类型即可，同时设置策略ID的前缀，用于区分每个策略的实例
    optimizer.set_strategy(StraDualThrust, "Dt_IF_SL_")

    # 添加固定参数
    optimizer.add_fixed_param(name="barCnt", val=50)
    optimizer.add_fixed_param(name="period", val="m5")
    optimizer.add_fixed_param(name="days", val=30)
    optimizer.add_fixed_param(name="code", val="CFFEX.IF.HOT")
    optimizer.add_fixed_param(name="k1", val=0.4)
    optimizer.add_fixed_param(name="k2", val=0.4)

    # 添加可变参数，适用于一般数值类参数
    optimizer.add_mutable_param(name="slTicks", start_val=-10, end_val=0, step_val=0.2, ndigits=1)

    # 配置回测环境，主要是将直接回测的一些参数通过这种方式动态传递，优化器中会在每个子进程动态构造回测引擎
    optimizer.config_backtest_env(deps_dir='../common/', cfgfile='configbt.yaml', storage_type="csv",
                                  storage_path="../storage/")
    optimizer.config_backtest_time(start_time=201909100930, end_time=202010121500)

    # 启动优化器
    optimizer.go(out_marker_file="strategies.json", out_summary_file="total_summary_sl.csv")


def runStopProfOptimizer():
    # 新建一个优化器，并设置最大工作进程数为8
    optimizer = WtCtaGAOptimizer(worker_num=4)

    # 设置要使用的策略，只需要传入策略类型即可，同时设置策略ID的前缀，用于区分每个策略的实例
    optimizer.set_strategy(StraDualThrust, "Dt_IF_SP_")

    # 添加固定参数
    optimizer.add_fixed_param(name="barCnt", val=50)
    optimizer.add_fixed_param(name="period", val="m5")
    optimizer.add_fixed_param(name="days", val=30)
    optimizer.add_fixed_param(name="code", val="CFFEX.IF.HOT")
    optimizer.add_fixed_param(name="k1", val=0.4)
    optimizer.add_fixed_param(name="k2", val=0.4)

    # 添加可变参数，适用于一般数值类参数
    optimizer.add_mutable_param(name="spTicks", start_val=0, end_val=500, step_val=5, ndigits=0)

    # 配置回测环境，主要是将直接回测的一些参数通过这种方式动态传递，优化器中会在每个子进程动态构造回测引擎
    optimizer.config_backtest_env(deps_dir='../common/', cfgfile='configbt.yaml', storage_type="csv",
                                  storage_path="../storage/")
    optimizer.config_backtest_time(start_time=201909100930, end_time=202010121500)

    # 启动优化器
    optimizer.go(out_marker_file="strategies.json", out_summary_file="total_summary_sp.csv")


def runStopAllOptimizer():
    # 新建一个优化器，并设置最大工作进程数为8
    optimizer = WtCtaGAOptimizer(worker_num=4)

    # 设置要使用的策略，只需要传入策略类型即可，同时设置策略ID的前缀，用于区分每个策略的实例
    optimizer.set_strategy(StraDualThrust, "Dt_IF_ALL_")

    # 添加固定参数
    optimizer.add_fixed_param(name="barCnt", val=50)
    optimizer.add_fixed_param(name="period", val="m5")
    optimizer.add_fixed_param(name="days", val=30)
    optimizer.add_fixed_param(name="code", val="CFFEX.IF.HOT")
    optimizer.add_fixed_param(name="k1", val=0.4)
    optimizer.add_fixed_param(name="k2", val=0.4)

    # 添加可变参数，适用于一般数值类参数
    optimizer.add_mutable_param(name="slTicks", start_val=-30, end_val=-10, step_val=1, ndigits=1)
    optimizer.add_mutable_param(name="spTicks", start_val=150, end_val=230, step_val=2, ndigits=1)

    # 配置回测环境，主要是将直接回测的一些参数通过这种方式动态传递，优化器中会在每个子进程动态构造回测引擎
    optimizer.config_backtest_env(deps_dir='../common/', cfgfile='configbt.yaml', storage_type="csv",
                                  storage_path="../storage/")
    optimizer.config_backtest_time(start_time=201909100930, end_time=202010121500)

    # 启动优化器
    optimizer.go(out_marker_file="strategies.json", out_summary_file="total_summary_all.csv")


if __name__ == "__main__":
    runBaseOptimizer()
    kw = input('press any key to exit\n')
```

出典を明記したうえで、ライセンスに従って全文を掲載しています。ライセンス: MIT

この要約は原文をもとにStratmillのリサーチエージェントが作成したもので、出典の複製ではありません。

遺伝的アルゴリズムによるDual Thrust戦略パラメータの最適化

サマリー

主なアイデア

タグ