独自のLLMをEAに統合する（第5部）：LLMによる取引戦略の開発とテスト(IV) - 取引戦略のテスト

目次

1. 目次
2. はじめに
3. この記事の例の開発環境
4. MQL5でLLMをロードする方法
5. GPT-2モデルをONNXモデルに変換する
6. EA戦略とサーバー機能の策定
7. 推論サービスの作成
8. EAクライアント
9. バックテスト
10. 結論

はじめに

これまでの記事では、さまざまな方法で事前学習済みのGPT-2モデルをファインチューニングし、目的に沿ったタスクを実行させる手法を紹介し、それぞれを複数の観点から比較しました。もちろん、紹介したのは代表的な手法に過ぎず、他にも多くのファインチューニング方法が存在します。これまでの実装例を参考に、他の手法を試して比較し、より適したモデルを選択することも可能です。この過程で問題が発生した場合は、ぜひ記事の最後にコメントをお寄せください。

現在、ファインチューニング済みのGPT-2モデルは、単純な定量取引戦略を実行するための初期機能を備えています。そこで本記事では、このモデルを実際の取引戦略にどのように統合するかを紹介します。例として使用するのは、アダプタチューニングによってファインチューニング済みのGPT-2モデルです（詳細は「独自のLLMをEAに統合する（第5部）：LLMを使った取引戦略の開発とテスト(III) - アダプタチューニング」をご参照ください）。特に断りがない限り、本記事で「GPT-2」と記載する場合はこのモデルを指しています。

ただし、本モデルはデモンストレーション目的で限られたデータを用いてファインチューニング済みのものであり、実際の取引環境に対応できるものではありません。テストや最適化をおこなわずにリアルな取引に使用することは絶対に避けてください。この点は非常に重要です。これまでの予測コードはPython環境で作成されていましたが、MetaTrader 5プラットフォームで使用されるMQL5は、高度に統合されたプログラミング言語であり、エキスパートアドバイザー(EA)の開発に非常に強力なツールを提供します。そのため、自動化された定量取引戦略を実現するには、MQL5環境へ移行する必要があります。本記事では、その手順を段階的に解説していきます。

学習済みモデルをPython環境からMQL5 EAに移行し、MetaTrader 5プラットフォーム上で直接実行してリアルタイムの取引判断を支援する方法を見ていきましょう。

この記事の例の開発環境

この記事で使用しているコード例の実行環境をご紹介します。もちろん、読者の開発環境が私と同じである必要はありませんが、コードの実行時に問題が発生した場合は、参考情報として私の環境設定を確認してみてください。

• オペレーティングシステム：Ubuntu 22.04.5 LTS（または対応するバージョンのWSL）
• Pythonバージョン：3.10.14
• 必要なPythonライブラリ:

1. torch-2.4.1
2. numpy-1.26.3
3. pandas-2.2.3
4. transformers-4.45.1
5. petf-0.13.0
6. matplotlib-3.9.2
7. onnx-1.17.0
8. onnxconverter-common-1.14.0
9. onnxruntime-1.20.1
10. onnxruntime-tools-1.7.0

次のステップに進む前に、前回の記事で紹介したアダプタチューニングを使ってモデルを学習済みであることを確認してください。なお、モデルのサイズがプラットフォームの制限を超えているため、私がすでに学習させた重みはアップロードできませんのでご注意ください。

MQL5でLLMを読み込む方法

学習済みのGPT-2モデルをMQL5EAに統合するには、まずPythonで学習されたこのモデルをMQL5環境でどのように読み込んで実行するかという問題を解決する必要があります。以下に、実現可能な方法をいくつか紹介します。

1. モデルをONNXに変換してEAに追加する

ONNX (Open Neural Network Exchange)は、ニューラルネットワークを表現するためのオープンフォーマットであり、異なるディープラーニングフレームワーク間での互換性を実現します。前回の「時系列マイニングのためのデータラベル（第6回）：ONNXを使用してEAに適用してテストする」）では、ONNXを使ってシンプルなモデルをEAに統合する方法を紹介しました。GPT-2モデルをONNX形式に変換してEAにインポートし、MQL5に組み込まれているONNXランタイムライブラリを使用してモデル推論を実行することも可能です。ONNXのMQL5サポートについては、ヘルプファイル「MQL5リファレンス/ONNXモデル」またはMQL5公式ドキュメント(https://www.mql5.com/ja/docs/onnx)を参照してください。

• 長所：

1. 高いパフォーマンス：ONNXランタイムは通常、パフォーマンスが最適化されており、EAで比較的効率的な推論を可能にします。
2. 高い統合性：MQL5にはONNXのサポートが組み込まれているため、外部プログラムやライブラリは不要です。
3. Python環境に依存せずに実行可能：変換後のONNXモデルは単体で動作します。

• 短所：

1. 変換が複雑：複雑な言語モデルをONNX形式に変換するのは困難で、演算子の互換性問題への対応が必要です。
2. デバッグが困難：ONNXモデルのデバッグはPythonモデルより不便です。

2. Winapiを使用してPython推論スクリプトを直接実行する

MQL5はWinapiへのアクセスを提供しており、「kernel32.dll」からWinExec()関数を呼び出すことで外部プログラムを実行することができます。この方法では、既存のPythonスクリプトを使用してGPT-2モデルを読み込み、推論を実行し、EAからWinExec()でスクリプトを呼び出してその出力結果を解析します（または、「shell32.dll」のShellExecuteW()関数でも同様の機能を実現できます）。この方法を実装するには、ある程度の開発経験とWindows開発に関する知識が必要です。

• 長所：

1. シンプルで直接的：モデルを変換する必要がなく、既存のPythonコードをそのまま利用できます。
2. 柔軟性：Pythonエコシステムにある豊富なライブラリやツールを活用できます。

• 短所：

1. パフォーマンスのオーバーヘッド：推論ごとにPythonプロセスを起動する必要があり、非常に非効率です。
2. 依存性：EAはPython環境および外部スクリプトに依存します。
3. データ交換：MQL5とPythonの間でデータをやり取りする仕組みが必要で、より複雑になります。
4. セキュリティ：予期しない事象によりEAがクラッシュする可能性があります。

メモ：この方法は決してお勧めできません。あくまで検証目的や制御された環境下でのみ使用してください。十分な理解と準備がない場合は利用しないでください。

3. ソケット通信を介してPython推論結果を取得する

2番目の方法と似ていますが、Winapiの代わりにソケット通信を使用します（実際には、一般的な推論フレームワークが提供するHTTPサービスのようにHTTPプロトコルを使うこともできますが、基本的にはソケット通信と同様の仕組みのため、ここでは詳しく説明しません）。具体的には、Python側でソケットサーバーを実行してモデルを読み込み推論を行い、EAはクライアントとしてサーバーに接続して入力データを送信し、推論結果を受け取ります。

• 長所：

1. パフォーマンスの向上：ソケット通信により、プロセス起動のオーバーヘッドがなく、安定して動作します。
2. 柔軟性：Pythonの利点を依然として活用できます。

• 短所：

1. 複雑さ：ソケットサーバーとクライアント間の通信ロジックを構築する必要があります。
2. 依存性：EAは外部のPython環境とソケットサーバーに依存します。サービスを設定するにはある程度の知識が必要です。
3. 安定性：ソケット接続の安定性がEAの動作に影響を与える可能性があります。

注：この手法の具体的な実装例について興味がある方は、以前の「時系列マイニングのためのデータラベル（第5回）：ソケットを使用してEAに適用してテストする」をご覧ください。

いくつかの異なる変換方法を検討しましたが、私は現在、GPT-2モデルをONNX形式に変換してEAに統合する方法を選ぶ傾向があります。これは、クロスプラットフォームの課題を解決できるうえに、EAの統合性や安定性も向上するからです。ただし、ONNXモデルのパラメータサイズが大きすぎる場合、MQL5で実行できないという問題があります（たとえば、現在のGPT-2モデルはMQL5のファイル読み込み制限を超えています）。

さらに困難な課題として、Transformerモデルにおけるトークナイザーの問題があります。GPT-2のようなモデルでは入力情報を処理するために専用のトークナイザーが用意されており、これをMQL5で再構築する必要があります。これは非常に大規模な作業になりますが、不可能ではありません。しかし、MQL5のファイルサイズ制限は依然として大きな壁となっています。

この問題を解決するためにINT8形式への量子化を試みましたが、それでもファイルサイズ制限を超えてしまい読み込むことができませんでした。INT4形式に量子化した場合、サイズは要件を満たしますが、MQL5はINT4形式の量子化モデルをサポートしていないため、残念ながらこの方法は断念せざるを得ません。それでもこの記事では、Adapterチューニング済みのGPT-2モデルをONNX形式に変換する手順の例を示し、今後この問題が解決されることを願っています。

本記事では最終的に、Pythonによる推論サービスとソケット通信を使用する方法を採用しました。この方法の利点は、データのセキュリティを確保しながら、EAの実装をシンプルにできる点です。EAでは戦略や取引ロジックにのみ集中すればよく、追加モジュールの統合に煩わされることはありません。またこの方法には、たとえモデル開発がリモートデバイス上でおこなわれ、ローカル環境と互換性がなくても、EA開発を続行できるというメリットもあります。

全体として、この方法は技術的にやや複雑で、実装には一定の知識が要求されるかもしれませんが、高い運用効率と、リアルタイム取引に不可欠なEAの独立性を確保できます。この件についてはすでに前回の記事で詳しく解説しているため、ここでは詳細な説明は割愛します。コード例に関してご不明な点がある場合は、前回の記事をご参照ください。

GPT-2モデルをONNXモデルに変換する

前のセクションでは、ファインチューニング済みのGPT-2モデルをONNX形式に変換してMQL5で使用する際に発生するさまざまな課題について説明しました。しかし、これは試してみる価値のある方向性だと私はまだ信じているため、この記事では追加のセクションを使用して、このパーソナライズされたファインチューニング済みモデルをONNX形式に変換する方法を紹介し、皆さんが現在の窮状に対する解決策を見つけられることを願っています。この部分に興味がない場合は、このセクションをスキップしてかまいません。

1. モデル変換方法

I. Direct Conversion (https://github.com/rayhern/convert-gpt2-xl-to-onnx)

このGitHubリポジトリは、HuggingFaceのtransformersライブラリとtorch.onnxエクスポーターを基に、GPT-2モデルを直接変換するためのスクリプトを提供します。ただし、作者が長期間メンテナンスをおこなっていないため、いくつかの制限があり、最新のtransformersライブラリバージョンと互換性がない可能性があります。

• 長所： GPT-2モデルの変換に特化して最適化された、直接使用できる比較的シンプルなスクリプトを提供します。
• 短所： リポジトリのメンテナンス状況が不明瞭であり、最新のtransformersバージョンとの互換性がなく、特定のバージョンのGPT-2モデルにのみ対応している可能性があります。

II. Microsoft ONNX API (https://github.com/microsoft/onnxruntime-genai)

Microsoftのonnxruntime-genaiライブラリは、生成AIモデル用のONNX変換および最適化APIのセットを提供します。

• 長所：ONNXランタイム向けに最適化されており、推論性能が向上し、Microsoftによってサポート・保守されています。
• 短所：他の方法と比べてやや複雑であり、onnxruntime-genaiライブラリのAPIの習得が必要です。

Ⅲ. torch.onnxを使用してモデルをエクスポートする

PyTorchには組み込みのONNXエクスポート機能(torch.onnx)があり、これを使ってPyTorchモデルをONNX形式にエクスポートできます。

• 長所：PyTorchフレームワークと密接に統合されており使いやすく、torch.onnxは広く利用されているONNXエクスポートツールです。
• 短所：特に新しい演算子やカスタム演算子の場合に、演算子の互換性の問題を手動で処理する必要があるかもしれません。また、モデルの正確性と性能を確保するために、エクスポートパラメータを調整する必要がある場合もあります。

Ⅳ. transformers.onnxを使用してモデルを変換する

HuggingFaceのtransformersライブラリには独自のONNX変換ツール(transformers.onnx)が含まれており、これを使うことで同ライブラリのモデルを簡単にONNX形式に変換できます。

• 長所：シンプルで使いやすく、コマンドラインインターフェイスで簡単に変換ができ、transformersライブラリと密接に統合され、複数の事前学習済みモデルをサポートし、HuggingFaceチームによって積極的に保守・更新されています。
• 短所：torch.onnxと比べると比較的新しいツールであり、互換性に問題が発生する可能性があります。

V. Optimumの使用

Optimumは、HuggingFaceが提供するモデルの最適化と高速化のためのツールライブラリで、ONNX変換機能も備えています。

• 長所：最適化された統合により、ONNX変換を他の最適化技術（量子化、プルーニングなど）と組み合わせることが可能で、HuggingFaceチームにより継続的にサポート・保守されています。
• 短所：Optimumライブラリの使い方を学習する必要があり、ある程度の技術的知識が要求されます。

これらの変換方法にはそれぞれ特長と課題があります。この記事で用いた方法に限定する必要はなく、ご自身のニーズに応じて最適な手法を選択してください。本記事では、例としてtransformers.onnxライブラリを使用してGPT-2モデルを変換しています。

2. GPT-2モデルをONNXモデルに変換する

モデル変換にtransformers.onnxを使用することを決定した後、詳細な変換プロセスを以下に示します。

I. 依存関係をインストールする

まず、transformersライブラリとonnxライブラリがインストールされていることを確認してください。インストールされていない場合は、以下のコマンドでインストールできます。

pip install transformers onnx

GPUアクセラレーションなど特定のハードウェア向けに最適化する必要がある場合は、以下のようにonnxruntime-gpuもインストールしてください。

pip install onnxruntime-gpu

II. 変換コマンド

transformers.onnxはシンプルなコマンドラインツールを提供しており、特別な要件がなければ次のように簡単にモデル変換が可能です。

python -m transformers.onnx --model=path/to/your/tuned_model --feature=causal-lm-with-past path/to/save/onnx_model

このコマンドの各部分の意味は以下のとおりです。

• python -m transformers.onnx：transformers.onnxツールを呼び出します。
• --model=path/to/your/tuned_model：ファインチューニング済みGPT-2モデルのパスを指定します（例：gpt2_Adapter-tuning）。
• --feature=causal-lm-with-past：モデル機能のタイプを指定します。今回は因果言語モデルを使用し、生成効率を高めるためにpast_key_valuesを有効にする必要があるため、causal-lm-with-pastを指定します。
• path/to/save/onnx_model：ONNXモデルの保存先ディレクトリを指定します（例：gpt2_onnx）。

以下は完全なコマンド例です。

python -m transformers.onnx --model=gpt2_Adapter-tuning --feature=causal-lm-with-past gpt2_onnx

このコマンドを実行すると、transformers.onnxは必要な設定ファイルを自動でダウンロードし、モデルをONNX形式に変換します。変換が成功すると、出力ディレクトリ（この例ではgpt2_onnx）にmodel.onnxという名前のファイルが生成され、場合によってはconfig.jsonなどのJSONファイルも含まれます。

ただし、変換時に特定の設定を調整してユースケースに適合させたい場合、このCLIツールでは柔軟性が足りません。したがって、複雑なアプリケーションでは、より精密にモデルを制御できる変換スクリプトの作成が推奨されます。

Ⅲ. 変換スクリプト

Adapter-Tuningによってファインチューニング済みのGPT-2モデルをONNXに変換する際には、変換プロセス中にAdapterモジュールおよび設定ファイルを読み込む必要があり、また互換性問題を避けるためにONNX OPバージョンを指定することが重要です。以下では、必要な機能を段階的に実装していきます。

まず、必要なPythonライブラリと、Adapter()およびGPT2LMHeadModelWithAdapters()クラスをインポートします。これらのクラスについては、前回の「独自のLLMをEAに統合する（第5部）：LLMを使った取引戦略の開発とテスト(III) - アダプタチューニング」で詳しく解説しました。既存のスクリプトからインポートすることも可能ですが、ここでは理解を深めるためにこれらのクラスを変換スクリプトに直接コピーして使用します。

import os
import logging
from pathlib import Path
from transformers.onnx import export, FeaturesManager
from transformers import AutoConfig, AutoTokenizer, GPT2LMHeadModel, modeling_outputs
from torch import nn
import torch.nn.functional as F
import onnx
# Set up basic configuration for logging
logging.basicConfig(level=logging.INFO)
tokenizer = AutoTokenizer.from_pretrained('gpt2')
# Define the Adapter class, which is a simple feed-forward network with dropout
class Adapter(nn.Module):
    def __init__(self, in_features, bottleneck_features=64):
        super(Adapter, self).__init__()
        # Down projection layer
        self.down_project = nn.Linear(in_features, bottleneck_features)
        # Up projection layer
        self.up_project = nn.Linear(bottleneck_features, in_features)
        # Dropout layer for regularization
        self.dropout = nn.Dropout(0.1)
        # Initialize weights of the layers
        self.init_weights()

    def init_weights(self):
        # Initialize weights for down projection layer
        nn.init.normal_(self.down_project.weight, mean=0.0, std=0.02)
        nn.init.constant_(self.down_project.bias, 0)
        # Initialize weights for up projection layer
        nn.init.normal_(self.up_project.weight, mean=0.0, std=0.02)
        nn.init.constant_(self.up_project.bias, 0)

    def forward(self, hidden_states):
        # Apply down projection and ReLU activation
        hidden_states = self.down_project(hidden_states)
        hidden_states = F.relu(hidden_states)
        # Apply dropout
        hidden_states = self.dropout(hidden_states)
        # Apply up projection
        hidden_states = self.up_project(hidden_states)
        # Apply dropout again
        hidden_states = self.dropout(hidden_states)
        return hidden_states

# Define the GPT2LMHeadModelWithAdapters class, which inherits from GPT2LMHeadModel
# and adds adapter layers to each transformer layer
class GPT2LMHeadModelWithAdapters(GPT2LMHeadModel):
    def __init__(self, config):
        super().__init__(config)
        # Create a list of adapter modules, one for each transformer layer
        self.adapters = nn.ModuleList([Adapter(config.n_embd) for _ in range(config.n_layer)])

    def forward(
        self,
        input_ids=None,
        past_key_values=None,
        attention_mask=None,
        token_type_ids=None,
        position_ids=None,
        head_mask=None,
        inputs_embeds=None,
        encoder_hidden_states=None,
        encoder_attention_mask=None,
        labels=None,
        use_cache=None,
        output_attentions=None,
        output_hidden_states=None,
        return_dict=None,
    ):
        # Get the outputs from the transformer
        transformer_outputs = self.transformer(
            input_ids,
            past_key_values=past_key_values,
            attention_mask=attention_mask,
            token_type_ids=token_type_ids,
            position_ids=position_ids,
            head_mask=head_mask,
            inputs_embeds=inputs_embeds,
            encoder_hidden_states=encoder_hidden_states,
            encoder_attention_mask=encoder_attention_mask,
            use_cache=use_cache,
            output_attentions=output_attentions,
            output_hidden_states=output_hidden_states,
            return_dict=return_dict,
        )
        hidden_states = transformer_outputs[0]

        # Apply each adapter to the hidden states
        for i, adapter in enumerate(self.adapters):
            hidden_states = hidden_states + adapter(hidden_states)

        # Get the logits for the language modeling head
        lm_logits = self.lm_head(hidden_states)

        # Compute loss if labels are provided
        loss = None
        if labels is not None:
            # Shift logits and labels for loss computation
            shift_logits = lm_logits[..., :-1, :].contiguous()
            shift_labels = labels[..., 1:].contiguous()
            # Flatten the logits and labels for cross-entropy loss
            loss_fct = nn.CrossEntropyLoss()
            loss = loss_fct(shift_logits.view(-1, shift_logits.size(-1)), shift_labels.view(-1))

        # Return the outputs in the appropriate format
        if not return_dict:
            output = (lm_logits,) + transformer_outputs[1:]
            return ((loss,) + output) if loss is not None else output

        return modeling_outputs.CausalLMOutputWithCrossAttentions(
            loss=loss,
            logits=lm_logits,
            past_key_values=transformer_outputs.past_key_values,
            hidden_states=transformer_outputs.hidden_states,
            attentions=transformer_outputs.attentions,
            cross_attentions=transformer_outputs.cross_attentions,
        )

次に、ファインチューニング済みのGPT-2モデルを読み込み、モデル変換プロセスを制御する必要があります。ファインチューニング済みのGPT-2モデルと対応するトークナイザーを読み込むためにload_model_and_tokenizer()関数を使用し、モデルをONNX形式に変換するためにexport_model_to_onnx()関数を使用します。プロセス全体と入出力パスを制御するためにはmain()関数を使用します。最後に、エクスポート結果を確認するためのcheck_onnx()関数と、ONNXモデルの量子化をおこなうためのquantization()関数を定義します。以下はその一例です。

# Function to load the model and tokenizer
def load_model_and_tokenizer(model_id):
    try:
        # Load the model configuration
        config = AutoConfig.from_pretrained(model_id)
        # Load the model
        model = GPT2LMHeadModelWithAdapters.from_pretrained(model_id)
        # Load the tokenizer
        # tokenizer = AutoTokenizer.from_pretrained('gpt2')
        return config, model,tokenizer
    except Exception as e:
        # Log any errors that occur during loading
        logging.error(f"Error loading model and tokenizer: {e}")
        raise

# Function to export the model to ONNX format
def export_model_to_onnx(model, config, tokenizer, output_path, opset):
    try:
        # Get the appropriate feature for the model
        model_type = config.model_type.replace("-", "_")
        feature = "causal-lm-with-past"
        # Get the ONNX configuration
        onnx_config_constructor = FeaturesManager.get_config(model_type, feature=feature)
        onnx_config = onnx_config_constructor(config)
        # Create the output directory if it doesn't exist
        if not os.path.exists(output_path.parent):
            os.makedirs(output_path.parent)

        # Export the model to ONNX
        export(
            model=model,
            config=onnx_config,
            opset=opset,
            output=output_path,
            preprocessor=tokenizer,
        )
        # Log success message
        logging.info(f"Model successfully converted to ONNX and saved in {output_path}")
    except Exception as e:
        # Log any errors that occur during export
        logging.error(f"Error exporting model to ONNX: {e}")
        raise

# Main function to orchestrate the process
def main():
    # Define the model ID, output path, and ONNX opset version
    model_id = "gpt2_Adapter-tuning"
    onnx_path = "./gpt2_onnx"
    out_path = Path(os.path.join(onnx_path, "gpt2_adapter_tuning.onnx"))
    opset = 14

    # Load the model and tokenizer
    config, model, tokenizer = load_model_and_tokenizer(model_id)
    # Export the model to ONNX
    export_model_to_onnx(model, config, tokenizer, out_path, opset)

def check_onnx():

    # Check the ONNX model
    onnx_model = onnx.load("gpt2_onnx/gpt2_adapter_tuning.onnx")
    onnx.checker.check_model(onnx_model)
    print("ONNX model check passed!")

def quantization():

    from onnxruntime.quantization import quantize_dynamic, QuantType

    # load model
    model_path = "gpt2_onnx/gpt2_adapter_tuning.onnx"
    onnx_model = onnx.load(model_path)

    #dynamic quantize INT4
    quantized_model_path = "gpt2_onnx/quantized_gpt2.onnx"
    quantize_dynamic(model_path, quantized_model_path, weight_type=QuantType.QUInt4)

    print(f"Save the quantized model to: {quantized_model_path}")

この部分のコード実装には特に難しい点はなく、コード内に詳細なコメントも記載されているため、ここでは細かく解説しません。重要な箇所に絞って説明をおこないます。

• ファインチューニング済みのモデルを読み込む際は、必ずアダプタモジュールを含むモデルクラスを使用する必要があります。

model = GPT2LMHeadModelWithAdapters.from_pretrained(model_id)

• ファイルパスはtransformers.onnx.export()が対応する形式に変換する必要があり、単なる文字列パスでは使えません。変換にはpathlibライブラリのPathクラスを使用します。

out_path = Path(os.path.join(onnx_path, "gpt2_adapter_tuning.onnx"))

• export()関数では、tokenizerとpreprocessorのいずれか一方のみを指定できます。それ以外の場合はエラーが発生します。通常はpreprocessorの使用が推奨されます。

export(model=model, config=onnx_config, opset=opset, output=output_path, preprocessor=tokenizer)

• opsetバージョンは、MQL5で正しく読み込むために、MQL5がサポートしているバージョンと一致している必要があります。ここではopset=14を選択します。

opset = 14

• モデルの入力パス（つまり、Adapter-Tuningでファインチューニング済みのGPT-2モデルを含むフォルダ）は、現在のプロジェクトパスの下の「gpt2_Adapter-tuning」フォルダに設定され、出力パスは同じくプロジェクトパスの下の「gpt2_onnx」フォルダに設定されます。

model_id = "gpt2_Adapter-tuning"
onnx_path = "./gpt2_onnx"

• check_onnx()関数とquantization()関数は必須ではなく、参考用として提供されています。

もちろん、これはあくまで基本的な変換スクリプトの例です。シーケンスの動的入力サポートなどの詳細設定は含まれていません。必要な場合は、サンプルスクリプトを基にして機能を追加してください。

完全な変換スクリプトは、torch2onnx.pyという名前の添付ファイルとして提供されています。

EA戦略とサーバー機能の策定

EAの動作モードはすでに決定しています。次に、サーバーが提供するサービス内容と、クライアントに統合される機能についての計画を明確にする必要があります。EAクライアントは主にデータ収集と取引の実行を担当し、Pythonサーバーはクライアントから送信されたデータを受信して推論を行い、その結果をクライアントへ返送します。EAクライアントとPythonサーバーは、ソケットを介して通信します。

1. EA戦略

次に、GPT-2の予測結果に基づいた取引戦略を設計します。この記事の主眼は、GPT-2モデルをMQL5 EAに統合する方法を示すことにあるため、例として簡単な取引戦略を作成します。これはデモンストレーション目的の単純な戦略例であり、実際の取引助言を意図したものではないことを強調しておきます。実用にあたっては、より完全かつ堅牢な取引戦略の開発、ならびに徹底的なバックテストとリスク評価が必要です。

以下はEA戦略のロジックです。

• 過去20時点の終値データを、1分ごとに取得します。
• データをサーバーに送信し、計算結果の返送を待ちます。
• ストップロスやテイクプロフィットは設定せず、常に1つのポジションのみを保有し、サーバーから返された取引シグナルに基づいて注文を実行します。

2. サーバー機能設計

サーバーサイドでは、EAクライアントからのデータ受信、モデルによる推論実行、推論結果に基づいた取引シグナルの計算と返送という主要な機能を実装する必要があります。

以下はサーバーサイド機能です。

• クライアントからのデータを受信します。
• GPT-2モデルとトークナイザーを読み込み、常時使用可能な状態に保ちます。
• 推論を実行し、その結果に基づいて現在の実際の価格と予測価格の平均値との差を計算します。差が0を超える場合は「買い」シグナル、0未満の場合は「売り」シグナルを送信し、0と等しい場合はシグナルを送信しません。
• モデル推論にCPUを使うかGPUを使うかを判定し、現在のデバイスが対応するモードを選択します。

次に、これらの機能に対応した実装を行います。

推論サービスの作成

推論サービスの構築方法については、以前に「時系列マイニングのためのデータラベル（第5回）：ソケットを使用したEAへの応用とテスト」(https://www.mql5.com/ja/articles/13254)で詳しく解説しています。この記事では、そこで使用したserver.pyスクリプトが提供されています。今回のコード部分も、前回の記事のserver.pyスクリプトの基本的なロジックを踏襲しつつ、ファインチューニング済みのGPT-2モデルに適合させ、その他の最適化や改良を加えています。

変更されたコードは主に以下の点が異なります。

• eva()関数に大幅な変更を加え、モデル推論をGPT-2モデルに適応させています。
• ソケットのハンドシェイク処理を最適化し、切断後でもサーバーを再起動せずにクライアントを再接続できるようにしました。これにより、バックテストがより柔軟になり、毎回サーバーを再起動する必要がなくなります。
• クライアントの接続状態を検出する処理を追加し、不要なリソースの消費を回避します。
• 予測結果に変化があった場合のみ結果を出力し、重複出力を抑制します。
• サーバーのクラッシュを防止するため、エラーハンドリングのロジックを追加しています。
• コード全体の処理フローを最適化しています。

コード部分の詳細には本記事では踏み込みません。必要な変更点のみに焦点を当てて説明します。

1. 必要なライブラリをインポートする

通常のライブラリに加えて、スクリプト内に組み込んだAdapterクラスとGPT2LMHeadModelWithAdaptersクラスもインポートする必要があります。これらのクラスは、GPT-2のファインチューニングに関する以前の記事から取得するか、本記事で提供しているtorch2onnx.pyから直接インポートすることができます。サンプルコードでは、2つのクラスをtorch2onnx.pyから直接インポートする方式を採用しています。

import socket
from time import sleep
import pandas as pd
import numpy as np
import warnings
import base64
import hashlib
import struct
from torch2onnx import GPT2LMHeadModelWithAdapters,Adapter
from transformers import AutoTokenizer
import logging
import torch
from statistics import mean
# Set logging and warning
logging.basicConfig(level=logging.INFO)
warnings.filterwarnings("ignore")
# Set device
dvc='cuda' if torch.cuda.is_available() else 'cpu'

# Global 
model_id = "gpt2_Adapter-tuning"
encoder_length=20
prediction_length=10
info_file="results.json"
host="0.0.0.0"
port=10055

2. load_model()関数にGPT-2モデルの読み込みロジックを追加する

元のスクリプト(server.py)では、load_model()関数を使用してモデルを読み込みます。ここでは、GPT-2モデルの読み込みロジックに加えて、GPT-2トークナイザーの読み込みロジックも追加する必要がある点に注意してください。

# Function to loda model
def load_model():
    try:
        # Load the model
        model = GPT2LMHeadModelWithAdapters.from_pretrained(model_id).to(dvc)
        # Load the tokenizer
        tokenizer = AutoTokenizer.from_pretrained('gpt2')
        print("Model loaded!")
        return  model,tokenizer
    except Exception as e:
        # Log any errors that occur during loading
        logging.error(f"Error loading model and tokenizer: {e}")
        raise

3. eva()関数にGPT-2モデルの推論ロジックを追加する

def eva(msg,model,tokenizer):
        
        # Get the data
        msg=np.fromstring(msg, dtype=float, sep= ',').tolist()
        # Parse the data
        input_data=msg[-encoder_length:]
        # Create the prompt
        prompt = ' '.join(map(str, input_data))
        # Generate the predication
        token=tokenizer.encode(prompt, return_tensors='pt').to(dvc)
        attention_mask = torch.ones_like(token).to(dvc)
        model.eval() 
        generated = tokenizer.decode(
            model.generate(
                token, 
                attention_mask=attention_mask,
                pad_token_id=tokenizer.eos_token_id,
                do_sample=True, 
                max_length=200)[0], 
            skip_special_tokens=True)
        generated_prices=generated.split('\n')[0]

        # Remove non-numeric formats
        def try_float(s):
            try:
                return float(s)
            except ValueError:
                return None
        generated_prices=generated_prices.split()
        generated_prices=list(map(try_float,generated_prices))
        generated_prices = [f for f in generated_prices if f is not None]

        generated_prices=generated_prices[0:prediction_length]
        
        # Calculate and send the results
        last_price=input_data[-1]
        prediction_mean=mean(generated_prices)
        if (last_price-prediction_mean) >= 0:
            # print('Send sell.')
            return "sell" 
        else:
            # print("Send buy.")
            return "buy"

入力の長さは、Adapter-Tuningを使用してGPT-2モデルを学習する際に使用したデータ形式と一致している必要があることに注意してください。

• input_data = msg[-encoder_length:]：クライアントから送信された最後の20個のデータポイントをモデルの入力として取得します。
• prompt = ' '.join(map(str, input_data))：データを文字列形式に変換し、プロンプトとして構成します。
• token = tokenizer.encode(prompt, return_tensors='pt').to(dvc)：事前学習済みのGPT-2モデルのトークナイザーでプロンプトをエンコードし、モデル推論で使用する現在のデバイスへ転送します。
• attention_mask = torch.ones_like(token).to(dvc)：モデル推論用のアテンションマスクを定義します。
• model.generate(token, attention_mask=attention_mask, pad_token_id=tokenizer.eos_token_id, do_sample=True, max_length=200)[0]：モデル推論を実行します。
• generated = tokenizer.decode(model.generate(...), skip_special_tokens=True)：推論結果をデコードし、特殊トークンをスキップするように設定します。
• generated_prices = generated.split('\n')[0]：デコードされた推論結果を分割します。
• try_float(s)：この関数は、推論結果に浮動小数点に変換できない要素が含まれているかを検出するために使用します。
• generated_prices = generated_prices.split()：予測結果をスペースで区切り、数値に変換できない区切り記号を除去します。
• generated_prices = list(map(try_float, generated_prices))：generated_pricesの全要素を浮動小数点形式に変換し、変換できない場合はtry_float(s)関数を用いてNoneを設定します。
• generated_prices = [f for f in generated_prices if f is not None]：generated_prices内のすべての要素を走査し、Noneである要素を削除します。
• generated_prices = generated_prices[0：prediction_length]：先頭から10個の予測値のみを取得して参照とします。
• if (last_price - prediction_mean) >= 0：クライアントから送信された直近の価格と予測値の平均との差を計算し、0以上であれば「売り」シグナル、0未満であれば「買い」シグナルを送信します。

推論にはtransformersライブラリを使用することを選択しています。なお、前述のtorch2onnx.pyスクリプトを使ってモデルをONNX形式に変換し、onnxruntimeライブラリで推論をおこなう方法もありますが、本記事ではその方法については説明しません。

4. サーバー

サーバーのすべての機能はserver_()クラスに統合されており、全体的なコードの変更はそれほど大きくありません。ここでは詳細な解説はおこなわず、変更された部分についてのみ説明します。

class server_:
    def __init__(self, host = host, port = port):
        self.sk = socket.socket(socket.AF_INET, socket.SOCK_STREAM,)
        self.host = host
        self.port = port
        self.sk.bind((self.host, self.port))
        self.re = ''
        self.model,self.tokenizer=load_model()
        self.stop=None
        self.sk.listen(1)
        self.sk_, self.ad_ = self.sk.accept()
        self.last_action=None
        print('server running：',self.sk_, self.ad_)  

    def msg(self):
        self.re = ''
        wsk=False
        while True:
            sleep(0.5)
            if self.is_connected():
                try:
                    data = self.sk_.recv(2500)
                except Exception as e:
                    break
                if not data:
                    break
                if (data[1] & 0x80) >> 7:
                    fin = (data[0] & 0x80) >> 7 # FIN bit
                    opcode = data[0] & 0x0f # opcode
                    masked = (data[1] & 0x80) >> 7 # mask bit
                    mask = data[4:8] # masking key
                    payload = data[8:] # payload data

                    # print('fin is：{},opcode is：{}，mask:{}'.format(fin,opcode,masked))
                    message = ""
                    for i in range(len(payload)):
                        message += chr(payload[i] ^ mask[i % 4])
                    data=message
                    wsk=True
                else:
                    data=data.decode("utf-8")

                if '\r\n\r\n' in data: 
                    self.handshake(data)
                    data=data.split('\r\n\r\n',1)[1]
                if "stop" in data:
                    self.stop=True
                    break
                if len(data)<50:
                    break
                self.re+=data
                bt=eva(self.re, self.model,self.tokenizer)
                bt=bytes(bt, "utf-8")
                # If the signal changes,then print the information
                if bt != self.last_action:
                    if bt == b'buy':
                        print('Send buy.')
                    elif bt == b'sell':
                        print('Send sell.')
                    self.last_action = bt 
                if wsk:
                    tk=b'\x81'
                    lgt=len(bt)
                    tk+=struct.pack('B',lgt)
                    bt=tk+bt
                self.sk_.sendall(bt)
            else:
                print("Disconnected！Try to connect the client...")
                try:
                    # reconnect
                    self.sk_.close()
                    self.sk.listen(1)
                    self.sk_, self.ad_ = self.sk.accept()
                    print('Reconnected:', self.sk_, self.ad_)
                    # handshake
                    while True:
                        sleep(0.5)
                        data = self.sk_.recv(2500)
                        data=data.decode("utf-8")
                        if '\r\n\r\n' in data:
                            self.handshake(data)
                            break
                    print("Reconnection succeed！")
                    # # clean the socket
                    # while True:
                    #     if not self.sk_.recv(2500):
                    #         break
                except Exception as e:
                    print(f"Reconnection failed: {e}")
        return self.re
        
    def __del__(self):
        print("server closed!")
        self.sk.close()
        if self.sk_ is not None:
            self.sk_.close()
            self.ad_.close()
    def handshake(self,data):
        try:           
            # Handshake
            key = data.split("\r\n")[4].split(": ")[1]
            GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
            ac = base64.b64encode(hashlib.sha1((key+GUID).encode('utf-8')).digest())
            response_tpl="HTTP/1.1 101 Switching Protocols\r\n" \
                        "Upgrade:websocket\r\n" \
                        "Connection: Upgrade\r\n" \
                        "Sec-WebSocket-Accept: %s\r\n" \
                        "WebSocket-Location: ws://%s/\r\n\r\n"
            response_str = response_tpl % (ac.decode('utf-8'), "127.0.0.1:10055")
            self.sk_.send(bytes(response_str, encoding='utf-8'))
            print('Handshake succeed!')
        except Exception as e:
            print(f"Connection failed: {e}")
            return None
        
    def is_connected(self):
        try:
        # Check remote 
            # remote_addr = self.sk_.getpeername()
            data = self.sk_.recv(1, socket.MSG_PEEK)
            return True
        except socket.error:
            self.last_action=None
            return False

• クライアントがオンラインかどうかを検出するためのクラス関数is_connected(self)を追加します。
• ハンドシェイクのロジックを統合し、メインの解析ロジックが煩雑にならないようにするためのクラス関数handshake(self, data)を追加します。
• 取引シグナルが変更されたかどうかを検出するためのクラスメンバーself.last_actionを追加します。頻繁な出力を避けるため、取引シグナルが変化した場合にのみ結果を出力します。クライアントが切断された場合は、再接続時に誤ったシグナルが送信されないようNoneにリセットします。

注 ：ホストアドレスは「0.0.0.0」に設定しています。「127.0.0.1」に設定すると、別のホストで実行されているリモートクライアントが接続できなくなるためです。「0.0.0.0」に設定することで、サーバーとクライアントが同一ホスト上にない場合でも接続が可能になります（EAクライアントサイドでは正しいホストのIPアドレスを設定する必要があります）。

全体のコードは、添付のserver.pyファイルに含まれています。 サーバー実行中は、ターミナルに対応する実行情報が表示されます。

EAクライアント

クライアントは主に、以前の記事（具体的には以前の記事で言及された内容）のロジックに従い、それに応じて一部のロジックを修正しています。特殊な状況下でMQL5組み込みソケットが接続できずに信号中断が発生することを回避するため、2つのソケット互換メソッド（1つはWinapiを使用してWebSocketを実装する方法、もう1つはMQL5の組み込みソケットモジュールを使用する方法）を引き続き保持しています。主な動作の流れは、OnInit()関数でソケットを初期化し、OnTick()関数で取引ロジックを処理し、OnTimer()関数で一定間隔ごとにサーバーへデータを送信し推論結果を受信する処理をおこなう形です。

1. 定数を定義する

#include <WinAPI\winhttp.mqh>

int sk=-1;
string host="127.0.0.1";
int port= 10055;
int data_len=100;
string pre=NULL;
HINTERNET ses_h,cnt_h,re_h,ws_h;

• sk：ソケットハンドル
• host、port：接続するサーバーアドレスとポート
• data_len：送信する価格データポイントの数
• pre：予測結果を保存する文字列
• ses_h、cnt_h、re_h、ws_h：それぞれWinHttpセッション、接続、リクエスト、Webソケットへのハンドル

2. ソケットの初期化

int OnInit()
  {
//--- create timer
   EventSetTimer(60);
   ses_h=cnt_h=re_h=ws_h=NULL;
//handshake
   ses_h=WinHttpOpen("MT5",
                     WINHTTP_ACCESS_TYPE_DEFAULT_PROXY,
                     NULL,
                     NULL,
                     0);
   //Print(ses_h);
   if (ses_h==NULL){
      Print("Http open failed!",string(kernel32::GetLastError()));
      return INIT_FAILED;
      }
   cnt_h=WinHttpConnect(ses_h,
                        host,
                        port,
                        0);
   //Print(cnt_h);
   if (cnt_h==NULL){
      Print("Http connect failed!",string(kernel32::GetLastError()));
      return INIT_FAILED;
      }
   re_h=WinHttpOpenRequest(cnt_h,
                           "GET",
                           NULL,
                           NULL,
                           NULL,
                           NULL,
                           0);
   if(re_h==NULL){
      Print("Request open failed!",string(kernel32::GetLastError()));
      return INIT_FAILED;
   }
   uchar nullpointer[]= {};
   if(!WinHttpSetOption(re_h,WINHTTP_OPTION_UPGRADE_TO_WEB_SOCKET,nullpointer,0))
     {
          Print("Set web socket failed!",string(kernel32::GetLastError()));
          return INIT_FAILED;
       }
   bool br;   
   br = WinHttpSendRequest( re_h,
                             NULL, 
                             0,
                             nullpointer, 
                             0, 
                             0, 
                             0);
   if (!br)
      {
         Print("send request failed!",string(kernel32::GetLastError()));
         return INIT_FAILED;
         }
   br=WinHttpReceiveResponse(re_h,nullpointer);         
   if (!br)
     {
       Print("receive response failed!",string(kernel32::GetLastError()));
       return INIT_FAILED;
       }
   ulong nv=0; 
   ws_h=WinHttpWebSocketCompleteUpgrade(re_h,nv);  
   if (!ws_h)
   {
      Print("Web socket upgrade failed!",string(kernel32::GetLastError()));
      return INIT_FAILED;
         }
    else{
      Print("Web socket connected！");
    }   
  
   WinHttpCloseHandle(re_h);
   re_h=NULL;
 
    sk=SocketCreate();
    Print(sk);
    Print(GetLastError());
    if (sk==INVALID_HANDLE) {
        Print("Failed to create socket");
        //return INIT_FAILED;
    }

    if (!SocketConnect(sk,host, port,1000)) 
    {
        Print("Failed to connect to built-in socket");
        //return INIT_FAILED;
    }
//---
   return(INIT_SUCCEEDED);
  }

初期化部分では、主にWinapi WebSocketとMQL5に組み込まれているソケットの初期化を実装します。この部分は前回の記事の内容とほとんど変わっていないため、本記事では詳しく取り上げません。

3. 取引戦略

void OnTick()
  {
//---
   MqlTradeRequest request;
   MqlTradeResult result;
   //int x=SymbolInfoInteger(_Symbol,SYMBOL_FILLING_MODE);

    if (pre!=NULL)
    {
        //Print("The predicted value is:",pre);
        ulong numt=0;
        ulong tik=0;
        bool sod=false;
        ulong tpt=-1;
        ZeroMemory(request); 
        numt=PositionsTotal();
        //Print("All tickets: ",numt);
        if (numt>0)
         {  tik=PositionGetTicket(numt-1);    
            sod=PositionSelectByTicket(tik);
            tpt=PositionGetInteger(POSITION_TYPE);//ORDER_TYPE_BUY or ORDER_TYPE_SELL
            if (tik==0 || sod==false || tpt==0) return; 
            }
        if (pre=="buy")
        {  
           
           if (tpt==POSITION_TYPE_BUY)
               return;
               
            request.action=TRADE_ACTION_DEAL;
            request.symbol=Symbol();
            request.volume=0.1;
            request.deviation=5;
            request.type_filling=ORDER_FILLING_IOC;
            request.type = ORDER_TYPE_BUY;  
            request.price = SymbolInfoDouble(Symbol(), SYMBOL_ASK);
           if(tpt==POSITION_TYPE_SELL)
             {
               request.position=tik;
               Print("Close sell order.");
                    }
           else{     
  
            Print("Open buy order.");
                     }
            OrderSend(request, result);
               }
        else{
           if (tpt==POSITION_TYPE_SELL)
               return;
               
            request.action = TRADE_ACTION_DEAL;      
            request.symbol = Symbol();  
            request.volume = 0.1;  
            request.type = ORDER_TYPE_SELL;  
            request.price = SymbolInfoDouble(Symbol(), SYMBOL_BID);  
            request.deviation = 5; 
            //request.type_filling=SymbolInfoInteger(_Symbol,SYMBOL_FILLING_MODE);
            request.type_filling=ORDER_FILLING_IOC;
           if(tpt==POSITION_TYPE_BUY)
               {
               request.position=tik;
               Print("Close buy order.");
                    }
           else{

               Print("OPen sell order.");
                    }
            
            OrderSend(request, result);
              }
        //is_pre=false;
        }
    pre=NULL;
  }

ロジックをより明確にするため、取引戦略全体をOnTick()関数に統合します。OnTick()関数が実行されたら、グローバル変数preが空かどうかを確認します。空でなければ、クライアントから送信された予測結果が存在することを意味します。

その後、予測結果（「買い」または「売り」）に基づいて取引リクエストを送信します。

• 「買い」の場合、既存のポジションがなければ新規に買いポジションを開き、もし売りポジションを持っていればそれを決済します。
• 「売り」の場合、既存のポジションがなければ新規に売りポジションを開き、もし買いポジションを持っていればそれを決済します。

利益確定（テイクプロフィット）や損切り（ストップロス）は設定せず、常に1つの注文のみを維持し、取引シグナルによってのみポジションを制御します。

4. サーバーとのやり取り

void OnTimer()
  {
//---
    MqlTradeRequest request;
    MqlTradeResult result;
    char recv_data[5];
    double priceData[100];
    string dataToSend;
    char ds[];
    int nc=CopyClose(Symbol(),0,0,data_len,priceData);
    for(int i=0;i<ArraySize(priceData);i++) dataToSend+=(string)priceData[i]+","; 
    int dsl=StringToCharArray(dataToSend,ds);    
    
    if (sk!=-1)
    {
       if (SocketIsWritable(sk))
           {
           Print("Send data:",dsl);
           int ssl=SocketSend(sk,ds,dsl);     
            }
       uint len=SocketIsReadable(sk); 
       if (len)
       {
         int rsp_len=SocketRead(sk,recv_data,len,500);
         if(rsp_len>0)
         {
           string result=NULL; 
           result+=CharArrayToString(recv_data,0,rsp_len);
           Print("The predicted value is:",result);
           if (StringFind(result,"buy"))
           {
            pre="buy";
           }
           if (StringFind(result,"sell")){
             pre="sell";

               }
            }
          }
     }
    else
    {
       ulong send=0;                         
       if (ws_h)
       { 
         send=WinHttpWebSocketSend(ws_h,
                                   WINHTTP_WEB_SOCKET_BINARY_MESSAGE_BUFFER_TYPE,
                                   ds,
                                   dsl);
          //Print("Send data failed!",string(kernel32::GetLastError()));    
         if(!send)
            {
               ZeroMemory(recv_data);
               ulong rb=0;
               WINHTTP_WEB_SOCKET_BUFFER_TYPE st=-1;
               ulong get=WinHttpWebSocketReceive(ws_h,recv_data,ArraySize(recv_data),rb,st);
                if (!get)
                {
                    pre=NULL; 
                    pre+=CharArrayToString(recv_data,0);
                    Print("The predicted value is:",pre);
                     }
                 }
            }
        }           
  }

サーバーの主な関数は、現在のチャートから100個のデータポイントを収集し、それをサーバーに送信して推論結果を受け取り、その結果に基づいてグローバル変数を更新することです。これにより、サーバーが提供する結果に従って取引戦略が正しく実行されるようにします。ここでは、2種類のソケット接続方法を用いてデータのやり取りを実装し、接続されているソケットの種類に応じて適切な方法を自動的に選択する仕組みになっています。

5. リソースの解放

void OnDeinit(const int reason)
  {
//--- destroy timer
   EventKillTimer();
   uchar stop[];
   
   int ls=StringToCharArray("stop",stop);
   
   SocketSend(sk,stop,ls);
   SocketClose(sk);
 // close the websocket
   WinHttpSendRequest(re_h,NULL,0,stop,0,0,0);
   BYTE closearray[]= {};
   ulong close=WinHttpWebSocketClose(ws_h,
                                    WINHTTP_WEB_SOCKET_SUCCESS_CLOSE_STATUS,
                                    closearray,
                                    0);
   if(close)
     {
      Print("websocket close error "+string(kernel32::GetLastError()));
      if(re_h!=NULL)
         WinHttpCloseHandle(re_h);

      if(ws_h!=NULL)
         WinHttpCloseHandle(ws_h);

      if(cnt_h!=NULL)
         WinHttpCloseHandle(cnt_h);

      if(ses_h!=NULL)
         WinHttpCloseHandle(ses_h);
     }   
  }

OnDeinit()関数では、使用したシステムリソースを適切に解放し、リソースの回復処理を行います。

また、GPT-2モデルの推論処理はEA内に実装していないため、EAのロジックはシンプルかつコンパクトになっています。ただし、このEAはリスク管理のロジックを一切組み込んでいないため、GPT-2の推論結果のみを基にポジションの保持や新規オープンを判断するのは非常にリスクの高い手法です。実際の取引には決して使用しないでください。

完全なコードは、この記事の添付ファイル「gpt2_EA.mql5」にて提供されています。

バックテスト

EAのパフォーマンスを評価するために、MetaTrader 5クライアントのストラテジーテスターでバックテストを実施します。まず、適切な履歴データの期間を選択し、バックテストのパラメータを設定します。なお、GPT-2モデルはNZDUSD通貨ペアで学習されているため、バックテスト時にはNZDUSD通貨ペアのみを選択してテストをおこなう必要があります。

バックテスト実行中：

バックテストが完了すると、結果は次のようになります。

バックテストレポートを確認することで、EAの収益性、最大ドローダウン、勝率などの指標を分析することができます。ただし、この取引戦略はシンプルであるため、バックテストの結果は必ずしも理想的とは言えません。これは主に、戦略においてパラメータの最適化やリスク管理が行われておらず、モデルの学習プロセスやデータ準備にも大きな改善の余地があるためです。全体として、この作業には多くの忍耐力が求められます。また、市場環境の変化やモデルの限界により、バックテストの結果が将来のリアルな取引でのEAのパフォーマンスを保証するものではない点にもご注意ください。モデルの制約によって、予測結果が不安定になる可能性もあります。

結論

本記事では、特定の金融データ（NZDUSD通貨ペア）でファインチューニング済みのGPT-2モデルをEAプログラムに統合する方法について、モデルのファインチューニング、推論ロジックの実装、サーバーとクライアントの構築、そして最終的な取引戦略との統合に至るまで、プロセス全体を体系的に解説しました。

本稿の取引戦略は比較的シンプルであり、本記事はあくまでデモンストレーションを目的としたものである点を強調しておく必要があります。実際の運用においては、複数のテクニカル指標の組み合わせ、市場センチメントの考慮、損切り・利確の設定など、より高度で堅牢な戦略の構築が求められます。

加えて、モデルの学習プロセスやデータ準備にはまだ大きな最適化の余地があり、市場環境の変化やモデル自体の制限により、予測結果が不安定になる可能性もあります。それにもかかわらず、本研究の意義は、大規模言語モデルが定量取引において持つ可能性を実証した点にあります。GPT-2のようなモデルは、従来の市場データの分析に加え、ニュース、ソーシャルメディアなどのテキストデータを扱うことで、より包括的な市場センチメント分析を実現し、トレーダーのより賢明な意思決定を支援します。このようなクロスモーダルな能力は、従来の金融モデルには見られない特徴であり、今後さらに探求していくべき重要な分野です。

次回の記事では、具体例を用いて、大規模言語モデルを定量取引に最適に活用する方法について詳しく解説します。

付録：

ファイル	詳細
torch2onnx.py	GPT-2モデルをONNX形式に変換するPythonスクリプト
server.py	GPT-2モデル推論サービスと結果を提供するPythonスクリプト
gpt2_EA.mq5	GPT-2モデルの推論結果をテストするためのEAプログラム