处理器并行性简介

由于具有多个处理器核心，几乎所有现代 PC 都能够同时执行多个任务。核心的数量逐年增长 – 2、3、4、6…Intel 最近展示了一台运行中的试验性 80 核心处理器（是的，这不是笔误，就是 80 核心；不幸的是，这样一台电脑不会出现在市面上，因为该处理器仅为探索技术的潜能而制造）。

不是所有的电脑用户（甚至不是所有的新手编程人员）都了解它是如何工作的。因此，有人肯定会问：为什么我们需要有这么多核心的处理器？即使在过去（单核时），电脑也可同时执行多个程序且这些程序都能运行。真相是，这并不是事实。我们来看一看下面的示意图。

图 1. 应用程序的并行执行

示意图的实例 A 显示了单个程序在单核处理器上运行的情形。处理器将其全部时间用于自身的实施，且程序随着时间 T 的推移执行一定量的工作。

实例 B – 启动 2 个程序。但处理器以这样一种方式进行安排，即物理上在任何时间点，某个核心只能执行一个命令，因此它必须在两个程序间不停地切换：它将执行第一个程序的部分工作，然后转为执行第二个程序。这种切换极为迅速，一秒钟可执行多次，因此看上去处理器像是在同时处理两个程序。然而事实上，程序的执行要比它们在处理器上单独执行多花一倍的时间。

实例 C 显示上述问题得到了有效的解决，如果处理器中的核心数与运行的程序数目相匹配的话。每个程序有了可以供其使用的单独核心，因此执行速度加快，就如实例 A 中的情形一样。

实例 D 反映了很多用户都有的一个常见错觉。他们认为，如果程序在多核心处理器上执行，则程序执行的速度可缩短为原来的几分之一。但通常这并不是事实，因为处理器无法独立将程序分为几个部分，然后同时执行它们。

例如，如果程序首先要求密码，然后执行验证，则同一时间在一个核心上执行密码提示和在另一个核心上执行验证是不可接受的。验证根本不会成功，因为密码在一开始时就并未输入。

处理器无法知晓程序员实施的所有设计，也不知道程序工作的整个逻辑，因此它无法独立地将程序在各核心之间进行划分。所以，如果我们在多核心系统上运行单个程序，它仅会使用一个核心，并且执行速度与在单核处理器上执行的速度相同。

实例 E 解释了要使程序使用所有核心以加快执行需要做的工作。由于编程人员熟悉程序的逻辑，他应该在程序开发期间标示出可以同时执行的程序部分。然后程序在其执行期间将该信息传达给处理器，处理器则将所需的核心数分配给程序。

MetaTrader 中的并行性

在前面的章节中，我们已指出要使用所有的 CPU 核心以及加快程序的执行需要做的工作。我们需要将程序可并行的代码部分分配至单独的线程。在很多编程语言中，有专门的类或运算符用于此目的。但是在 MQL5 语言中，还没有这种内置工具。那我们该如何做呢？

我们有两种方法解决此问题：

在本文中，我们不讨论第一种方法。显而易见，我们可通过 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 开始");
   
   for (int i=0; i<1e9; i++)
     for (int j=0; j<1e1; j++);
     
   Print(id,": OnCalculate 结束");
   return(0);   
  }
//+------------------------------------------------------------------+

以及与其类同的 e-flood EA：

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

此外，通过打开各种图表窗口的组合（一个图表、具有相同交易品种的两个图表、具有不同交易品种的两个图表）以及在其上附加一个或两个这种指标或 EA，我们可以观察终端使用 CPU 核心的方式。

这些指标和 EA 也会发送消息至日志，观察它们出现的顺序十分有趣。我没有提供这些日志，因为您可以自己生成；在本文中，我们感兴趣的是找出终端使用的核心数以及是在什么样的图表组合下使用。

我们可以通过 Windows 的“任务管理器”来衡量工作核心数目：

图 2. CPU 核心

所有的测量结果收录于下表中：

第 7 种组合是创建指标的常用方式，在许多交易策略中使用。

唯一特别的是，我在两个不同的货币对上创建了两个指标，因为组合 1 和组合 2 清楚表明在同一货币对上放置多个指标是毫无意义的。对于这种组合，我使用 EA e-flood-starter，它生成了 i-flood 的两个副本：

//+------------------------------------------------------------------+
//|                                              e-flood-starter.mq5 |
//+------------------------------------------------------------------+
void OnInit() 
  {
   string s="EURUSD";
   for(int i=1; i<=2; i++) 
     {
      Print("指标已创建, 句柄=",
            iCustom(s,_Period,"i-flood",IntegerToString(i)));
      s="GBPUSD";
     }
  }
//+------------------------------------------------------------------+

因此，核心的所有计算均得到执行，而且现在我们也了解了在哪种组合情况下 MetaTrader 会使用多个核心。接下来，我们尝试将这些知识应用至并行计算理念的实施。

设计并行系统

当提及并行系统的交易终端时，我们指的是共同执行某些常见任务（例如，执行交易或在图表上绘制）的一组指标或 EA（或是两者均有）。即该组作为一个大的指标或大的 EA 工作，但同时将计算负载分配至所有可用的处理器核心。

这样的一个系统包含两种类型的软件构件：

• CM – 计算模块。它们的数量可以从 2 一直到处理器核心的数目。需要并行处理的所有代码正是放置在 CM 中。 从前面的章节中我们可以得知，CM 可作为一个指标或 EA 来实施 – 对于任何实施形式，都存在一个使用所有处理器核心的组合；
• MM – 主模块。执行系统的主要功能。因此，如果 MM 是一个指标，则它在图表上进行绘制；如果 MM 是一个 EA，则它执行交易功能。MM 还管理所有的 CM。

例如，对于 MM EA 和一个 2 核处理器，系统工作的架构将如下所示：

图 3. 具有 2 个 CPU 核心的系统的架构。

我们需要认识到，我们开发的系统不是您在特定的时刻可按需调用的传统程序。MM 和 CM 是 EA 或指标，即独立和单独的程序。它们独立运作，没有直接的联系，不能直接与对方沟通。

这些程序的执行仅在终端中出现任意事件（例如，报价或计时器嘀嗒信号到来）时开始。而在事件之间，这些程序希望互相传递的所有数据必须存储在这之外的一个可公共访问的地方（我们称之为“数据交换缓冲区”）。因此，上述架构在终端中的实施方式如下所示：

图 4. 实施细节

对于该系统的实施，我们需要解决以下问题：

• 我们将在系统中使用我们在前面的章节中得出的哪种多核心组合？
• 由于系统由多个 EA 或指标组成，我们如何能更好地组织它们之间的数据交换（双向），即剪贴板上的数据在物理上是怎样的？
• 我们如何组织它们的行动协调和同步？

这些问题都不止有一个答案，我将在后文中一一列示。在实践中，具体的选择应根据具体情况作出。我们将在下一章中讨论这部分内容。同时，让我们思索所有可能的答案。

组合

组合 7 对于日常实际应用而言是最方便的（所有其他的组合均在前面的章节中给出），因为无需在终端中打开额外的窗口并在其上放置 EA 或指标。整个系统位于单个窗口内，且所有指标（CM-1 和 CM-2）由 EA (MM) 自动创建。无需额外的窗口和手动操作消除了交易人员的困惑，以及与之相关的错误。

在某些交易策略中，其他组合可能更为有用。例如，我们可以基于它们中的任何一个创建运行于“主从式”原则的整个软件系统。其中相同的 CM 对于多个 MM 通用。这些通用 CM 不仅可以作为“计算器”的次要角色，还可以作为存储某类统一策略信息的“服务器”，甚至作为它们集体工作的协调器。例如，CM 服务器能够集中控制某些投资组合策略和货币对的均值分布，同时保持所需的整体风险水平。

数据交换

我们可以使用下列 3 种方式中的任意一种在 MM 和 CM 之间传输信息：

1. 终端的全局变量；
2. 文件；
3. 指标缓冲区。

第 1 种方法在只有少量数字变量传输时是最优的。如果需要传输文本数据，需要将文本数据编码为数字，因为全局变量仅有双精度类型。

这时可以选择第 2 种方法，因为任何类型的数据都可以写入文件。并且这在需要传输大量数据的情形下是很方便的方法（且可能快过第 1 种方法）。

第 3 种方法适合于 MM 和 CM 均为指标的情况。只有双进度类型的数据可被传输，但在传输大型数字数组时要更为方便。但该方法也存在一个缺点：在新柱的形成期间，缓冲区中元素的编号是偏移的。因为 MM 和 CM 位于不同的货币对上，新柱将不会同时出现。我们必须将这些偏移考虑在内。

同步

当终端接收到针对 MM 的报价时即开始处理报价，它无法立即将控制传递至 CM。它仅能（如上面的示意图所示）形成一个任务（将其置于全局变量、文件或指标缓冲区中），然后等待 CM 执行。由于所有的 CM 位于不同的货币对上，等待可能需要花费一些时间。这是因为某个货币对可能接收到报价，而其他货币对尚未接收到，且报价每隔数秒甚至数分钟到来一次（例如，这可能发生在夜间的非流动性货币对上）。

因此，要使 CM 获得控制，我们不应使用基于报价的 OnTick 和 OnCalculate 事件。相反，我们需要使用 OnTimer 事件（MQL5 的创新），它基于指定的频率执行（例如 1 秒）。在这种情况下，系统的延迟就得到了有效的遏止。

此外，除了 OnTimer 事件，我们还可以使用循环技术：即在 OnInit 或 OnCalculate 事件中针对 CM 设置一个无限循环。循环的每次迭代即是对计时器嘀嗒信号的模拟。

创建测试“EA 交易”

让我们选择一些具有并行化意义的交易策略，以及合适的算法。

例如，它可以是简单策略：编译从 N 开始的最近的柱的序列，并在历史数据中找出与其最为相似的序列。了解历史数据中价格发生偏移的位置后，我们打开相关交易。

如果序列的长度相对较小，该策略在 MetaTrader 5 中运行十分迅速 – 仅需数秒。但如果我们选择一个较大的长度编译 – 例如，时间表 M1 在最近 24 小时内的所有柱（将有 1440 个柱），且如果我们往回搜索一年的历史数据（约 375,000 个柱），这将需要大量的时间。然而，该搜索可轻松地并行化：基于可用的处理器核心数目等分历史数据，然后为每个核心分配特定的搜索区域即已足够。

并行系统的参数如下所示：

• MM – 实施图形化交易策略的 EA；
• 并行计算在自 EA 自动生成（即使用组合 7）的 CM 指标中完成；
• CM 指标中的计算代码位于 OnInit 的无限循环内部；
• MM-EA 和 CM 指标间的数据交换通过终端的全局变量实现。

为便于开发和后续使用，我们将以这样的方式创建 EA（视设置而定）：使它能够作为并行（在指标中计算）EA 和作为普通（即不使用指标）EA 运行。获得的 e-MultiThread“EA 交易”的代码如下：

//+------------------------------------------------------------------+
//|                                                e-MultiThread.mq5 |
//+------------------------------------------------------------------+
input int Threads=1; // 应该使用多少个内核
input int MagicNumber=0;

// 策略参数
input int PatternLen  = 1440;   // 分析序列 (模式) 的长度
input int PrognozeLen = 60;     // 预测长度 (柱数)
input int HistoryLen  = 375000; // 搜索的历史长度

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[];

//--- 确保有足够的历史
   int HistNeed=HistoryLen+Threads+PatternLen+PatternLen+PrognozeLen-1;
   if(TerminalInfoInteger(TERMINAL_MAXBARS)<HistNeed)
     {    
      Print("修改终端设定 /". 图表上的最大柱数/" 为指定值, 不能小于 ",
            HistNeed," 并重新启动终端");
      return(1);      
     }
   while(Bars(_Symbol,_Period)<HistNeed)
     {      
      Print("终端中的历史长度不足 (",Bars(_Symbol,_Period),") , 上传中...");
      CopyClose(_Symbol,_Period,0,HistNeed,rates);
     }
   Print("终端中的历史长度: ",Bars(_Symbol,_Period));

//--- 为多核模式创建计算指标
   if(Threads>1)
     {
      GlobalVarPrefix="MultiThread_"+IntegerToString(MagicNumber)+"_";
      GlobalVariablesDeleteAll(GlobalVarPrefix);

      ArrayResize(Calc,Threads);

      // 每个核心的历史长度
      int HistPartLen=MathCeil(HistoryLen/Threads);
      // 包含边界序列
      int HistPartLenPlus=HistPartLen+PatternLen+PrognozeLen-1;

      string s;
      int snum=0;
      // 创建所有用于计算的指标
      for(int t=0; t<Threads; t++)
        {      
         // 对于每个指标 - 它们自己的货币对,
         // 它不应该和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("指标已创建, 货币对 ",s,", 句柄 ",handle);
        }
     }

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

   // 使用最近的柱填充序列
   while(CopyClose(_Symbol,_Period,0,PatternLen,CurPattern)<PatternLen) Sleep(1000);

   // 如果有开启的仓位, 衡量它的 "年龄"
   // 并且为交易的计划剩余时间 
   // 修改预测范围
   CalcPrognozeLen();

   // 寻找历史中最接近的序列
   // 并且以此为基础预测价格的移动
   FindHistoryPrognoze();

   // 进行必要的交易操作
   Trade();

   TicksEnd=GetTickCount();
   // 记录调试信息
   PrintReport();
  }
//+------------------------------------------------------------------+
void FindHistoryPrognoze()
  {
   Prognoze=0;
   double MaxRating;

   if(Threads>1)
     {
      //--------------------------------------
      // 使用计算指标
      //--------------------------------------
      // 遍历所有记录用指标 
      for(int t=0; t<Threads; t++)
        {
         // 发送计算任务的参数
         SetParam(t,"PrognozeLen",ExistsPrognozeLen);
         // "开始计算" 信号 
         SetParam(t,"Query");
        }

      for(int t=0; t<Threads; t++)
        {
         // 等待结果
         while(!ParamExists(t,"Answer"))
            Sleep(100);
         DelParam(t,"Answer");

         // 取得结果
         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");

         // 选择最佳结果
         if((t==0) || (rating>MaxRating))
           {
            MaxRating = rating;
            Prognoze  = progn;
           }
        }
     }
   else
     {
      //----------------------------
      // 没有使用的指标
      //----------------------------
      // 在EA交易中计算所有内容, 放入数据流中
      FindPrognoze(_Symbol,CurPattern,0,HistoryLen,ExistsPrognozeLen,
                   Prognoze,MaxRating,HistPatternBarStart);
     }
  }
//+------------------------------------------------------------------+
void CalcPrognozeLen()
  {
   ExistsPrognozeLen=PrognozeLen;

   // 如果有开启的仓位, 确定 
   // 自从开启以来经过了多少个柱
   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()
  {

   // 平掉开启的仓位, 假如和预测不符
   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((Prognoze!=0) && (!PositionSelect(_Symbol)))
     {
      CTrade trade;
      if(Prognoze > 0) trade.Buy (Lots);
      if(Prognoze < 0) trade.Sell(Lots);
     }
  }
//+------------------------------------------------------------------+
void PrintReport()
  {
   Print("------------");
   Print("EA: 开始于 ",TicksStart,
         ", 结束于 ",TicksEnd,
         ", 耗时 (ms) ",TicksEnd-TicksStart);
   Print("EA: 共预测 ",ExistsPrognozeLen," 个柱");

   if(Threads>1)
     {
      for(int t=0; t<Threads; t++)
        {
         Print("指标 ",t+1,
               ": 预测 ", Calc[t].prognoze,
               ", 评分 ", Calc[t].rating,
               ", 序列开始于 ",TimeToString(Calc[t].start_time)," 在历史中");
         Print("指标 ",t+1,
               ": 起始时间 ",  Calc[t].ts,
               ", 结束时间 ",   Calc[t].te,
               ", 期间 (ms) ",Calc[t].te-Calc[t].ts);
        }
     }
   else
     {
      Print("没有使用的指标");
      datetime time[];
      CopyTime(_Symbol,_Period,HistPatternBarStart,1,time);
      Print("EA: 序列从 ",TimeToString(time[0])," 开始");
     }

   Print("EA: 预测 ",Prognoze);
   Print("------------");
  }
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
  {
   // 把 "结束" 命令发送到指标
   if(Threads>1)
      for(int t=0; t<Threads; t++)
         SetParam(t,"End");
  }
//+------------------------------------------------------------------+

“EA 交易”使用的计算指标 i-Thread 的代码如下：

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

//--- 输入参数
input string VarPrefix;  // 全局变量的前缀 (类似于幻数)
input int    ThreadNum;  // 核心编号 (这样在不同内核的指标就可以
                        // 把它们的任务和它们 "临近" 内核的任务做区分了)
input string DataSymbol; // MM-EA 工作的货币对
input int    PatternLen; // 分析序列的长度
input int    BarStart;   // 从历史的哪个柱开始搜索类似序列
input int    BarCount;   // 在历史中搜索多少个柱

//--- 指标缓冲区
double Buffer[];
//---
double CurPattern[];

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

   GlobalVarPrefix=VarPrefix;

   // 无限循环 - 这样指标就会一直 "监听", 
   // 为了来自EA的新命令
   while(true)
     {
      // 如果有结束命令, 结束指标的工作
      if(ParamExists(ThreadNum,"End"))
         break;

      // 等待信号以开始计算
      if(!ParamExists(ThreadNum,"Query"))
        {
         Sleep(100);
         continue;
        }
      DelParam(ThreadNum,"Query");

      uint TicksStart=GetTickCount();

      // 取得任务的参数
      int PrognozeLen=GetParam(ThreadNum,"PrognozeLen");

      // 从最近的柱填充序列
      while(CopyClose(DataSymbol,_Period,0,PatternLen,CurPattern)
            <PatternLen) Sleep(1000);

      // 进行计算
      int HistPatternBarStart;
      double Prognoze,Rating;
      FindPrognoze(DataSymbol,CurPattern,BarStart,BarCount,PrognozeLen,
                   Prognoze,Rating,HistPatternBarStart);

      // 发送计算的结果
      SetParam(ThreadNum,"Prognoze",Prognoze);
      SetParam(ThreadNum,"Rating",Rating);
      SetParam(ThreadNum,"PatternStart",HistPatternBarStart);
      SetParam(ThreadNum,"TS",TicksStart);
      SetParam(ThreadNum,"TE",GetTickCount());
      // 信号 "万事俱备"
      SetParam(ThreadNum,"Answer");
     }
  }
//+------------------------------------------------------------------+
int OnCalculate(const int rates_total,
                const int prev_calculated,
                const int begin,
                const double &price[])
  {
   // 事件所需的处理函数
   return(0);
  }
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
  {
   SetParam(ThreadNum,"End");
  }
//+------------------------------------------------------------------+

“EA 交易”和指标使用通用 ThreadCalc.mqh 库。

其代码如下：

//+------------------------------------------------------------------+
//|                                                   ThreadCalc.mqh |
//+------------------------------------------------------------------+
string GlobalVarPrefix;
//+------------------------------------------------------------------+
// 它寻找和指定价格序列最接近的内容.
// 范围为指定的历史范围
// 返回相似度的评估以及 
// 在历史中进一步改变的方向
//+------------------------------------------------------------------+
void FindPrognoze(
                  string DataSymbol,    // 交易品种
                  double  &CurPattern[],// 当前模式
                  int BarStart,         // 起始柱
                  int BarCount,         // 搜索柱数
                  int PrognozeLen,      // 预测长度

                  // 结果
                  double  &Prognoze,        // 预测方向 (-,0,+)
                  double  &Rating,          // 评分
                  int  &HistPatternBarStart // 找到序列的起始柱编号
                  ) 
  {

   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;
   // 在历史中遍历所有价格序列的柱的转换
   for(int bar=BarCount-PatternLen-PrognozeLen; bar>=0; bar--) 
     {
      // 做更新以避免序列中的价格水平差异
      double dr=CurPattern[0]-rates[bar];

      // 计算序列之间水平差异总数
      // 差异为价格和样本值的偏差
      double r=0;
      for(int i=0; i<PatternLen; i++)
         r+=MathPow(MathAbs(rates[bar+i]+dr-CurPattern[i]),2);

      // 找到水平差别最小的序列
      if((r<rmin) || (rmin<0)) 
        {
         rmin=r;
         HistPatternBarStart   = bar;
         int HistPatternBarEnd = bar + PatternLen-1;
         Prognoze=rates[HistPatternBarEnd+PrognozeLen]-rates[HistPatternBarEnd];
        }
     }
   // 把柱数转换为指标系统中的坐标
   HistPatternBarStart=BarStart+BarCount-HistPatternBarStart-PatternLen;

   // 把差异转换为相似度评分
   Rating=-rmin;
  }
//====================================================================
// 为方便操作全局变量的一系列函数.
// 做为包含计算线程数的参数 
// 以及变量的名称, 自动把它们转换为
// 唯一的全局名称.
//====================================================================
//+------------------------------------------------------------------+
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));
  }
//+------------------------------------------------------------------+

能够在工作中使用多个处理器核心的交易系统现已就绪！

测量 EA 的工作速度

普通模式

打开 EURUSD M1 图表并启动我们在之前的章节中创建的“EA 交易”。在设置中，指定图形的长度为 24 小时（1440 个分钟柱）、在历史数据中的搜索深度为 1 年（375,000 个柱）。

图 4.“EA 交易”输入参数

参数“线程”设置为 1。这表示 EA 的所有计算都在一个线程中完成（在单个核心上）。同时，它将不使用计算指标，自行完成所有计算。基本上，遵循普通 EA 的工作原理。

执行日志如下所示：

图 6.“EA 交易”日志（1 线程）

并行模式

现在，我们删除此 EA 及其仓位。重新添加 EA，但这次将参数“线程”设置为 2。

现在，EA 需要在其工作中创建和使用占用两个处理器核心的 2 个计算指标。执行日志如下所示：

图 7.“EA 交易”日志（2 线程）

速度比较

通过分析这两个日志，我们得出 EA 执行的大致时间为：

• 普通模式下 – 52 秒；
• 2 核心模式下 – 27 秒。

因此，通过在 2 核心 CPU 上执行并行化处理，我们能够将 EA 的速度提高到 1.9 倍。我们可以假设，在使用具有多个核心的处理器时，执行速度将获得更多提升，并与核心数量成正比。

工作的正确性控制

除了执行时间，日志还提供了其他信息，以便我们确认所有测量均正确执行。说明 EA:Beginning work … ending work …”（EA：开始工作…结束工作）和 “Indicator …:Beginning work … ending work …”（指标…：开始工作…结束工作）表明指标在 EA 给出该命令后才开始计算。

我们还要确认在并行模式下，交易策略在 EA 启动期间没有违规。从日志上我们可以清楚看到，EA 在并行模式下的启动几乎紧随它在普通模式下的启动。即市场情况在两种情形下相似。日志表明，在图形的历史数据中找到的日期在两种情形下同样是极其相似的。所以一切正常：策略的算法在两种情形下运转同样良好。

下面便是在日志中说明的市场情况的图形。这是在普通模式下运行 EA 时的当前市场情况（长度 – 1440 个分钟柱）：

图 8. 当前市场情况

EA 在历史数据中找到的相似图形，如下所示：

图 9. 相似市场情况

在并行模式下运行 EA 时，“指标 1”找到相同的图形。“指标 2”如下面的日志所示，在另一个半年的历史数据中查找图形，因此找到一个不同的相似图形：

图 10. 相似市场情况

下面是 EA 在并行模式下工作时 MetaTrader 5 中的全局变量的情况：

图 11. 全局变量

经由全局变量的 EA 和指标之间的数据交换成功实施。

总结

在这项研究中我们发现，使用 MetaTrader 5 的标准方法并行化处理较为占用资源的算法是可能的，并且我们针对此问题提出的解决方案可方便地用于现实世界的交易策略中。

运行于多核心系统的该程序真正实现了工作速度的成比例增长。处理器中的核心数目逐年递增，为使用 MetaTrader 的交易人员提供了有效利用这些硬件资源的大好良机。我们可以无忧地创建占用更多资源的交易策略，同时仍然能够实时分析市场。