因果推論とアルゴリズム取引における回帰モデルの活用

はじめに

これまでの一連の記事では、時系列データを分類するさまざまな手法について取り上げてきましたが、回帰モデルについては扱ってきませんでした。回帰モデルは、二値分類とは異なり、ある観測値が特定のクラスに属する確率ではなく、連続値そのものを予測できます。そのため、自動売買システムの構築への活用範囲を大きく広げます。

二値分類は、入力データを2つの異なるカテゴリまたはクラスのいずれかに分類することを目的とした、機械学習の基本的なタスクです。FXの自動売買システムでは、一般的に「買い」（0）または「売り」（1）のシグナルを予測することを意味します。このアプローチでは、複雑な市場の動きを単純な売買方向の判断へと単純化します。

しかし、二値分類を定量的なトレードへ適用する際の最も本質的な制約は、予測される価格変動の大きさや強さを定量化できないことです。二値分類モデルは、価格が上昇するか下降するかだけを示し、その変動幅がどの程度になるかについての情報は提供しません。この粒度の不足が、トレードにおける意思決定を根本的に制限します。

分類器の予測精度だけでは価格変動の大きさは考慮されないため、取引判断の指標としては十分とはいえません。この点は非常に重要です。なぜなら、方向予測の精度が高い（たとえば70%の確率で方向を正しく予測できる）からといって、必ずしも利益につながるわけではないからです。 

重要なのは、高い方向予測精度が収益性を保証するわけではないという事実です。たとえば、方向予測の正解率が30%しかなくても利益を上げられる場合がある一方で、70%の正解率を達成していても損失となる場合があります。これは、取引戦略の最終的な収益が、勝ちトレードによる利益額と負けトレードによる損失額とのバランスによって決まり、単純な勝率だけでは決まらないことを示しています。

二値分類モデルでは、実際の価格変動幅に関係なく、方向予測が正しければすべて同じ「正解」として扱われます。そのため、小さくほとんど利益にならない価格変動と、大きな利益をもたらす価格変動とを区別できません。その結果、多数の小さな利益が少数の大きな損失によって打ち消されたり、その逆の状況が発生したりして、一見すると高い予測精度を示していても、最終的なPnL（損益）はマイナスになる可能性があります。

価格変動の大きさを定量化できないということは、自動売買システムが期待利益の大きい取引を優先したり、価格の方向は正しく予測できていても期待利益より潜在的損失が大きい取引を回避したりすることができないことを意味します。価格変動幅に関する情報がないため、自動売買システムは、自身の判断が実際の損益にどのように影響するかを十分に把握できず、高い方向予測精度を達成していても、累積リターンが最適化されない、あるいはマイナスになる可能性があります。

ラベリング関数の変更

次のようなシナリオを考えてみましょう。特徴量の集合に基づいて予測すべき金融時系列データがあるとします。二値分類の場合、将来の取引方向（買いまたは売り）を決定でき、これらのラベルは常に固定されています。将来の価格変動幅をより正確に評価するために、ラベル付けの方法を変更することはできません。実際の価格変動幅にかかわらず、ラベリング上はすべての取引が等価なものとして扱われます。

一方、取引の方向だけでなく、将来の価格変動の大きさも予測できるとしたらどうでしょうか。これにより、取引システムをより細かく調整できるようになります。たとえば、追加のフィルターを作成することで、取引上意味のあると予測される価格変動のみを抽出し、重要性の低い価格変動を除外できるようになります。

回帰モデルを学習させるには、学習用の特徴量と目的変数を準備する必要があります。特徴量は分類モデルと回帰モデルで共通のものを使用できますが、目的変数はそれぞれ異なります。

それでは、回帰モデル向けのラベリングの一例を実装したシンプルな関数を作成してみましょう。

@njit
def calculate_labels_r(close_data, min_val, max_val):
    labels = []
    for i in range(len(close_data) - max_val):
        rand = random.randint(min_val, max_val)
        labels.append(close_data[i + rand] - close_data[i])
    return labels

def get_labels_r(dataset, min = 1, max = 15) -> pd.DataFrame:
    # Extract closing prices from the dataset
    close_data = dataset['close'].values
    labels = calculate_labels_r(close_data, min, max)
    # Trim the dataset to match the length of calculated labels
    dataset = dataset.iloc[:len(labels)].copy() 
    # Add the calculated labels as a new column
    dataset['labels'] = labels
    # Remove rows with NaN values (potentially introduced in 'calculate_labels')
    dataset = dataset.dropna()
    return dataset

二値分類用のラベリングとの主な違いは、単に方向（買いまたは売り）を判定するのではなく、価格変動を求める点にあります。具体的には、将来の価格から現在の価格を差し引くことで価格変動を算出します。このコードはNumbaによって高速化されているため、ターゲットのラベリングは非常に高速に実行されます。

上記の関数では、{min_val; max_val}の範囲内でランダムに選択された将来の価格と現在価格との差のみを考慮しています。しかし、この方法は必ずしも適切とはいえません。なぜなら、途中で発生する価格変動のうち、取引上重要となる可能性のあるものが考慮されていないためです。そこで、価格変動の計算方法を改良した別の関数を提案します。以下にその実装を示します。

@njit
def calculate_labels_mean_r(close_data, min_val, max_val):
    labels = []
    for i in range(len(close_data) - max_val):
        # Calculate the average price value in the window from min_val to max_val
        future_prices = close_data[i + min_val : i + max_val + 1]
        mean_future_price = np.mean(future_prices)
        # Calculate the difference between the average future value and the current price
        labels.append(mean_future_price - close_data[i])
    return labels

def get_labels_r(dataset, min = 1, max = 15) -> pd.DataFrame:
    # Extract closing prices from the dataset
    close_data = dataset['close'].values
    # Calculate buy/hold labels based on future price movements
    labels = calculate_labels_mean_r(close_data, min, max)
    # Trim the dataset to match the length of calculated labels
    dataset = dataset.iloc[:len(labels)].copy() 
    # Add the calculated labels as a new column
    dataset['labels'] = labels
    # Remove rows with NaN values (potentially introduced in 'calculate_labels')
    dataset = dataset.dropna()
    return dataset

この関数では、指定された区間内のすべての価格変動を考慮し、それらの平均値を計算します。その後、将来価格の平均値と現在価格との差を算出します。これに伴い、get_labels_r()関数では、従来のcalculate_labels_r()ではなく、新たにcalculate_labels_mean_r()ラベリング関数を呼び出すようになります。用途に応じて、異なるラベリング関数を呼び出して比較・検証することも可能です。

因果推論に基づく仕組みの追加

予測精度をさらに向上させるため、因果推論に関する記事で紹介したものと類似したアルゴリズムを使用します。主な違いは、分類器ではなく回帰モデルを使用する点です。

def meta_learners(data, models_number: int, iterations: int, depth: int):
    data = data.copy()
    data = data[(data.index < hyper_params['forward']) & (data.index > hyper_params['backward'])].copy()

    X = data[data.columns[1:-1]]
    y = data['labels']
    data['meta_labels'] = 0

    for i in range(models_number):
        X_train, X_val, y_train, y_val = train_test_split(
            X, y, train_size = 0.5, test_size = 0.5, shuffle = True)
        
        # fit debias model with train and validation subsets
        meta_m = CatBoostRegressor(iterations = iterations,
                                depth = depth,
                                verbose = False,
                                use_best_model = True)
        
        meta_m.fit(X_train, y_train, eval_set = (X_val, y_val), plot = False)
        
        coreset = X.copy()
        coreset['labels'] = y
        coreset['labels_pred'] = meta_m.predict(X)
        data['meta_labels'] += abs(coreset['labels'] - coreset['labels_pred'])

    data['meta_labels'] = data['meta_labels'] / models_number
    return data

この関数では、元のデータセットからランダムに抽出したデータサブセットを用いて複数の回帰モデルを学習させ、その後、実際のターゲット値と予測値を比較します。その結果、メタモデルは0または1（取引する/しない）を予測するのではなく、予測値と実測値のずれの平均的な大きさを予測します。これにより、期待値から大きく外れた予測をフィルタリングできるようになります。

学習済みモデルの評価とテスト

回帰モデルをテストするため、テスターを修正しました。回帰モデル用のテスターには、名称の末尾に「r」が付加されています。それでは、複数のモデルを学習させてみましょう。この記事では、10個のモデルを学習させ、その中から最も適切だと判断したモデルを選択します。

hyper_params = {
    'symbol': 'EURUSD_H1',
    'export_path': '/Users/dmitrievsky/drive_c/Program Files/MetaTrader 5/MQL5/Include/Trend following/',
    'model_number': 0,
    'markup': 0.00010,
    'stop_loss':  0.00500,
    'take_profit': 0.00200,
    'periods': [i for i in range(5, 100, 30)],
    'backward': datetime(2010, 1, 1),
    'forward': datetime(2024, 1, 1),
}

models = []
for i in range(10):
    print('Learn ' + str(i) + ' model')
    data = get_labels_r(get_features(get_prices()), min=1, max=15)
    dataset = meta_learners(data=data, models_number=5, iterations=15, depth=3)
    models.append(fit_final_models(dataset, tol=3e-2))

ここで注目すべきなのは、最終モデルの学習関数に渡されるtolパラメータです。メインモデルの頑健性を可能な限り高めることが目的であるため、すべての学習サンプルを使用して学習させる意味はありません。その代わりに、予測値と実際の値との誤差がtolの値未満であるサンプルのみを使用して学習をおこないます。

予測誤差は実際にはポイント単位で計算されるため、tol=3e-2は最大誤差が0.03、すなわち4桁価格表示で300ポイントであることを意味します。フィルターとしてはかなり大きな値に思えるかもしれませんが、ここで扱うのは絶対値の差である点に注意が必要です。予測値は正にも負にもなり得るため、この設定にも一定の妥当性があります。このパラメータについては、さまざまな値を試しながら最適な設定を検討するとよいでしょう。以下に、この関数を示します。

def fit_final_models(dataset, tol=1e-2) -> list:
    # features for model\meta models. We learn main model only on filtered labels
    X = dataset[dataset['meta_labels'] < tol]
    X, X_meta = X[X.columns[1:-2]], dataset[dataset.columns[1:-2]]
    # labels for model\meta models
    y = dataset[dataset['meta_labels'] < tol]
    y, y_meta = y[y.columns[-2]], dataset[dataset.columns[-1]]
    
    # fit main model with train and validation subsets
    model = RandomForestRegressor(n_estimators=50, max_depth=10)
    model.fit(X, y)
    # fit meta model with train and validation subsets
    meta_model = RandomForestRegressor(n_estimators=50, max_depth=10)
    meta_model.fit(X_meta, y_meta)

    data = get_features(get_prices())
    R2 = test_model_r(data, 
                    [model, meta_model], 
                    hyper_params['stop_loss'], 
                    hyper_params['take_profit'],
                    hyper_params['forward'],
                    hyper_params['backward'],
                    hyper_params['markup'],
                    plt=False)
    
    if math.isnan(R2):
        R2 = -1.0
        print('R2 is fixed to -1.0')
    print('R2: ' + str(R2))
    result = [R2, model, meta_model]
    return result

それでは、モデルをソートしてカスタムテスター関数を呼び出しましょう。

models.sort(key=lambda x: x[0])
data = get_features(get_prices())
test_model_r(data, 
        models[-1][1:], 
        hyper_params['stop_loss'], 
        hyper_params['take_profit'],
        hyper_params['forward'],
        hyper_params['backward'],
        hyper_params['markup'],
        plt=True)

このモデルは過学習を起こしており、新しいデータに対しては性能が低下します。

図1：基本的なラベリングを使用したモデルのテスト

同じ手順を実行しますが、今回は将来価格の平均値を計算するcalculate_labels_mean_r()取引ラベリング関数を使用します。

図2：平均化ラベリングを使用したモデルのテスト

2つ目の取引ラベリング関数は、新しいデータに対して平均的により安定した結果を示しました。これは、将来価格の平均値を考慮していることが要因であると考えられます。

カスタムテスターには、メインの回帰モデルに対して閾値を設定する機能がありません。そのため、モデルの予測値を単純に正と負に分類するだけであり、シグナルは依然として大まかなものになっています。しかし、この点についてはMetaTrader 5ターミナル上で対応します。

MetaTrader 5ターミナルへのモデルのエクスポート

次に、モデルをONNX形式でターミナルへエクスポートし、取引システムを設定する必要があります。エクスポート関数は、これまでと同様の構成になっています。

export_model_to_ONNX(model = models[-1],
                     symbol = hyper_params['symbol'],
                     periods = hyper_params['periods'],
                     periods_meta = hyper_params['periods'],
                     model_number = hyper_params['model_number'],
                     export_path = hyper_params['export_path'])

なお、CatBoostの回帰モデルは、ONNX形式ではMetaTrader 5ターミナルで使用できなかったため、代わりにRandom Forestを使用しました。

入力テンソルの次元は、学習開始前に指定したハイパーパラメータ（特徴量数）に応じて自動的に調整されます。その後、convert_sklearn()関数を使用してモデルをONNX形式へ変換し、ハイパーパラメータで指定したディレクトリに保存します。

def export_model_to_ONNX(**kwargs):
    model = kwargs.get('model')
    symbol = kwargs.get('symbol')
    periods = kwargs.get('periods')
    periods_meta = kwargs.get('periods_meta')
    model_number = kwargs.get('model_number')
    export_path = kwargs.get('export_path')

    initial_type = [('float_input', FloatTensorType([None, len(hyper_params['periods'])]))]
    onnx_model = convert_sklearn(model[1], initial_types=initial_type)
    # save main model to ONNX
    with open(export_path +'catmodel ' + symbol + ' ' + str(model_number) +'.onnx', "wb") as f:
        f.write(onnx_model.SerializeToString())
    onnx_model_meta = convert_sklearn(model[2], initial_types=initial_type)
    # save meta model to ONNX
    with open(export_path +'catmodel_m ' + symbol + ' ' + str(model_number) +'.onnx', "wb") as f:
        f.write(onnx_model_meta.SerializeToString())
    
    code = '#include <Math\Stat\Math.mqh>'
    code += '\n'
    code += '#resource "catmodel '+ symbol + ' '+str(model_number)+'.onnx" as uchar ExtModel_' + symbol + '_' + str(model_number) + '[]'
    code += '\n'
    code += '#resource "catmodel_m '+ symbol + ' '+str(model_number)+'.onnx" as uchar ExtModel2_' + symbol + '_' + str(model_number) + '[]'
    code += '\n\n'
    code += 'int Periods' + symbol + '_' + str(model_number) + '[' + str(len(periods)) + \
        '] = {' + ','.join(map(str, periods)) + '};'
    code += '\n'
    code += 'int Periods_m' + symbol + '_' + str(model_number) + '[' + str(len(periods_meta)) + \
        '] = {' + ','.join(map(str, periods_meta)) + '};'
    code += '\n\n'

    # get features
    code += 'void fill_arays' + symbol + '_' + str(model_number) + '( double &features[]) {\n'
    code += '   double pr[], ret[];\n'
    code += '   ArrayResize(ret, 1);\n'
    code += '   for(int i=ArraySize(Periods'+ symbol + '_' + str(model_number) + ')-1; i>=0; i--) {\n'
    code += '       CopyClose(NULL,PERIOD_H1,1,Periods' + symbol + '_' + str(model_number) + '[i],pr);\n'
    code += '       ret[0] = MathStandardDeviation(pr);\n'
    code += '       ArrayInsert(features, ret, ArraySize(features), 0, WHOLE_ARRAY); }\n'
    code += '   ArraySetAsSeries(features, true);\n'
    code += '}\n\n'

    # get features
    code += 'void fill_arays_m' + symbol + '_' + str(model_number) + '( double &features[]) {\n'
    code += '   double pr[], ret[];\n'
    code += '   ArrayResize(ret, 1);\n'
    code += '   for(int i=ArraySize(Periods_m' + symbol + '_' + str(model_number) + ')-1; i>=0; i--) {\n'
    code += '       CopyClose(NULL,PERIOD_H1,1,Periods_m' + symbol + '_' + str(model_number) + '[i],pr);\n'
    code += '       ret[0] = MathStandardDeviation(pr);\n'
    code += '       ArrayInsert(features, ret, ArraySize(features), 0, WHOLE_ARRAY); }\n'
    code += '   ArraySetAsSeries(features, true);\n'
    code += '}\n\n'

    file = open(export_path + str(symbol) + ' ONNX include' + ' ' + str(model_number) + '.mqh', "w")
    file.write(code)

    file.close()
    print('The file ' + 'ONNX include' + '.mqh ' + 'has been written to disk')

MetaTrader 5ターミナルでの閾値の設定

2つの分類モデルの代わりに2つの回帰モデルを使用するようになったことで、具体的な閾値を設定できるようになりました。

図3：MetaTrader 5ターミナルでのシグナル発生閾値の設定

• buy_thresholdおよびsell_thresholdは、メインの回帰モデルが出力するシグナルをフィルタリングするためのパラメータです。予測値がこの閾値を下回る場合は、ポジションは建てられません。たとえば、予測される価格変動が10pips未満であれば、スプレッドや手数料を考慮すると利益が見込めないため、そのような取引をおこなう合理性はほとんどありません。
• meta_threshholdは、前述した因果推論の仕組みに基づいて、メインモデルのシグナルをさらにフィルタリングします。このパラメータは、予測値が将来の実際の価格変動からどの程度乖離する可能性があるかを評価します。予測誤差が大きすぎると判断された場合、その取引も実行されません。

それでは、これらの閾値を設定した状態で、MetaTrader 5のストラテジーテスターを用いてモデルをテストしてみます。

図4：指定閾値を用いたモデルのテスト

フォワード期間は2024年初頭から開始していることを改めて確認しておきます。このモデルは、そのフォワード期間においても比較的安定した性能を示しています。この結果は、適切な閾値を定義し、設定することの重要性を示しています。閾値については、読者自身で最適化を試してみることもできます。

利用可能なモデルの組み合わせは、特徴量の種類（本記事では両方のモデルとも標準偏差を特徴量として使用）や、モデル自体のパラメータ設定によって大きく異なります。たとえば、異なるパラメータで学習した別のモデルでは、閾値を調整しなくても、新しいデータに対して良好な性能を示しました。

図5：異なる学習パラメータを用いて学習・テストした別モデル

閾値を調整した後、このモデルは2024年初頭以降も安定した成長を示しました。

図6：閾値設定後のMetaTrader 5ターミナルにおけるモデルのテスト

さらに、第2の取引ラベリング関数を用いて学習し、閾値によるフィルタリングを組み合わせると、より興味深く、精度の高い結果が得られました。

図7：平均取引ラベリング関数とtol = 1e-2を用いたモデルのテスト

さらに、学習時のtolパラメータを1e-2から1e-3へ変更すると、結果は一段と改善されます。

図8：平均取引ラベリング関数とtol = 1e-3を用いたモデルのテスト

補足情報

モデルをエクスポートするには、まずskl2onnxパッケージをインストールし、インポートする必要があります。

from skl2onnx import convert_sklearn
from skl2onnx.common.data_types import FloatTensorType

自動売買ボットのコード内にあるONNXモデルの起動処理も、新しい回帰モデルを正しく処理できるように変更されています。

vectorf y_main(1), y_meta(1);
   
OnnxRun(ExtHandle, ONNX_DEBUG_LOGS, f, y_main);
OnnxRun(ExtHandle2, ONNX_DEBUG_LOGS, f_m, y_meta);

float sig = y_main[0];
float meta_sig = y_meta[0];

閾値の調整と最適化のための新しい入力変数を追加しました。

input double buy_threshold = 0.00001;
input double sell_threshold = -0.00001;
input double meta_threshold = 0.001;

 ONNXモデル名は#defineディレクティブを介してアクセスできるようになり、異なる名前のモデルを簡単に含めることができるようになりました。

#define model ExtModel_EURUSD_H1_0
#define model_m ExtModel2_EURUSD_H1_0
#define periods PeriodsEURUSD_H1_0
#define periods_m Periods_mEURUSD_H1_0
#define fill_arrays fill_araysEURUSD_H1_0
#define fill_arrays_m fill_arays_mEURUSD_H1_0

売買シグナルは、閾値に応じて条件が満たされたときに生成されます。

if((Ask-Bid < max_spread*_Point) && MathAbs(meta_sig) < meta_threshold &&
      AllowTrade(OrderMagic))
      if(countOrders(OrderMagic) < max_orders &&
         CheckMoneyForTrade(_Symbol, LotsOptimized(), ORDER_TYPE_BUY))
        {
         double l = LotsOptimized();
         if(sig > buy_threshold && Allow_Buy)
           {
            int res = -1;
            do
              {
               double stop = Bid - stoploss * _Point;
               double take = Ask + takeprofit * _Point;
               res = mytrade.PositionOpen(_Symbol, ORDER_TYPE_BUY, l, Ask, stop, take, bot_comment);
               Sleep(50);
              }
            while(res == -1);
           }
         else
           {
            if(sig < sell_threshold && Allow_Sell)
              {
               int res = -1;
               do
                 {
                  double stop = Ask + stoploss * _Point;
                  double take = Bid - takeprofit * _Point;
                  res = mytrade.PositionOpen(_Symbol, ORDER_TYPE_SELL, l, Bid, stop, take, bot_comment);
                  Sleep(50);
                 }
               while(res == -1);
              }
           }
        }

修正されたストラテジーテスターおよびモデルエクスポート関数は、対応するモジュールに追加され、記事に添付されています。

結論

本記事では、回帰モデルを基盤とした取引システムを構築するための可能な方法の一つを紹介しました。ただし、これは唯一の方法ではありません。このアプローチにより、機械学習を利用したボットをより細かく調整できるようになります。また、閾値を設定することで、本来は収益性が低いモデルを利益の出るモデルへ変換することも可能になります。任意の特徴量や取引ラベリング関数を使用でき、さらにこのアルゴリズムを他の金融商品や異なる時間足でテストすることもできます。

Python files.zipアーカイブには、Python環境向けの開発ファイルが含まれています。

ファイル名	説明
causal regression.py	モデル学習用のメインスクリプト
labeling_lib.py	更新された取引ラベリングモジュール
tester_lib.py	更新された機械学習ベースのカスタムストラテジーテスター
export_lib.py	モデルをターミナルにエクスポートするモジュール
EURUSD_H1.csv	MetaTrader 5からエクスポートされた価格データ

MQL5 files.zipアーカイブには、MetaTrader 5ターミナル用のファイルが含まれています。

ファイル名	説明
regression trader.ex5	記事で作成したボットのコンパイル済みファイル
regression trader.mq5	記事で使用したボットのソースコード
Include//Trend following folder	ONNXモデルおよびボット接続用ヘッダファイル