MetaTrader 5での並列計算

プロセッサの並列性序論

現代 PCの大半は、同時に複数のタスクを処理する能力があります。これは、複数のプロセッサ コアのお陰です。そしてそのコア数は日々増えていっています。2、 3、 4、 6 ...コアという具合にです。Intel は最近実験的に80 コアでの動作を 実演 しました。（そうです。タイプミスではありませんよ。八十コアです。ただ残念ながら、このコンピュータは店頭に並びはしませんが。というのもこのプロセッサは技術の潜在的能力を研究する目的のためだけに作られたものだからです。）

コンピュータユーザーの誰も（そして初心者プログラマー全員）が、それがどんな風に動作するのか理解できるわけではありません。よって、必ずこういう質問が出るものです。「なぜそんなにたくさんのコアを備えたプロセッサが必要なの？前（コア一つの頃）にもコンピュータは多くのプログラムを同時に実行しそれらは全て動作したのに。」真実は、そうではない、です。次の図を見てみます。

図1　アプリケーションの同時実行

図のケース A はシングルコアプロセッサでプログラムを1件実行したら何が起こるかを示しています。プロセッサはそのプログラムの実装に全時間を費やし、プログラムは時間Tをかけて一定量の作業を行います。

ケース B　2 件のプログラムを立ち上げています。プロセッサは、時間のあるポイントで物理的にそのコアの一つが1件のコマンドのみを実行し、そのため2つのプログラム間で常に切り替えを行うようにできています。最初のプログラムをいくらか実行し、それからふたつめを、というようにです。これはひじょうにすばやく起こります。1秒間に何度もです。よってプロセッサが両方のプログラムを同時に実行しているように見えるのです。ですが、実際には、それぞれのプログラムが別々のプロセッサで実行されるより、この実行には2倍の時間がかかるのです。

ケース C では、プロセッサ内のコア数がプログラム数に対応していれば、ケースBでの問題は効果的に解決することを示しています。各プログラムは自由に使える個別のコアを持ち、実行スピードは上がっています。そのスピードはちょうどケースAと同じです。

ケース D　多くのユーザーの共通した思い込みへの答えです。 皆、プログラムがマルチコアプロセッサで実行されていると、数倍早く処理されると信じています。一般的には、これは事実ではありません。なぜならプロセッサは単独でプログラムをパーツに分け、それらを一斉に処理することができないからです。

たとえば、プログラムが最初にパスワードを尋ね、その認証が行われたら、コアでパスワードプロンプト処理をして、同時に別の認証を行うことはできません。認証は決して成立しないのです。というのも、最初の時点でパスワードはまだ入力されていないからです。

プロセッサはプログラマが実装した設計、またプログラム動作のロジックすべてを知っているわけではありません。ゆえに、自立的にコア同士でプログラムを分割するのは不可能です。よって、1本のプログラムをマルチコア システムで実行しても、コアの一つが使われるだけで、シングルコア プロセッサで処理したかのようなスピードで実行されるにすぎません。

ケース E コアをすべて使い、より速く実行するにはプログラムに何をする必要があるか説明しています。プログラマーはプログラムのロジックがわかっているので、開発中にプログラム上で同時実行可能な箇所に印をつけます。実行中、プログラムはこの情報をプロセッサに伝え、プロセッサはプログラムに必要なコア数を割り当てるのです。

MetaTraderにおける並列性

前章では、CPU コアをすべて使用し、プログラム実行速度を上げるには何が必要か解明しました。並列化可能なプログラムコードをなんらかの形で分かれたスレッドに割り当てる必要があるのです。多くのプログラム言語には、このための特殊なクラスやオペレータが備わっています。しかし、 MQL5 言語にはそのような内蔵インスツルメントはありません。では、われわれがなすべきことは？

解決策は 2 種類あります。

1. DLLの使用	2. MetaTradeの非言語リソースの使用
並列化のための内蔵ツールを備えている言語で DLL を作成することにより、MQL5-EA でも同様に並列化が可能です。	MetaTrader開発者からの情報によりますと、クライアント端末のアーキテクチャはマルチスレッドとのことです。よって、特定の条件下で受信するマーケットデータは別のスレッドで処理されます。EAやインディケータの数に対してプログラムコードを分離する方法をみつけることができれば、MetaTrader はプログラム実行に数多くのCPU コアを使用することが可能となります。

本稿では最初の方法については述べません。DLL で好きなものを実装できるのは明らかです。われわれは MetaTrader の標準的手法のみを利用し、MQL5以外の言語を必要としない解決策を模索するのです。

それゆえ、二番目の手法について多く述べていきます。MetaTraderで複数コアがどのようにサポートされているか正確に知るために一連の実験を行う必要があります。このために、テストインディケータと検証用EAを作成します。それはCPUに大きな負荷をかける進行中の作業を行います。

以下の i-flood インディケータを書きました。

//+------------------------------------------------------------------+
//|                                                      i-flood.mq5 |
//+------------------------------------------------------------------+
#property indicator_chart_window

input string id;
//+------------------------------------------------------------------+
void OnInit()
  {
   Print(id,": OnInit");
  }
//+------------------------------------------------------------------+
int OnCalculate(const int rt,const int pc,const int b,const double &p[])
  {
   Print(id,": OnCalculate Begin");
   
   for (int i=0; i<1e9; i++)
     for (int j=0; j<1e1; j++);
     
   Print(id,": OnCalculate End");
   return(0);   
  }
//+------------------------------------------------------------------+

そしてその類似体e-flood EA です。

//+------------------------------------------------------------------+
//|                                                      e-flood.mq5 |
//+------------------------------------------------------------------+
input string id;
//+------------------------------------------------------------------+
void OnInit()
  {
   Print(id,": OnInit");
  }
//+------------------------------------------------------------------+
void OnTick()
  {
   Print(id,": OnTick Begin");
   
   for (int i=0; i<1e9; i++)
     for (int j=0; j<1e1; j++);
     
   Print(id,": OnTick End");
  }
//+------------------------------------------------------------------+

多様なチャートウィンドウ（1つのチャート、同一シンボルの2つのチャート、異なるシンボルの2つのチャート）の組合せを開き、本インディケータまたはEAのコピーを2部配置することで、端末がCPUコアをどのように使用するか観察することが可能です。

これらインディケータおよびEA はまた、ログにメッセージを送信し、その表示の連続を観測するのは興味深いものです。これらログはご自身で作成できるので、ここでは提供しません。関心はいくつコアを見つけるのか、またチャートのどの組合せが端末で使用されるのか、ということです。

Windowsの『タスクマネージャ』で動作中のコア数を計測することができます。

図2　CPUコア

全計測結果は以下の図に集約されています。

組合せ No.	ターミナルコンテンツ	CPU 使用量
1	1チャートに2 インディケータ	1 コア
2	異なるチャートに2 インディケータ＆同一組合せ	1 コア
3	異なるチャートに2 インディケータ＆異なる組合せ	2 コア
4	同一チャートに2 EA：この状況は不可	-
5	異なるチャートに2 EA＆同一組合せ	2 コア
6	異なるチャートに2 EA＆同一組合せ	2 コア
7	異なる組合せに2 インディケータ＆EAより作成	2 コア

7番目の組合せがイディケータ作成の一般的な方法で、多くのトレーディング戦略で使用されます。

ただひとつ特殊な機能としては、私は2つの異なる通貨ペアに2つのインディケータを作成しました。その理由は、組合せ1と2に関しては同じ通貨ペアにインディケータを複数配置しても意味がないためです。この組合せに対し、EA e-flood スタータを使用しました。これは i-flood のコピーを2部作成するものです。

//+------------------------------------------------------------------+
//|                                              e-flood-starter.mq5 |
//+------------------------------------------------------------------+
void OnInit() 
  {
   string s="EURUSD";
   for(int i=1; i<=2; i++) 
     {
      Print("Indicator is created, handle=",
            iCustom(s,_Period,"i-flood",IntegerToString(i)));
      s="GBPUSD";
     }
  }
//+------------------------------------------------------------------+

よって、コア計算のすべては実行され、これで MetaTrader がどの組合せに対して複数コアを使用するかわかります。次に、この知識を並列計算の考え方に取り入れてみようと思います。

並列システムを設計します

並列システム用のトレーディング端末というのは、たとえばトレードを行ったり、チャート上の描画を行うなどの共通タスクを共に処理するインディケータや EA（または両者混合）のグループを指します。このグループは、一つの大きなインディケータまたは一つの大きな EA であるという考え方です。しかし同時に、すべての利用可能なプロセッサ コアの計算負荷を分配します。

そのようなシステムは2種類のソフトウェア コンポーネントで構成されます。

• CM ： 計算モジュールそれらの数は2～プロセッサコア数です。並列が配置される必要のあるコアすべてのCMにあります。前章で確認したように、CMはインディケータ、またはEAとして実装可能です。どのような実装形式にしても、組み合わせがあり、それはすべてのプロセッサコアを使用します。
• MM ： 主要モジュールシステムの主な関数を処理します。よって MMがインディケータであれば、チャート上の描画を行い、MMがEAであればトレーディング関数を処理します。MM はまた、すべてのCM を管理します。

たとえば、MM EA については、2つのコアプロセッサで、システム動作のスキームは以下のようなものとなります。

図3　2 CPU コアを伴うスキーム

われわれのプログラムは、定められた時に必要なプロシージャを呼ぶという従来のプログラムとは異なることをご理解ください。MM および CM は EAまたはインディケータで、すなわちこれは自立したスタンドアロン プログラムです。それらの間に直接の連携はありません。おのおの自立的に処理を行い、お互い直接通信することはありえません。

こういったプログラムはどれもイベント（たとえば、相場の着信またはタイマーのティック）が端末内に出現したときのみ実行を開始します。イベント間には、これらプログラムが相互に伝達したいデータはすべて外部、パブリックにアクセスできる場所（『データ交換バッファ』と呼びます。）に格納される必要があります。よって上記のスキームは以下の方法で端末に実装されます。

図4　実装詳細

本システムの実装については、以下の質問に答える必要があります。

• 前章で判明したどの複数コアの組合せをわれわれのシステムで使用するのか？
• システムは複数の EAまたはインディケータで構成されているため、相互間データ交換（双方向）を行うよりよい方法は（クリップボードデータは物理的にどのような姿をしているのか）？
• それらの処理をどのように連携させ、同期させることができるのか？

質問それぞれに対しては、2つ以上の答えがありますが、すべて以下に記されています。実際、特定の選択肢は特別な状況に応じて選択される必要があります。これは次章で行いたいと思います。ここでは、可能な回答について考察します。

組合せ

組合せ7 は標準的なもので、実用面でもっとも都合のよい組合せです。（その他の組合せはすべて前章に挙げています。）理由としては、端末で余分なウィンドウを開いて EAやインディケータをそこに配置する必要がないことです。システム全体は単一のウィンドウに配置され、インディケータ (CM-1 および CM-2) はすべて EA (MM)によって自動的に作成されます。余分なウィンドウやマニュアル操作がないことで、トレーダーは混乱を避けることができ、そういった混乱に伴うエラーも防ぐことができます。

トレーディング戦略の中には、別の組合せの方が効果的だというものもあるでしょう。たとえば、ある一つについて、『クライアントサーバー』の原則に基づき作動するソフトウェアシステム全体を作成することも可能です。同じCMが複数のMMに対して共通な場合、そのような共通の CM は『コンピュータ』の二次的役割をするだけでなく、統一された類の全戦略情報を格納する『サーバー』として、あるいは包括的動作の調整役の役目を果たします。たとえばCMサーバーは戦略や通貨ペアのポートフォリオにおける資産の分配を中央で管理し、一方で求める全般的なリスクレベルを維持することが可能かもしれません。

データ交換

以下の3とおりの方法によりMM 、CM 間の情報送信が可能です。

1. 端末のグローバル関数
2. ファイル
3. インディケータ バッファ

最初の手法は変換される数字変数が少数ある場合に最適です。テキストデータの変換が必要な場合は、ある程度数字にコード化される必要があります。グローバル変数はダブルタイプのみ扱うためです。

代替えとしては2番目の手法です。これはファイルにはどんなことも書き込み可能だからです。そして、これは大量データを変換する必要があるときは都合のよい（そしておそらく1番目より速い）手法です。

3番目の手法は MM および CM が インディケータである場合に適しています。ダブルタイプのデータのみ変換可能ですが、大きな数値配列を変換するには便利です。しかし、欠点もあります。新規バー形成中には、バッファ内エレメントの番号はずれてしまいます。MM と CM の通貨ペアは異なるため、新規バーは同時に作成されません。よってこのズレを考慮する必要があります。

同期

端末がMMに対するクオートを受信し、その処理を始めるとCMへのコントロールに即座に変換するわけではありません。ただタスクを作成し（それをグローバル変数、ファイル、またはインディケータバッファにセットします。）、CM が実行されるのを待ちます。（上に図示されているとおりです。）CMはすべて異なる通貨ペアにセットされているため、待ち時間がかかるのです。これはあるペアがクオートを受信し、一方で別のペアはまだそれを受信せず、そのペアがクオートを受信するまで数秒から数分かかります。（たとえば、夜間非流動性のペアについて発生する可能性があります。）

よってCM がコントロールを取得するには、クオート依存の OnTick イベントおよび OnCalculateイベントは使用するべきではありません。代わりに、OnTimer イベント（MQL5の改良です）を使用する必要があります。これは指定の頻度（たとえば1秒）で実行されるものです。この場合、システムの遅延は著しく制限されます。

また、OnTimerの代わりに、サイクル手法の利用も可能です。というのは OnInit または OnCalculate にCMに対する無限サイクルをセットするのです。それぞれのループはタイマーティックの類似体です。

注意： いくつか実験を行い、組合せ7を使う場合、タイマーが問題なく作成されていても、OnTimer イベントはインディケータ内で動作しない（なんらかの理由により）ことを知りました。

また、OnInit および OnCalculateの無限ループには注意が必要です。というのも、一つの CMインディケータが MM-EAと同一の通貨ペアにセットされていたとしたら、EA は動作を停止します。（OnTick イベントの作成が中断されます）端末開発者が減少理由を説明しています。

開発者談：スクリプトおよび EAは個別のスレッドにて動作します。一方でインディケータはすべて単一シンボルに対して同一スレッドで動作します。インディケータと同一ストリームでこのシンボルに対する他のすべての処理が連続して実行されます。その処理はティック処理、履歴の同期、インディケータ計算などです。よって、インディケータが無限処理を行うと、そのシンボルに対する他のすべてのイベントはまったく実行されなくなります。

プログラム	実行	備考
スクリプト	自身のスレッド内には、スクリプトと同じ数の実行スレッドが存在します。	循環スクリプトは他のプログラム動作を妨害することはありません。
Expert Advisor	自身のスレッド内には、EAにあると同じ数の実行スレッドが存在します。	循環スクリプトは他のプログラム動作を妨害することはありません。
インディケータ	あるシンボルのすべてのインディケータに対する単一の実行スレッドインディケータには、それに対する実行スレッドと同数のシンボルが存在します。	一つのインディケータ内の無限サイクルはそのシンボルについての他のインディケータすべての動作を停止します。

検証用 Expert Advisorの作成

いくつかトレーディング戦略を選択します。それは並列化する意味のあるもので、これに適したアルゴリズムのものとします。

たとえば、これはシンプルな戦略です。前回バーNからのシーケンスをコンパイルし、履歴内でこのシーケンスにもっとも似たものを探します、というもの。履歴内で価格がどこに移動したか知るために、適当な取引をオープンします。

シーケンス長が比較的小さければ、この戦略はMetaTrader 5 でひじょうに早く動作します。数秒でしょうか。また、長いシーケンスを取れば、たとえば過去24時間について時間枠M1のバーすべて（1,440バーにあたります）、また履歴を過去1年にさかのぼって検索する（約375,000バー）などとすると、それにはかなり時間がかかります。ですが、この検索は簡単に並列化することが可能です。利用可能なプロセッサコア数に履歴を等分し、各コアを特定のロケーション検索に割り当てるだけでよいのです。

並列システムのパラメータは以下のとおりです。

• MM はパターン化されたトレーディング戦略を実装する EAです。
• 並列計算はEaから自動的に作成される CMインディケータで行われます。（組合せ7を使用）
• CMインディケータ内の計算コードはOnInitの無限サイクルにセットします。
• MM-EA 、CMインディケータ間のデータ交換は端末のグローバル変数を用いて行われます。

開発とその後の使用時都合がよいように、設定に応じて通常の（インディケータの使用をしない）EAとして並列処理を行う（インディケータ内計算を用いて）ようにEAを作成します。取得されたe-MultiThread Expert Advisorのコード

//+------------------------------------------------------------------+
//|                                                e-MultiThread.mq5 |
//+------------------------------------------------------------------+
input int Threads=1; // How many cores should be used
input int MagicNumber=0;

// Strategy parameters
input int PatternLen  = 1440;   // The length of the sequence to analyze (pattern)
input int PrognozeLen = 60;     // Forecast length (bars)
input int HistoryLen  = 375000; // History length to search

input double Lots=0.1;
//+------------------------------------------------------------------+
class IndData
  {
public:
   int               ts,te;
   datetime          start_time;
   double            prognoze,rating;
  };

IndData Calc[];
double CurPattern[];
double Prognoze;
int  HistPatternBarStart;
int  ExistsPrognozeLen;
uint TicksStart,TicksEnd;
//+------------------------------------------------------------------+
#include <ThreadCalc.mqh>
#include <Trade\Trade.mqh>
//+------------------------------------------------------------------+
int OnInit()
  {

   double rates[];

//--- Make sure there is enough history
   int HistNeed=HistoryLen+Threads+PatternLen+PatternLen+PrognozeLen-1;
   if(TerminalInfoInteger(TERMINAL_MAXBARS)<HistNeed)
     {    
      Print("Change the terminal setting \"Max. bars in chart\" to the value, not lesser than ",
            HistNeed," and restart the terminal");
      return(1);      
     }
   while(Bars(_Symbol,_Period)<HistNeed)
     {      
      Print("Insufficient history length (",Bars(_Symbol,_Period),") in the terminal, upload...");
      CopyClose(_Symbol,_Period,0,HistNeed,rates);
     }
   Print("History length in the terminal: ",Bars(_Symbol,_Period));

//--- For a multi-core mode create computational indicators
   if(Threads>1)
     {
      GlobalVarPrefix="MultiThread_"+IntegerToString(MagicNumber)+"_";
      GlobalVariablesDeleteAll(GlobalVarPrefix);

      ArrayResize(Calc,Threads);

      // Length of history for each core
      int HistPartLen=MathCeil(HistoryLen/Threads);
      // Including the boundary sequences
      int HistPartLenPlus=HistPartLen+PatternLen+PrognozeLen-1;

      string s;
      int snum=0;
      // Create all computational indicators
      for(int t=0; t<Threads; t++)
        {      
         // For each indicator - its own currency pair,
         // it should not be the same as for the EA
         do
            s=SymbolName(snum++,false);
         while(s==_Symbol);

         int handle=iCustom(s,_Period,"i-Thread",
                            GlobalVarPrefix,t,_Symbol,PatternLen,
                            PatternLen+t*HistPartLen,HistPartLenPlus);

         if(handle==INVALID_HANDLE) return(1);
         Print("Indicator created, pair ",s,", handle ",handle);
        }
     }

   return(0);
  }
//+------------------------------------------------------------------+
void OnTick()
  {
   TicksStart=GetTickCount();

   // Fill in the sequence with the last bars
   while(CopyClose(_Symbol,_Period,0,PatternLen,CurPattern)<PatternLen) Sleep(1000);

   // If there is an open position, measure its "age"
   // and modify the forecast range for the remaining 
   // planned life time of the deal
   CalcPrognozeLen();

   // Find the most similar sequence in the history
   // and the forecast of the movement of its price on its basis
   FindHistoryPrognoze();

   // Perform the necessary trade actions
   Trade();

   TicksEnd=GetTickCount();
   // Debugging information in
   PrintReport();
  }
//+------------------------------------------------------------------+
void FindHistoryPrognoze()
  {
   Prognoze=0;
   double MaxRating;

   if(Threads>1)
     {
      //--------------------------------------
      // USE COMPUTATIONAL INDICATORS
      //--------------------------------------
      // Look through all of the computational indicators 
      for(int t=0; t<Threads; t++)
        {
         // Send the parameters of the computational task
         SetParam(t,"PrognozeLen",ExistsPrognozeLen);
         // "Begin computations" signal 
         SetParam(t,"Query");
        }

      for(int t=0; t<Threads; t++)
        {
         // Wait for results
         while(!ParamExists(t,"Answer"))
            Sleep(100);
         DelParam(t,"Answer");

         // Obtain results
         double progn        = GetParam(t, "Prognoze");
         double rating       = GetParam(t, "Rating");
         datetime time[];
         int start=GetParam(t,"PatternStart");
         CopyTime(_Symbol,_Period,start,1,time);
         Calc [t].prognoze   = progn;
         Calc [t].rating     = rating;
         Calc [t].start_time = time[0];
         Calc [t].ts         = GetParam(t, "TS");
         Calc [t].te         = GetParam(t, "TE");

         // Select the best result
         if((t==0) || (rating>MaxRating))
           {
            MaxRating = rating;
            Prognoze  = progn;
           }
        }
     }
   else
     {
      //----------------------------
      // INDICATORS ARE NOT USED
      //----------------------------
      // Calculate everything in the EA, into one stream
      FindPrognoze(_Symbol,CurPattern,0,HistoryLen,ExistsPrognozeLen,
                   Prognoze,MaxRating,HistPatternBarStart);
     }
  }
//+------------------------------------------------------------------+
void CalcPrognozeLen()
  {
   ExistsPrognozeLen=PrognozeLen;

   // If there is an opened position, determine 
   // how many bars have passed since its opening
   if(PositionSelect(_Symbol))
     {
      datetime postime=PositionGetInteger(POSITION_TIME);
      datetime curtime,time[];
      CopyTime(_Symbol,_Period,0,1,time);
      curtime=time[0];
      CopyTime(_Symbol,_Period,curtime,postime,time);
      int poslen=ArraySize(time);
      if(poslen<PrognozeLen)
         ExistsPrognozeLen=PrognozeLen-poslen;
      else
         ExistsPrognozeLen=0;
     }
  }
//+------------------------------------------------------------------+
void Trade()
  {

   // Close the open position, if it is against the forecast
   if(PositionSelect(_Symbol))
     {
      long type=PositionGetInteger(POSITION_TYPE);
      bool close=false;
      if((type == POSITION_TYPE_BUY)  && (Prognoze <= 0)) close = true;
      if((type == POSITION_TYPE_SELL) && (Prognoze >= 0)) close = true;
      if(close)
        {
         CTrade trade;
         trade.PositionClose(_Symbol);
        }
     }

   // If there are no position, open one according to the forecast
   if((Prognoze!=0) && (!PositionSelect(_Symbol)))
     {
      CTrade trade;
      if(Prognoze > 0) trade.Buy (Lots);
      if(Prognoze < 0) trade.Sell(Lots);
     }
  }
//+------------------------------------------------------------------+
void PrintReport()
  {
   Print("------------");
   Print("EA: started ",TicksStart,
         ", finished ",TicksEnd,
         ", duration (ms) ",TicksEnd-TicksStart);
   Print("EA: Forecast on ",ExistsPrognozeLen," bars");

   if(Threads>1)
     {
      for(int t=0; t<Threads; t++)
        {
         Print("Indicator ",t+1,
               ": Forecast ", Calc[t].prognoze,
               ", Rating ", Calc[t].rating,
               ", sequence from ",TimeToString(Calc[t].start_time)," in the past");
         Print("Indicator ",t+1,
               ": started ",  Calc[t].ts,
               ", finished ",   Calc[t].te,
               ", duration (ms) ",Calc[t].te-Calc[t].ts);
        }
     }
   else
     {
      Print("Indicators were not used");
      datetime time[];
      CopyTime(_Symbol,_Period,HistPatternBarStart,1,time);
      Print("EA: sequence from ",TimeToString(time[0])," in the past");
     }

   Print("EA: Forecast ",Prognoze);
   Print("------------");
  }
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
  {
   // Send the "finish" command to the indicators
   if(Threads>1)
      for(int t=0; t<Threads; t++)
         SetParam(t,"End");
  }
//+------------------------------------------------------------------+

Expert Advisorによって使用される計算インディケータのコード i-Thread

//+------------------------------------------------------------------+
//|                                                     i-Thread.mq5 |
//+------------------------------------------------------------------+
#property indicator_chart_window
#property indicator_chart_window
#property indicator_buffers 1
#property indicator_plots   1

//--- input parameters
input string VarPrefix;  // Prefix for global variables (analog to MagicNumber)
input int    ThreadNum;  // Core number (so that indicators on different cores could
                        // differentiate their tasks from the tasks of the "neighboring" cores)
input string DataSymbol; // On what pair is the MM-EA working
input int    PatternLen; // Length of the sequence for analysis
input int    BarStart;   // From which bar in the history the search for a similar sequence began
input int    BarCount;   // How many bars of the history to perform a search on

//--- indicator buffers
double Buffer[];
//---
double CurPattern[];

//+------------------------------------------------------------------+
#include <ThreadCalc.mqh>
//+------------------------------------------------------------------+
void OnInit()
  {
   SetIndexBuffer(0,Buffer,INDICATOR_DATA);

   GlobalVarPrefix=VarPrefix;

   // Infinite loop - so that the indicator always "listening", 
   // for new commands from the EA
   while(true)
     {
      // Finish the work of the indicator, if there is a command to finish
      if(ParamExists(ThreadNum,"End"))
         break;

      // Wait for the signal to begin calculations
      if(!ParamExists(ThreadNum,"Query"))
        {
         Sleep(100);
         continue;
        }
      DelParam(ThreadNum,"Query");

      uint TicksStart=GetTickCount();

      // Obtain the parameters of the task
      int PrognozeLen=GetParam(ThreadNum,"PrognozeLen");

      // Fill the sequence from the last bars
      while(CopyClose(DataSymbol,_Period,0,PatternLen,CurPattern)
            <PatternLen) Sleep(1000);

      // Perform calculations
      int HistPatternBarStart;
      double Prognoze,Rating;
      FindPrognoze(DataSymbol,CurPattern,BarStart,BarCount,PrognozeLen,
                   Prognoze,Rating,HistPatternBarStart);

      // Send the results of calculations
      SetParam(ThreadNum,"Prognoze",Prognoze);
      SetParam(ThreadNum,"Rating",Rating);
      SetParam(ThreadNum,"PatternStart",HistPatternBarStart);
      SetParam(ThreadNum,"TS",TicksStart);
      SetParam(ThreadNum,"TE",GetTickCount());
      // Signal "everything is ready"
      SetParam(ThreadNum,"Answer");
     }
  }
//+------------------------------------------------------------------+
int OnCalculate(const int rates_total,
                const int prev_calculated,
                const int begin,
                const double &price[])
  {
   // The handler of this event is required
   return(0);
  }
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
  {
   SetParam(ThreadNum,"End");
  }
//+------------------------------------------------------------------+

Expert Advisor とインディケータは共通の ThreadCalc.mqh ライブラリを使用します。

以下がコードです。

//+------------------------------------------------------------------+
//|                                                   ThreadCalc.mqh |
//+------------------------------------------------------------------+
string GlobalVarPrefix;
//+------------------------------------------------------------------+
// It finds the price sequence, most similar to the assigned one.
// in the specified range of the history
// Returns the estimation of similarity and the direction 
// of the further changes of prices in history.
//+------------------------------------------------------------------+
void FindPrognoze(
                  string DataSymbol,    // symbol
                  double  &CurPattern[],// current pattern
                  int BarStart,         // start bar
                  int BarCount,         // bars to search
                  int PrognozeLen,      // forecast length

                  // RESULT
                  double  &Prognoze,        // forecast (-,0,+)
                  double  &Rating,          // rating
                  int  &HistPatternBarStart // starting bar of the found sequence
                  ) 
  {

   int PatternLen=ArraySize(CurPattern);

   Prognoze=0;
   if(PrognozeLen<=0) return;

   double rates[];
   while(CopyClose(DataSymbol,_Period,BarStart,BarCount,rates)
         <BarCount) Sleep(1000);

   double rmin=-1;
   // Shifting by one bar, go through all of the price sequences in the history
   for(int bar=BarCount-PatternLen-PrognozeLen; bar>=0; bar--) 
     {
      // Update to eliminate the differences in the levels of price in the sequences
      double dr=CurPattern[0]-rates[bar];

      // Calculate the level of differences between the sequences - as a sum 
      // of squares of price deviations from the sample values
      double r=0;
      for(int i=0; i<PatternLen; i++)
         r+=MathPow(MathAbs(rates[bar+i]+dr-CurPattern[i]),2);

      // Find the sequence with the least difference level
      if((r<rmin) || (rmin<0)) 
        {
         rmin=r;
         HistPatternBarStart   = bar;
         int HistPatternBarEnd = bar + PatternLen-1;
         Prognoze=rates[HistPatternBarEnd+PrognozeLen]-rates[HistPatternBarEnd];
        }
     }
   // Convert the bar number into an indicator system of coordinates
   HistPatternBarStart=BarStart+BarCount-HistPatternBarStart-PatternLen;

   // Convert the difference into the rating of similarity
   Rating=-rmin;
  }
//====================================================================
// A set of functions for easing the work with global variables.
// As a parameter contain the number of computational threads 
// and the names of the variables, automatically converted into unique
// global names.
//====================================================================
//+------------------------------------------------------------------+
string GlobalParamName(int ThreadNum,string ParamName) 
  {
   return GlobalVarPrefix+IntegerToString(ThreadNum)+"_"+ParamName;
  }
//+------------------------------------------------------------------+
bool ParamExists(int ThreadNum,string ParamName) 
  {
   return GlobalVariableCheck(GlobalParamName(ThreadNum,ParamName));
  }
//+------------------------------------------------------------------+
void SetParam(int ThreadNum,string ParamName,double ParamValue=0) 
  {
   string VarName=GlobalParamName(ThreadNum,ParamName);
   GlobalVariableTemp(VarName);
   GlobalVariableSet(VarName,ParamValue);
  }
//+------------------------------------------------------------------+
double GetParam(int ThreadNum,string ParamName) 
  {
   return GlobalVariableGet(GlobalParamName(ThreadNum,ParamName));
  }
//+------------------------------------------------------------------+
double DelParam(int ThreadNum,string ParamName) 
  {
   return GlobalVariableDel(GlobalParamName(ThreadNum,ParamName));
  }
//+------------------------------------------------------------------+

複数のプロセッサコアが使用可能なわれわれのトレーディングシステムができました！

使用に際しては、この例では無限ループを使用した CMインディケータを採用していることを忘れないようにします。

このシステムを用いて端末で他のプログラムを実行するのであれば、CMインディケータで使われていない通貨ペアに関してこのプログラムを使用する必要があります。そのようなコンフリクトを避けるには MM-EA の入力パラメータでCMインディケータに対する通貨ペアを直接指定できるようシステムを変更するのがよいでしょう。

EA動作速度の計測

標準モード

EURUSD M1 チャートを開き、前章で作成したExpert Advisorを起動します。パターン長指定を24時間（1,440バー）とし、履歴検索の深さを1年（375,000バー）と設定します。

図4　Expert Advisor 入力パラメータ

『スレッド』パラメータを1に設定します。これは、EA 計算のすべてが1つのスレッド（シングルコア）で作成されることを意味します。計算インディケータは使用しませんが、それ自体ですべてを計算します。基本的に標準EAの動作原則に従っています。

その実行ログです。

図6　Expert Advisor ログ (1 thread)

並列モード

この EA とそれによってオープンされたポジションを削除します。ふたたびEA を追加しますが、今回はパラメータ『スレッド』を2と設定します。

ここで、EA は2つのプロセッサコアを取り、2つの計算インディケータを作成し、動作に使用する必要があります。その実行ログです。

図7　Expert Advisor ログ （2 スレッド）

スピード比較

両ログを分析すると、EA のおよその実行スピードについて以下の結論を得ます。

• 標準モード - 52秒
• 2コアモード - 27 秒

よって2コアの CPUで並列処理をすることで1.9 倍EAのスピードを高めることができたのです。大容量コアのプロセッサを使用する場合、コア数に比例して実行速度はもっと早くなるであろう、と予想することができます。

動作の正確性管理

実行時間の上にログは別の情報も与えてくれます。それはすべての計測が正確に行われたことを確信させてくれるものです。「EA: 動作開始 ... 操作終了... 」 ライン、および「インディケータ ...: 動作開始 ... 動作終了 ...」 は EAがコマンドを与える1秒前にはインディケータはまだ計算を開始していないということを示しています。

EAを並列モードで起動している間、トレーディング戦略に妨害はないということを確認します。ログによると、EAは標準モードで起動したほとんど直後に並列モードで起動しているというのは明らかです。よってどちらの場合もマーケット状況は似ていると言えます。ログはパターンの履歴に見える日付はどちらの場合もたいへん似ているということを示しています。どいうことですべてはうまくいっています。戦略アルゴリズムはどちらの場合にもきわめて良好に動作しています。

ログ状況に述べられているパターンは以下です。それがEAを標準モードで実行したとき時点のマーケット状況です。（長さ ：1,440分足）

図8　マーケット現況

EA は履歴に次のような似たパターンを確認しました。

図9　類似のマーケット状況

EA を並列モードで実行する場合、「インディケータ1」により同様のパターンが確認されました。ログによると以下のように「インディケータ2」は履歴の別の半年にパターンを検索しており、それゆえ異なる類似パターンを見出しました。

図10　類似のマーケット状況

そして以下がEAが並列で動作している間、MetaTrader 5 の グローバル変数です。

図11　グローバル変数

これでグローバル変数 を用いたEAとインディケータ間のデータ交換は問題なく実装されました。

おわりに

今回の調査では、MetaTrader 5の標準手法によってリソースに富んだアルゴリズムを並列化するのは可能であると確認しました。また、この問題に対して発見された解決法は現実のトレーディング戦略で使用するのに便利です。

本プログラムは複数コアシステムで、まちがいなく比例的に速く動作します。プロセッサ内のコア数は年々増加し、MetaTraderを使っているトレードがこういったハードウェアリソースを効果的に利用する機会があるのはすばらしいことです。それにより、さらにリアルタイムでマーケットを分析することのできる、よりリソースに富んだトレーディング戦略を安全に作成することが可能となるのです。