MQL5とデータ処理パッケージの統合（第2回）：機械学習と予測分析

はじめに

この記事では、特に機械学習(ML)と予測分析に焦点を当てます。データ処理パッケージは、クオンツトレーダーや金融アナリストに新たな可能性をもたらします。MQL5に機械学習機能を組み込むことで、トレーダーは取引戦略を従来のルールベースのシステムから、変化する市場環境に持続的に適応する洗練されたデータ駆動型モデルへと進化させることができます。

このプロセスでは、Pythonの強力なデータ処理機能とscikit-learnのような機械学習ライブラリをMQL5と組み合わせて使用します。この統合により、トレーダーは過去のデータを用いて予測モデルを訓練し、バックテスト技術でその有効性を検証し、リアルタイムで取引判断を下すためにモデルを活用することができるようになります。これらのツールを柔軟に組み合わせることで、典型的なテクニカル指標を超えた戦略を構築し、予測分析やパターン認識を取り入れることで取引結果を大幅に向上させることができます。

履歴データの収集

まず、MetaTrader 5から.csv形式で保存された履歴データが必要です。MetaTraderプラットフォームを起動し、MetaTrader 5のパネルの上部にある[ツール]>[オプション]に移動し、[チャート]オプションに移動します。次に、ダウンロードするチャートのバー数を選択します。バーの数は無制限にするのが一番です。日付を扱うことになるし、ある期間にいくつのバーがあるかわからないからです。

その後、実際のデータをダウンロードします。そのためには、 [表示] に移動し、次に[銘柄] に移動して、[仕様] タブに移動し、ダウンロードするデータの種類に応じて[バー] または [ティック] に移動するだけです。ダウンロードする履歴データの開始日と終了日を入力し、リクエストボタンをクリックすると、データがダウンロードされ、.csv形式で保存されます。

以上の手順で、MetaTrader取引プラットフォームから履歴データをダウンロードすることができました。次に、分析用のJupyter Lab環境をダウンロードして設定する必要があります。Jupyter Labをダウンロードして設定するには、公式のwebサイトにアクセスし、簡単な手順に従ってダウンロードできます。使用しているOSの種類に応じてpip、conda、brewなど、さまざまなインストール方法が用意されています。

Jupyter LabでMetaTrader 5の履歴データを処理する

MetaTrader 5の履歴データをJupyter Labに正常にロードするには、データをダウンロードしたフォルダを把握しておくことが重要です。その後、Jupyter Labでそのフォルダに移動します。まず、データをロードし、列名を確認する必要があります。列名を正しく処理し、誤った列名を使用した場合に発生する可能性のあるエラーを回避するために、列名をしっかりと検査することが重要です。

以下がpythonコードです。

import pandas as pd

# assign variable to the historical data
file_path = '/home/int_junkie/Documents/ML/predi/XAUUSD.m_H1_201510010000_202408052300.csv'

data = pd.read_csv(file_path, delimiter='\t')

# Display the first few rows and column names
print(data.head())
print(data.columns)

以下が出力です。

2015年から2024年までのおよそ9年間のMetaTrader 5の履歴データを使用しています。この種のデータは、広範な市場サイクルを捉えるのに役立ちます。このデータセットは、市場のさまざまな局面を反映している可能性が高く、こうしたサイクルをより深く理解し、モデル化することが可能になります。長期的なデータセットは、より包括的なシナリオを提供することで、過剰適合のリスクを減らします。

より広範なデータセットで訓練されたモデルは、特に1時間足などの下位時間枠のデータセットにおいて、未経験のデータに対しても効果的に一般化できる可能性が高まります。これは、特に時系列分析において重要であり、より多くの観測データによって結果の信頼性が向上します。例えば、世俗的なトレンド（長期的な市場の方向性）や、予測に関連する反復的な季節的影響を検出することができます。

履歴データからの折れ線グラフ

data.plot.line(y = "<CLOSE>", x = "<DATE>", use_index = True)

以下が出力です。

上記のコードは、金融資産の時系列データなどを可視化する際に使用します。データフレームのインデックスに既に日付が含まれている場合は「x="<DATE>"」の指定を省略し、「use_index=True」を使用することができます。

del data["<VOL>"]
del data["<SPREAD>"]

次に、履歴データから指定された列を削除します。これらの列を削除するためにpandasライブラリを使用します。

data.head()

以下が出力です。

上の出力から、指定した列が確かに削除されたことがわかります。

# We add a colunm for tommorows price
data["<NexH>"] = data["<CLOSE>"].shift(-1)

1. data["<NexH>"]：

• dataデータフレームに<NexH>（次の時間）という新しい列を追加します。この列の値は、各行に対する次の1時間の終値を表します。

2. data["<CLOSE>"].shift(-1)：

• 「data["<CLOSE>"]」は、各日時の終値を含むデータフレームの既存の列を指します。
• 「.shift(-1)」メソッドは、<CLOSE>列の日付を1行分上にシフトさせます（引数が「-1」であるため）。
• その結果、もともと特定の日付に対応していた値が、前の日付に対応する行に表示されるようになります。

以下が出力です。

data["<TRGT>"] = (data["<NexH>"] > data["<CLOSE>"]).astype(int)

data

次に、上記のコードを使って、次の期間の高値(NexH)が現在の終値(<CLOSE>)より大きいかどうかを示すバイナリ値（0または1）を含む新しい列をdataデータフレームに作成します。

1. data["<TRGT>"]：

• dataデータフレームの<TRGT>（ターゲット）という新しい列です。この列には、条件に基づいたバイナリのターゲット値（0または1）が格納されます。

2. (data["<NexH>"] > data["<CLOSE>"])：

• この式は、各行の<NexH>列の値（次の期間の高値）と<CLOSE>列の値（現在の終値）を比較します。
• これはブーリアン系列で、各値はTrue（次の高値が現在の終値より大きい場合）またはFalse（そうでない場合）のいずれかです。

3. .astype(int)：

• この関数は、ブーリアン値（TrueまたはFalse）を整数値（それぞれ1または0）に変換します。
• trueは1になり、falseは0になります。

以下が出力です。

from sklearn.ensemble import RandomForestClassifier

model = RandomForestClassifier(n_estimators = 50, min_samples_split = 50, random_state = 1)

train = data.iloc[:-50]
test = data.iloc[-50:]

predictors = ["<CLOSE>","<TICKVOL>", "<OPEN>", "<HIGH>", "<LOW>"]
model.fit(train[predictors], train["<TRGT>"])

以下が出力です。

1. ランダムフォレスト分類器のインポート

• RandomForestClassifierは、複数の決定木を構築し、それらの出力を結合することで、予測精度を向上させ、過剰適合を抑制するアセンブル機械学習モデルです。

2. モデルの初期化

• estimatorは森に含まれる決定木の数を指定します。この場合、モデルは50本の木を作ります。
• min_sample_splitは、内部ノードを分割するのに必要なサンプルの最小数を設定します。この値を大きくすると、十分なデータが利用可能な場合にのみ分割がおこなわれるようになり、過剰適合が減少します。
• random_stateは、結果の再現性を保証しないランダムシードを固定します。同じシード（例えば1）を使えば、コードを実行するたびに同じ結果が得られます。

3. データを訓練セットとテストセットに分割する

• data.iloc[:-50]は、最後の50行を除くすべての行を学習データとして選択します。
• data.iloc[-50:]は、最後の50行をテストデータとして選択します。
• この分割は時系列データで一般的に使用され、モデルは過去のデータで学習され、最新のデータでテストされ、将来予測の性能を評価します。

4.予測変数の指定

• predictorsリストには、モデルが予測をおこなうために使用する特徴を表す列名が含まれます。これらには ("<CLOSE>"、 "<TICKVOL>"、"<OPEN>"、 "<HIGH>"、 "<LOW>")が含まれます。
  

このコードは、過去のデータに基づいて将来の市場行動を予測するランダムフォレスト分類器を準備します。モデルは終値、ティックボリュームなどの特徴を用いて訓練されます。データを訓練セットとテストセットに分けた後、モデルを訓練データに適合され、過去のパターンから学習して将来の予測をおこないます。

モデルの精度を測る

from sklearn.metrics import precision_score

prcsn = model.predict(test[predictors])

sklearn.metricsモジュールから関数precision-scoreをインポートします。精度スコアは分類モデルを評価するのに使われる指標で、クラスが不均衡な場合に特に有用です。これは、予測されたポジティブな結果のうち、実際にポジティブな結果がいくつ出たかを測定するものです。精度が高いということは、陽性率が低いということです。

prcsn = pd.Series(prcsn, index = test.index)

次に、予測値(prcsn)をテストデータセットからのインデックスを保持したままPandasのSeriesに変換します。

precision_score(test["<TRGT>"], prcsn)

テストセットの実際の目標値と予測値を比較することで、モデルの予測精度を求めます。

以下が出力です。

cmbnd = pd.concat([test["<TRGT>"], prcsn], axis = 1)

cmbnd.plot()

実際の目標値とモデルの予測値を1つのデータフレームにまとめ、分析を容易にします。

以下が出力です。

def predors(train, test, predictors, model):
    model.fit(train[predictors], train["<TRGT>"])
    prcsn = model.predict(test[predictors])
    prcsn = pd.Series(prcsn, index = test.index, name = "Predictions")
    cmbnd = pd.concat([test["<TRGT>"], prcsn], axis = 1)
    return cmbnd

この関数は、訓練データセットとテストデータセット、予測変数のリスト、そして機械学習モデルを受け取ります。関数は訓練データを用いてモデルを訓練し、テストデータに対して予測をおこないます。そして、実際の目標値と予測値を並べたデータフレームを返します。

def backtestor(data, model, predictors, start = 2500, step = 250):
    all_predictions = []

    for i in range(start, data.shape[0], step):
        train = data.iloc[0:i].copy()
        test = data.iloc[i:(i + step)].copy()
        predictions = predors(train, test, predictors, model)
        all_predictions.append(predictions)
    return pd.concat(all_predictions)

この関数は、機械学習モデルを用いて時系列データセットのローリングバックテストをおこないます。バックテストは、データが時間の経過とともに徐々に明らかになる実際の取引環境でおこなわれたかのように予測をシミュレートすることで、モデルのパフォーマンスを評価します。

predictions = backtestor(data, model, predictors)

指定されたデータセット(data)、機械学習モデル(model)、予測変数(predictors)を用いてbacktestor関数を実行します。これはローリングバックテストを実行し、結果の予測値は変数predicitionsに格納されます。

predictions["Predictions"].value_counts()

予測データフレームのPredictions列の各ユニークな値の出現回数を数えます。

以下が出力です。

precision_score(predictions["<TRGT>"], predictions["Predictions"])

モデルの予測精度を計算します。Precisionは、ポジティブな予測の精度を測る指標です。

以下が出力です。

predictions["<TRGT>"].value_counts() / predictions.shape[0]

TRGT列の各ユニークな値が、予測値の総数に占める割合を計算します。

以下が出力です。

horizons = [2, 5, 55, 125, 750]
new_predictors = []

# Ensure only numeric columns are used for rolling calculations
numeric_columns = data.select_dtypes(include=[float, int]).columns

for i in horizons:
    # Calculate rolling averages for numeric columns only
    rolling_averages = data[numeric_columns].rolling(i).mean()
    
    # Generate the ratio column
    ratio_column = f"Close_Ratio_{i}"
    data[ratio_column] = data["<CLOSE>"] / rolling_averages["<CLOSE>"]
    
    # Generate the trend column
    trend_column = f"Trend_{i}"
    data[trend_column] = data["<TRGT>"].shift(1).rolling(i).sum()
    
    new_predictors += [ratio_column, trend_column]
data

さまざまな時間枠におけるローリング平均とトレンドに基づいて、新しい特徴量を生成します。追加予測変数は、様々な期間における市場に関するより多くの情報を提供することで、モデルのパフォーマンスを向上させるのに役立ちます。

以下が出力です。

data = data.dropna()

データフレームで欠損値のある行を削除します。

def predict(train, test, predictors, model):
    model.fit(train[predictors], train["<TRGT>"])
    prcsn = model.predict_proba(test[predictors])[:1]
    prcsn[prcsn >= .6] = 1
    prcsn[prcsn < .6] = 0
    prcsn = pd.Series(prcsn, index = test.index, name = "Predictions")
    cmbnd = pd.concat([test["<TRGT>"], prcsn], axis = 1)
    return cmbnd

モデルは，選択された予測変数とターゲット変数を用いて訓練データ集合で訓練されます。閾値と予測には、0.6のカスタム閾値を適用しています。クラス1の確率が0.6以上の場合、モデルは「1」を予測します。そうでなければ0を予測します。この調整により、モデルはより保守的になり、取引シグナルを出す前に、より高い信頼性を必要とします。

predictions = backtestor(data, model, new_predictors)

predictions["Predictions"].value_counts()

precision_score(predictions["<TRGT>"], predictions["Predictions"])

以下が出力です。

精度はわずかに上がり、四捨五入すれば0.52になります。

モデルを訓練してONNXにエクスポートする

import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import onnx
import skl2onnx
from skl2onnx import convert_sklearn
from skl2onnx.common.data_types import FloatTensorType

# Load and preprocess your data (example)
# Replace this with your actual data loading process
#data = pd.read_csv('your_data.csv')  # Replace with your actual data source
#data = data.dropna()

# Define predictors and target
predictors = ["<CLOSE>", "<TICKVOL>", "<OPEN>", "<HIGH>", "<LOW>"]
target = "<TRGT>"

# Split data into train and test sets
train, test = train_test_split(data, test_size=0.2, shuffle=False)

# Define and train the model
model = RandomForestClassifier(n_estimators=50, min_samples_split=50, random_state=1)
model.fit(train[predictors], train[target])

# Export the trained model to ONNX format
initial_type = [('float_input', FloatTensorType([None, len(predictors)]))]
onnx_model = convert_sklearn(model, initial_types=initial_type)

# Save the ONNX model to a file
with open("random_forest_model.onnx", "wb") as f:
    f.write(onnx_model.SerializeToString())

モデルを学習し、skl2onnxを用いてONNX形式に変換してエクスポートします。保存したモデルをMQL5のFilesフォルダにコピーし、アクセスできるようにします。

MQL5ですべてをまとめる

Oninit()でモデルを読み込みます。

#include <Trade/Trade.mqh>
#define   ModelName          "RandomForestClassifier"
#define   ONNXFilename       "random_forest_model.onnx"

// Single ONNX model resource
#resource "\\Files\\random_forest_model.onnx" as const uchar ExtModelDouble[];

input double lotsize = 0.1;     // Trade lot size
input double stoploss = 20;     // Stop loss in points
input double takeprofit = 50;   // Take profit in points

// Trading functions
CTrade m_trade;

グローバルスコープ上のインクルード変数とグローバル変数。また、ONNXモデルがバイナリリソースとしてMQL5に組み込まれることも指定しています。resourceは外部ファイルをインクルードするために使われます。

//+------------------------------------------------------------------+
//| Run classification using double values                           |
//+------------------------------------------------------------------+
bool RunModel(long model, vector &input_vector, vector &output_vector)
{
    ulong batch_size = input_vector.Size() / 5; // Assuming 5 input features
    if (batch_size == 0)
        return (false);

    output_vector.Resize((int)batch_size);

    // Prepare input tensor
    double input_data[];
    ArrayResize(input_data, input_vector.Size());

    for (int k = 0; k < input_vector.Size(); k++)
        input_data[k] = input_vector[k];

    // Set input shape
    ulong input_shape[] = {batch_size, 5}; // 5 input features for each prediction
    OnnxSetInputShape(model, 0, input_shape);

    // Prepare output tensor
    double output_data[];
    ArrayResize(output_data, (int)batch_size);

    // Set output shape (binary classification)
    ulong output_shape[] = {batch_size, 2}; // Output shape for probability (0 or 1)
    OnnxSetOutputShape(model, 0, output_shape);

    // Run the model
    bool res = OnnxRun(model, ONNX_DEBUG_LOGS, input_data, output_data);

    if (res)
    {
        // Copy output to vector (only keeping the class with highest probability)
        for (int k = 0; k < batch_size; k++)
            output_vector[k] = (output_data[2 * k] < output_data[2 * k + 1]) ? 1.0 : 0.0;
    }

    return (res);
}

RunModel関数は二値分類をおこなうように学習したONNXモデルを実行します。この関数は，どちらのクラスの確率が高いかに基づいて予測クラス（0または1）を決定し，その結果を出力ベクトルに格納します．

//+------------------------------------------------------------------+
//| Generate input data for prediction                               |
//+------------------------------------------------------------------+
vector input_data()
{
    vector input_vector;
    MqlRates rates[];

    // Get the last 5 bars of data
    if (CopyRates(Symbol(), PERIOD_H1, 5, 1, rates) > 0)
    {
        input_vector.Resize(5 * 5); // 5 input features for each bar

        for (int i = 0; i < 5; i++)
        {
            input_vector[i * 5] = rates[i].open;
            input_vector[i * 5 + 1] = rates[i].high;
            input_vector[i * 5 + 2] = rates[i].low;
            input_vector[i * 5 + 3] = rates[i].close;
            input_vector[i * 5 + 4] = rates[i].tick_volume;
        }
    }

    return (input_vector);
}

//+------------------------------------------------------------------+
//| Check if there is a new bar                                      |
//+------------------------------------------------------------------+
bool NewBar()
{
    static datetime last_time = 0;
    datetime current_time = iTime(Symbol(), Period(), 0);

    if (current_time != last_time)
    {
        last_time = current_time;
        return (true);
    }
    return (false);
}

//+------------------------------------------------------------------+
//| Check if a position of a certain type exists                     |
//+------------------------------------------------------------------+
bool PosExists(int type)
{
    for (int i = PositionsTotal() - 1; i >= 0; i--)
    {
        if (PositionGetInteger(POSITION_TYPE) == type && PositionGetString(POSITION_SYMBOL) == Symbol())
            return (true);
    }
    return (false);
}

//+------------------------------------------------------------------+
//| Script program initialization                                    |
//+------------------------------------------------------------------+
int OnInit()
{
    Print("Initializing ONNX model...");

    // Initialize the ONNX model
    long model = OnnxCreateFromBuffer(ExtModelDouble, ONNX_DEFAULT);
    if (model == INVALID_HANDLE)
    {
        Print("Error loading ONNX model: ", GetLastError());
        return INIT_FAILED;
    }

    // Store the model handle for further use
    GlobalVariableSet("model_handle", model);
    return (INIT_SUCCEEDED);
}

//+------------------------------------------------------------------+
//| Expert tick function                                             |
//+------------------------------------------------------------------+
void OnTick()
{
    if (NewBar()) // Trade at the opening of a new candle
    {
        vector input_vector = input_data();
        vector output_vector;

        // Retrieve the model handle
        long model = GlobalVariableGet("model_handle");
        if (model == INVALID_HANDLE)
        {
            Print("Invalid model handle.");
            return;
        }

        bool prediction_success = RunModel(model, input_vector, output_vector);
        if (!prediction_success || output_vector.Size() == 0)
        {
            Print("Prediction failed.");
            return;
        }

        long signal = output_vector[0]; // The predicted class (0 or 1)

        MqlTick ticks;
        if (!SymbolInfoTick(Symbol(), ticks))
            return;

        if (signal == 1) // Bullish signal
        {
            if (!PosExists(POSITION_TYPE_BUY)) // No buy positions exist
            {
                if (!m_trade.Buy(lotsize, Symbol(), ticks.ask, ticks.bid - stoploss * Point(), ticks.ask + takeprofit * Point())) // Open a buy trade
                    Print("Failed to open a buy position, error = ", GetLastError());
            }
        }
        else if (signal == 0) // Bearish signal
        {
            if (!PosExists(POSITION_TYPE_SELL)) // No sell positions exist
            {
                if (!m_trade.Sell(lotsize, Symbol(), ticks.bid, ticks.ask + stoploss * Point(), ticks.bid - takeprofit * Point())) // Open a sell trade
                    Print("Failed to open a sell position, error = ", GetLastError());
            }
        }
    }
}

//+------------------------------------------------------------------+
//| Script program deinitialization                                  |
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
{
    // Release the ONNX model
    long model = GlobalVariableGet("model_handle");
    if (model != INVALID_HANDLE)
    {
        OnnxRelease(model);
    }
}

初期化中にモデルが読み込まれ、後で使用するためにそのハンドルが保存されます。OnTick()関数では、新しいバーが検出されたときにスクリプトがモデルを実行します。モデルの予測（弱気なら0、強気なら1）を使い、それに従って取引を実行します。買い取引は、モデルが強気トレンドを予測したときにおこなわれ、売り取引は弱気トレンドを予測したときにおこなわれます。

結論

まとめると、データ処理パッケージ(Jupyter Lab)を用いて過去のデータを処理し、機械学習を活用してモデルを開発し、予測が可能な状態に訓練しました。この過程では、シームレスな統合と運用に必要な重要なステップを探りました。その後、モデルをリソースとして組み込み、実行時に適切に初期化されて利用できるようにすることで、MQL5内でのモデルの読み込みと処理に焦点を当てました。

結論として、ONNXモデルをMQL5の取引環境に統合し、機械学習による意思決定を強化しました。このプロセスは、MQL5環境にモデルを読み込むことから始まりました。次に、関連する市場データを収集し、それを特徴ベクトルに前処理し、予測のためにモデルに投入するようにEAを設定しました。ロジックは、モデルの出力に基づいて取引を実行するように設計されています。ポジションは、新しいバーが検出され、競合する取引がアクティブでない場合にのみ建てられます。さらに、システムはポジションの確認、エラー管理、リソースの解放をおこない、堅牢で効率的な取引ソリューションを提供します。この実装は、金融分析とAI主導の洞察のシームレスな融合を実証し、リアルタイムで市場状況に適応する自動取引戦略を可能にします。

参照文献

モデルの作成