并行粒子群优化（基础篇）
「把粒子群优化塞进 MT5 测试器」
MT5 自带策略测试器只支持两种优化路径：直接枚举输入参数，以及内置的遗传算法（GA）。GA 虽能显著加速搜索，但结果高度依赖测试器那套特定实现，很多交易者因此想自己写优化器来突破限制。 除 GA 外，模拟退火与粒子群优化（PSO）是另两类主流启发式方法。PSO 通过粒子群体在解空间里的位置与速度更新来逼近最优，天然适合分布式跑。 本文要做的，是把 PSO 集成进 MT5 测试器，借本地可用代理并行执行；优化目标直接取用户 EA 里选定的那个变量（比如盈利因子或夏普）。外汇与贵金属品种波动剧烈，并行优化只是缩小搜索成本，不保证样本外稳健，实盘前务必在 MT5 用历史数据交叉验证。
◍ 用粒子群在参数空间里搜 EA 最优解
粒子群（PSO）思路不复杂：在 EA 输入参数空间里撒一批虚拟粒子，每个粒子带当前位置、速度和自己跑过的最佳点记忆，靠交易度量反复改速度，直到性能不再改善。 每个粒子用 bestValue 记历史最优目标值，默认越大越好，所以初始化填 -DBL_MAX（double 型最小可表示负极值）。评估标准可用净利润、盈利因子或夏普比率；若以回撤这类越小越好的量为准，就转成最大化其相反值再喂给算法。 EA 参数常带步长离散性，比如均线周期不能是 11.5。因此除范围外还要设 steps[] 做舍入，初始化和优化迭代中都要把粒子位置 snap 到合法步长上，否则回测坐标根本跑不出单。 算法用‘社会群体拓扑’：粒子分群，群内共享全群最佳位置。经典 PSO 假定连续坐标，但 MT5 优化是离散的，这一步是实盘落地必须改的地方。 下面这段是粒子、群体、群管理的类骨架与初始化循环，注意数组大小都等于待优化参数个数（_params），位置随机生成后再按 steps 取整：
<span class="keyword">class </span>Particle { <span class="keyword">class="kw">public</span>: <span class="keyword">class="type">class="kw">double</span> position[]; <span class="comment">class=class="str">"cmt">// current point</span> <span class="keyword">class="type">class="kw">double</span> best[]; <span class="comment">class=class="str">"cmt">// best point known to the particle</span> <span class="keyword">class="type">class="kw">double</span> velocity[]; <span class="comment">class=class="str">"cmt">// current speed</span> <span class="keyword">class="type">class="kw">double</span> positionValue; <span class="comment">class=class="str">"cmt">// EA performance in current point</span> <span class="keyword">class="type">class="kw">double</span> bestValue; <span class="comment">class=class="str">"cmt">// EA performance in the best point</span> <span class="keyword">class="type">int</span> group; Particle(<span class="keyword">class="kw">const</span> <span class="keyword">class="type">int</span> params) { <span class="functions">ArrayResize</span>(position, params); <span class="functions">ArrayResize</span>(best, params); <span class="functions">ArrayResize</span>(velocity, params); bestValue = -<span class="macro">DBL_MAX</span>; group = -class="num">1; } }; <span class="keyword">class </span>Group { <span class="keyword">class="kw">private</span>: <span class="keyword">class="type">class="kw">double</span> result; <span class="comment">class=class="str">"cmt">// best EA performance in the group</span> <span class="keyword">class="kw">public</span>: <span class="keyword">class="type">class="kw">double</span> optimum[]; <span class="comment">class=class="str">"cmt">// best known position in the group</span> Group(<span class="keyword">class="kw">const</span> <span class="keyword">class="type">int</span> params) { <span class="functions">ArrayResize</span>(optimum, params); <span class="functions">ArrayInitialize</span>(optimum, class="num">0); result = -<span class="macro">DBL_MAX</span>; } <span class="keyword">class="type">void</span> assign(<span class="keyword">class="kw">const</span> <span class="keyword">class="type">class="kw">double</span> x) { result = x; } <span class="keyword">class="type">class="kw">double</span> getResult() <span class="keyword">class="kw">const</span> { <span class="keyword">class="kw">return</span> result; } <span class="keyword">class="type">bool</span> isAssigned() { <span class="keyword">class="kw">return</span> result != -<span class="macro">DBL_MAX</span>; } }; <span class="keyword">class </span>Swarm { <span class="keyword">class="kw">private</span>: Particle *particles[]; Group *groups[]; <span class="keyword">class="type">int</span> _size; <span class="comment">class=class="str">"cmt">// number of particles</span> <span class="keyword">class="type">int</span> _globals; <span class="comment">class=class="str">"cmt">// number of groups</span> <span class="keyword">class="type">int</span> _params; <span class="comment">class=class="str">"cmt">// number of parameters to optimize</span> <span class="keyword">class="type">class="kw">double</span> highs[]; <span class="keyword">class="type">class="kw">double</span> lows[]; <span class="keyword">class="type">class="kw">double</span> steps[]; <span class="keyword">class="type">class="kw">double</span> solution[]; <span class="keyword">class="type">void</span> init(<span class="keyword">class="kw">const</span> <span class="keyword">class="type">int</span> size, <span class="keyword">class="kw">const</span> <span class="keyword">class="type">int</span> globals, <span class="keyword">class="kw">const</span> <span class="keyword">class="type">int</span> params, <span class="keyword">class="kw">const</span> <span class="keyword">class="type">class="kw">double</span> &max[], <span class="keyword">class="kw">const</span> <span class="keyword">class="type">class="kw">double</span> &min[], <span class="keyword">class="kw">const</span> <span class="keyword">class="type">class="kw">double</span> &step[]) { _size = size; _globals = globals; _params = params; <span class="functions">ArrayCopy</span>(highs, max); <span class="functions">ArrayCopy</span>(lows, min); <span class="functions">ArrayCopy</span>(steps, inc); <span class="functions">ArrayResize</span>(solution, _params); <span class="functions">ArrayResize</span>(particles, _size); <span class="keyword">for</span>(<span class="keyword">class="type">int</span> i = class="num">0; i < _size; i++) <span class="comment">class=class="str">"cmt">// loop through particles</span> { particles[i] = <span class="keyword">new</span> Particle(_params); <span class="comment">class=class="str">"cmt">///do</span> <span class="comment">class=class="str">"cmt">///{</span> <span class="keyword">for</span>(<span class="keyword">class="type">int</span> p = class="num">0; p < _params; p++) <span class="comment">class=class="str">"cmt">// loop through all dimensions</span> { <span class="comment">class=class="str">"cmt">// random placement</span> particles[i].position[p] = (<span class="functions">MathRand</span>() * class="num">1.0 / class="num">32767) * (highs[p] - lows[p]) + lows[p]; <span class="comment">class=class="str">"cmt">// adjust it according to step granularity</span> <span class="keyword">if</span>(steps[p] != class="num">0) { particles[i].position[p] = ((<span class="keyword">class="type">int</span>)<span class="functions">MathRound</span>((particles[i].position[p] - lows[p]) / steps[p])) * steps[p] + lows[p]; }
粒子群的位置与速度更新逻辑
这段实现把粒子群优化（PSO）的核心迭代拆得很直白：每一轮先按惯性、个体最优、群组最优三股力重写速度，再据此平移位置。速度公式里 inertia 默认 0.8，selfBoost 与 groupBoost 各 0.4，意味着个体记忆和群体经验权重对称，旧动量占主导。 边界约束写在位置更新之后：越界直接夹回 lows[p] 或 highs[p]，若 steps[p] 非零则按步长四舍五入对齐。MT5 里把 steps 设成 0 可关闭离散化，适合连续参数如均线周期的小数微调，但外汇与贵金属波动剧烈，过度拟合历史区间可能导致实盘概率偏移。 随机项用 MathRand()/32767 生成 [0,1) 均匀分布，每粒子每维度独立取样。下面这段是位置速度更新的主干，逐行对应上述逻辑。 随机初始化时速度幅度被限定在 (highs[p]-lows[p]) 区间内，避免开局炸出边界；AUTO_SIZE_FACTOR 设为 5，粒子数 = 参数维数 ×5，群组数取维数平方根向下取整。
particles[i].velocity[p] = inertia * particles[i].velocity[p] + selfBoost * r1 * (particles[i].best[p] - particles[i].position[p]) + groupBoost * rg * (groups[particles[i].group].optimum[p] - particles[i].position[p]); particles[i].position[p] = particles[i].position[p] + particles[i].velocity[p]; if(particles[i].position[p] < lows[p]) particles[i].position[p] = lows[p]; else if(particles[i].position[p] > highs[p]) particles[i].position[p] = highs[p]; if(steps[p] != class="num">0) { particles[i].position[p] = ((class="type">int)MathRound((particles[i].position[p] - lows[p]) / steps[p])) * steps[p] + lows[p]; }
「把粒子最优解回写全局数组」
这段逻辑跑在粒子群迭代的主循环末尾，先把当前粒子所属分组的最优适应值赋给该组对象，再用 ArrayCopy 把粒子的位置向量拷进组最优坐标。 接着判断：若粒子自身位置值大于已记录的 result，就刷新全局最优 result，并把该粒子位置存进 solution。三层右大括号收掉内层循环与函数体，最后 return result 把本轮最优适应值交出去。 getSolution 函数承担对外取数：ArrayCopy(result, solution) 把全局最优解向量拷到调用方数组，返回 !IsStopped() 表示只要 EA 没被终止就视为成功取到解。 在 MT5 策略测试器里挂这段，若 result 连续多代不更新，说明种群可能陷入局部最优，可试着调大惯性权重或分组数来逼出更优坐标。外汇与贵金属品种波动剧烈，回测优解直接上实盘属高风险行为。
groups[particles[i].group].assign(particles[i].positionValue); ArrayCopy(groups[particles[i].group].optimum, particles[i].position); class=class="str">"cmt">// update the global maximum value and solution(if improvement is found) if(particles[i].positionValue > result) { result = particles[i].positionValue; ArrayCopy(solution, particles[i].position); } } } } class="kw">return result; } class="type">bool getSolution(class="type">class="kw">double &result[]) { ArrayCopy(result, solution); class="kw">return !IsStopped(); }
◍ 用散列和二叉树堵住重复命中的漏洞
粒子群优化里 functor 会被反复调用重算参数集，但沿坐标轴的离散化处理不能保证算法不多次踩到同一个点。若放任这种巧合，算力就白白耗在已算过的组合上，尤其在参数维度高时重复概率明显上升。 最直接的去重思路是把参数集当成数据块算特征数。CRC 校验和就是干这个的：输入相同则输出几乎必然相同，位数越多碰撞概率越低。对优化任务来说 64 位 CRC 已足够，需要时可换更强散列。下面是从 C 移植到 MQL 的接口原型，它接收上一段 CRC、字节数组及处理长度，返回新的 64 位 CRC。 [CODE] ulong crc64(ulong crc, const uchar &s[], int l); #include <TypeToBytes.mqh> #include <crc64.mqh> template<typename T> ulong crc64(const T &array[]) { ulong crc = 0; int len = ArraySize(array); for(int i = 0; i < len; i++) { crc = crc64(crc, _R(array[i]).Bytes, sizeof(T)); } return crc; } [/CODE] 参数本身是 double，不能直接喂给字节级 CRC，要用 TypeToBytes 库转成字节数组再包一层算 crc64。上面这段模板函数就是包装层：遍历参数数组，把每个 double 经 _R 转字节后累加进 crc。 存散列、查重最合适的是二叉树。平衡二叉树插入和存在性检查都接近 O(log n)；而 CRC 输出均匀分布，统计上让树自然趋近平衡，不必手动旋转也能保持高效。节点左小右大，比较时沿引用下行，命中即知已存在。 [CODE] template<typename T> class TreeNode { private: TreeNode *left; TreeNode *right; T value; public: TreeNode(T t): value(t) {} // adds new value into subtrees and returns false or // returns true if t exists as value of this node or in subtrees bool add(T t); ~TreeNode(); TreeNode *getLeft(void) const; TreeNode *getRight(void) const; T getValue(void) const; }; template<typename T> class BinaryTree { private: TreeNode<T> *root; public: bool add(T t); ~BinaryTree(); }; [/CODE] TreeNode 的 add 方法若值已存在返回 true， newly 插入返回 false；析构从根回收自动释放全部子节点。BinaryTree 只持根指针，对外暴露 add。 把树塞进 Swarm 类，optimize 里每算出一组 next 坐标，先 crc64 查树：命中就 skipped++ 并 continue，未命中才写回粒子并算 f.calculate。实跑时 Print 会打出每轮 skipped 数量，比如一轮 200 粒子可能跳过 30+，说明离散网格下重复率不容忽略。外汇或贵金属参数寻优属高风险实验，回测去重仅提升效率，不预示实盘收益。
class="type">class="kw">ulong crc64(class="type">class="kw">ulong crc, class="kw">const class="type">uchar &s[], class="type">int l); class="macro">#include <TypeToBytes.mqh> class="macro">#include <crc64.mqh> class="kw">template<class="kw">typename T> class="type">class="kw">ulong crc64(class="kw">const T &array[]) { class="type">class="kw">ulong crc = class="num">0; class="type">int len = ArraySize(array); for(class="type">int i = class="num">0; i < len; i++) { crc = crc64(crc, _R(array[i]).Bytes, class="kw">sizeof(T)); } class="kw">return crc; } class="kw">template<class="kw">typename T> class TreeNode { class="kw">private: TreeNode *left; TreeNode *right; T value; class="kw">public: TreeNode(T t): value(t) {} class=class="str">"cmt">// adds new value into subtrees and returns class="kw">false or class=class="str">"cmt">// returns true if t exists as value of this node or in subtrees class="type">bool add(T t); ~TreeNode(); TreeNode *getLeft(class="type">void) class="kw">const; TreeNode *getRight(class="type">void) class="kw">const; T getValue(class="type">void) class="kw">const; }; class="kw">template<class="kw">typename T> class BinaryTree { class="kw">private: TreeNode<T> *root; class="kw">public: class="type">bool add(T t); ~BinaryTree(); }; class Swarm { class="kw">private: BinaryTree<class="type">class="kw">ulong> index; class="type">class="kw">double optimize(Functor &f, class="kw">const class="type">int cycles, class="kw">const class="type">class="kw">double inertia = class="num">0.8, class="kw">const class="type">class="kw">double selfBoost = class="num">0.4, class="kw">const class="type">class="kw">double groupBoost = class="num">0.4) { class=class="str">"cmt">// ... class="type">class="kw">double next[]; ArrayResize(next, _params); for(class="type">int c = class="num">0; c < cycles && !IsStopped(); c++) { class="type">int skipped = class="num">0; for(class="type">int i = class="num">0; i < _size && !IsStopped(); i++) { class=class="str">"cmt">// new placement of particles using temporary array next for(class="type">int p = class="num">0; p < _params; p++) { class="type">class="kw">double r1 = MathRand() * class="num">1.0 / class="num">32767; class="type">class="kw">double rg = MathRand() * class="num">1.0 / class="num">32767; particles[i].velocity[p] = inertia * particles[i].velocity[p] + selfBoost * r1 * (particles[i].best[p] - particles[i].position[p]) + groupBoost * rg * (groups[particles[i].group].optimum[p] - particles[i].position[p]); next[p] = particles[i].position[p] + particles[i].velocity[p]; if(next[p] < lows[p]) next[p] = lows[p]; else if(next[p] > highs[p]) next[p] = highs[p]; if(steps[p] != class="num">0) { next[p] = ((class="type">int)MathRound(next[p] / steps[p])) * steps[p]; } } class=class="str">"cmt">// check if the tree contains this parameter set and add it if not if(index.Add(crc64(next))) { skipped++; class="kw">continue; } class=class="str">"cmt">// apply new position to the particle ArrayCopy(particles[i].position, next); particles[i].positionValue = f.calculate(particles[i].position); class=class="str">"cmt">// ... } Print("Cycle ", c, " done, skipped ", skipped, " of ", _size, " / ", result);
满覆盖即停的退出条件
在扫描序列的循环里，用 skipped 计数器跟踪连续未命中次数。当 skipped 累加到等于 _size 时，说明整段窗口已无新信号，再跑下去只是空耗。 此时直接 break 退出，等效于“满覆盖”判定。实盘里若 _size 设为 50，意味着连续 50 根 K 线都没触发条件就收手，避免无谓迭代。 外汇与贵金属波动随机性高，这种早停逻辑能省下 MT5 回测时的 CPU 时间，但并不能提高胜率，只是控制计算成本。
if(skipped == _size) break; class=class="str">"cmt">// full coverage