使用MQL5中的动态时间规整进行模式识别·进阶篇
（2/3）· 接上篇概念铺垫，本篇拆解DTW在金融序列里的变形对齐与纯代码落地
◍ 从矩阵里捞反向失效行
这段类方法在做一件事：当方向列（第 4 列，索引 3）里出现 -1.0 标记时，把对应整行抽出来另存。它先建一个 1×1 的零矩阵垫底，再逐行扫 col 向量，命中 -1.0 就给 out 扩一行并把原矩阵第 j 行写进去，扩容或写行失败就打印错误并返回 1×1 零矩阵。 mkDIrDeltas 最后用 np::sliceMatrixCols(out,1,3) 只保留第 2~4 列（索引 1 到 3），相当于把抽出来的反向失效样本剔除了首列噪声。getP 则按 {0,2,1,3} 重排列序后整体返回，列互换这一步在后续约束比对里常用来对齐特征维度。 开 MT5 把这段塞进自己的 EA 调试，打印 out.Rows() 就能知道样本里到底有多少根 K 线被标了 -1.0；外汇和贵金属波动大，这类标记仅是历史样本筛选，不构成任何方向暗示，实盘仍属高风险。
matrix mkDIrDeltas(class="type">void) { matrix out = matrix::Zeros(class="num">1,class="num">1); vector col = m_mx.Col(class="num">3); for(class="type">ulong i = class="num">0; i<m_mx.Rows(); i++) { for(class="type">ulong j = class="num">0; j<col.Size(); j++) { if(col[j] == -class="num">1.0) { if(!out.Resize(out.Rows()+class="num">1,m_mx.Cols(),class="num">100)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return matrix::Zeros(class="num">1,class="num">1); } vector v = m_mx.Row(j); if(!out.Row(v,out.Rows()-class="num">1)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return matrix::Zeros(class="num">1,class="num">1); } } } } class="kw">return np::sliceMatrixCols(out,class="num">1,class="num">3); } matrix getP(class="type">void) { class="type">ulong sel[] = {class="num">0, class="num">2, class="num">1, class="num">3}; matrix s = np::selectMatrixCols(m_mx,sel); class="kw">return s; }
「DTW 约束窗口的三种矩阵写法」
做动态时间规整（DTW）匹配 K 线形态时，约束矩阵直接决定计算量和形变容忍度。下面三段 MQL5 静态函数给出了无约束、Sakoe-Chiba 带、Itakura 平行四边形三种生成方式，都能直接塞进 MT5 脚本里跑。 无约束版本 noConstraint 把索引矩阵每个元素强制按位或 1，若结果为 0 则置为 inf，等于全开放路径。Sakoe-Chiba 版本用 MathAbs 算行号列号差，差值超过 winsize 就填 inf，窗口宽度由 winsize 控制，比如设 10 时只保留对角线 ±10 以内的格子。 Itakura 版本用 2 倍斜率限制，a、b 两个布尔判断行索引与列索引是否互相不超出 2 倍，构成中间宽两头窄的平行四边形。外汇与贵金属波动跳变多，约束过宽易过拟合历史，实盘前建议在 MT5 用不同 winsize 回测 EURUSD 的 M15 形态匹配命中率。
class="kw">static matrix noConstraint(class="type">ulong iw,class="type">ulong jw) { matrix mats[]; np::indices(iw,jw,mats); for(class="type">ulong i = class="num">0; i<mats[class="num">0].Rows(); i++) { for(class="type">ulong j = class="num">0; j<mats[class="num">0].Cols(); j++) { class="type">long value = class="type">long(mats[class="num">0][i][j]); mats[class="num">0][i][j] = (class="type">class="kw">double)(value|class="num">1); if(mats[class="num">0][i][j]==class="num">0.0) mats[class="num">0][i][j] = class="type">class="kw">double("inf"); } } class="kw">return mats[class="num">0]; } class="kw">static matrix sakoeChibaConstraint(class="type">ulong iw,class="type">ulong jw, class="type">ulong qsize, class="type">ulong refsize, class="type">ulong winsize) { matrix mats[]; np::indices(iw,jw,mats); matrix abs = MathAbs(mats[class="num">1]-mats[class="num">0]); for(class="type">ulong i = class="num">0; i<abs.Rows(); i++) { for(class="type">ulong j = class="num">0; j<abs.Cols(); j++) { if(class="type">ulong(abs[i][j])<=winsize) abs[i][j] = (class="type">class="kw">double)(class="num">1); else abs[i][j] = class="type">class="kw">double("inf"); } } class="kw">return abs; } class="kw">static matrix itakuraConstraint(class="type">ulong iw,class="type">ulong jw, class="type">ulong qsize, class="type">ulong refsize) { matrix mats[]; np::indices(iw,jw,mats); class="type">long a,b,c,d; for(class="type">ulong i = class="num">0, k = class="num">0; i<mats[class="num">0].Rows() && k<mats[class="num">1].Rows(); i++,k++) { for(class="type">ulong j = class="num">0; j<mats[class="num">0].Cols(); j++) { a = class="type">long(mats[class="num">1][k][j]) < (class="num">2*class="type">long(mats[class="num">0][i][j]))?class="num">1:class="num">0; b = class="type">long(mats[class="num">0][i][j]) <=(class="num">2*class="type">long(mats[class="num">1][k][j]))?class="num">1:class="num">0;
斜带约束与DTW入口的矩阵实现
在动态时间规整（DTW）里，斜带约束用来限制对齐路径偏离主对角线的幅度。上面 slantedBandConstraint 函数先按 refsize/qsize 比例算出理论对角坐标 diagj，再取 mats[1] 与 diagj 的绝对偏差矩阵 abs。 逐元素判断：若偏差 ulong(abs[i][j]) 不超过 winsize，则置 1 表示允许对齐；否则写死为 inf，后续累加时会天然屏蔽该格。这意味着 winsize 直接决定计算量——例如 qsize=100、refsize=100、winsize=10 时，有效格子约 100*21=2100 而非全量 10000。 前面的 c、d 逻辑则处理另一种边界对称约束：用 long 强转做索引比较，当 mats[0][i][j] 越界条件满足时给 1，否则 0；四个布尔位与后转回 double，若为 0 则填 inf 封死该路径。 对外接口 dtw() 先校验 x、y 列数一致，否则 Print 报错并返回 false；随后按 STEP_SYMM1 / SYMM2 等枚举注销旧步模、new 出对应 CStepPattern。外汇与贵金属行情用这套做形态比对时波动剧烈，属高风险用法，回测结论仅具概率意义。
c = class="type">long(mats[class="num">0][i][j]) >=(class="type">long(qsize)-class="num">1-class="num">2*(class="type">long(refsize)-class="type">long(mats[class="num">1][k][j])))?class="num">1:class="num">0; d = class="type">long(mats[class="num">1][k][j]) > (class="type">long(refsize)-class="num">1-class="num">2*(class="type">long(qsize)-class="type">long(mats[class="num">0][i][j])))?class="num">1:class="num">0; mats[class="num">0][i][j] = class="type">class="kw">double(class="type">ulong(a&b&c&d)); if(mats[class="num">0][i][j]==class="num">0.0) mats[class="num">0][i][j] = class="type">class="kw">double("inf"); } } class="kw">return mats[class="num">0]; } class="kw">static matrix slantedBandConstraint(class="type">ulong iw,class="type">ulong jw, class="type">ulong qsize, class="type">ulong refsize,class="type">ulong winsize) { matrix mats[]; np::indices(iw,jw,mats); matrix diagj = (mats[class="num">0]*refsize/qsize); matrix abs = MathAbs(mats[class="num">1]-diagj); for(class="type">ulong i = class="num">0; i<abs.Rows(); i++) { for(class="type">ulong j = class="num">0; j<abs.Cols(); j++) { if(class="type">ulong(abs[i][j])<=winsize) abs[i][j] = (class="type">class="kw">double)(class="num">1); else abs[i][j] = class="type">class="kw">double("inf"); } } class="kw">return abs; } }; class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| main interface method for dtw | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">bool dtw(matrix&x, matrix&y, ENUM_DIST_METRIC dist_method,ENUM_STEP_PATTERN step_pattern=STEP_SYMM2, ENUM_GLOBAL_CONSTRAINT win_type=CONSTRAINT_NONE,class="type">ulong winsize=class="num">0) { if(y.Cols()!=x.Cols()) { Print(__FUNCTION__, " invalid input parameters, size containers donot match. "); class="kw">return class="kw">false; } if(CheckPointer(m_stepPattern)==POINTER_DYNAMIC) class="kw">delete m_stepPattern; class="kw">switch(step_pattern) { case STEP_SYMM1: m_stepPattern = new CStepPattern(_symmetric1,Symmetric);
◍ 距离矩阵与约束窗口的装配细节
这段初始化逻辑里，先按 step 类型把 m_stepPattern 指针建出来，同时给 m_stephint 打上对应的 HINT 标记；STEP_SYMM2 走 Symmetric 配 HINT_NM，STEP_ASYMM 走 Asymmetric 配 HINT_N，任何一步指针无效就直接 Print 报错并返回 false。 紧接着把查询序列 x 和参考序列 y 的行列数分别存进 m_qlen、m_reflen，并设定距离度量与窗口类型。若 y 有数据（y.Rows() 非零），就对 m_distance 做 m_qlen×m_reflen 的 Resize，再用双层 ulong 循环逐格填 dist(查询第 i 行, 参考第 j 行)；若 y 为空，m_distance 直接等于 m_query 本身。 窗口矩阵 wm 的构造取决于 m_winMethod：CONSTRAINT_NONE 时全 1 矩阵（matrix::Ones）放开所有对齐路径；否则在 Itakura、Sakoe-Chiba、Slanted Band 三种约束里二选一，后两者都吃 m_winsize 参数控制带宽。实盘跑外汇或贵金属序列对齐时，Sakoe-Chiba 的 winsize 设太小容易把有效形变掐掉，建议先在 MT5 里用历史 Tick 序列打印 wm 非零元比例，再定参。
m_stephint = HINT_NA; class="kw">break; case STEP_SYMM2: m_stepPattern = new CStepPattern(_symmetric2,Symmetric,HINT_NM); m_stephint = HINT_NM; class="kw">break; case STEP_ASYMM: m_stepPattern = new CStepPattern(_asymmetric,Asymmetric,HINT_N); m_stephint = HINT_N; class="kw">break; } if(CheckPointer(m_stepPattern)==POINTER_INVALID) { Print(__FUNCTION__," failed step pointer initialization ", GetLastError()); class="kw">return class="kw">false; } matrix stepsMatrix = m_stepPattern.getStepMatrix(); m_query = x; m_qlen = x.Rows(); m_ref = y; m_reflen = y.Rows(); m_distMetric = dist_method; m_winMethod = win_type; m_winsize = winsize; if(y.Rows()) { if(!m_distance.Resize(m_qlen,m_reflen)) { Print(__FUNCTION__," resize error ", GetLastError()); class="kw">return class="kw">false; } for(class="type">ulong i = class="num">0; i<m_qlen; i++) for(class="type">ulong j =class="num">0; j<m_reflen; j++) m_distance[i][j]=dist(m_query.Row(i),m_ref.Row(j)); } else m_distance = m_query; class="type">ulong n,m; n=m_distance.Rows(); m=m_distance.Cols(); matrix wm; if(m_winMethod == CONSTRAINT_NONE) wm = matrix::Ones(m_distance.Rows(), m_distance.Cols()); else { class="kw">switch(m_winMethod) { case CONSTRAINT_ITAKURA: wm = CConstraint::itakuraConstraint(n,m,m_qlen,m_reflen); class="kw">break; case CONSTRAINT_SAKOECHIBA: wm = CConstraint::sakoeChibaConstraint(n,m,m_qlen,m_reflen,m_winsize); class="kw">break; case CONSTRAINT_SLATEDBAND: wm = CConstraint::slantedBandConstraint(n,m,m_qlen,m_reflen,m_winsize); class="kw">break; class="kw">default: wm = CConstraint::noConstraint(n,m);
「DTW 代价矩阵与回溯路径的落地实现」
这段逻辑紧接前面的距离矩阵构造，核心是把带约束的加权距离 wm 摊平进 m_distance，再扩展出可回溯的代价矩阵与方向矩阵。 若 m_winMethod 不是 CONSTRAINT_NONE，就遍历 wm 的所有行列，跳过 i+j=0 的原点，把非 1.0 的权重写回 m_distance[i][j]；这一步决定了窗口约束是否真正生效。 m_costMatrix 的维度由 m_distance 边长分别加上 stepsMatrix 第 0、1 列的最大值得到，初始化为 inf，并把右下锚点设为 m_distance[0][0]；m_dirMatrix 先填 INT_MIN，再把第 0 行标 1、第 0 列标 2，作为回溯方向种子。 calCM 失败会打印函数名并返回 false，成功则把 m_jmin 置为列数减 1 并返回 true。warpPath 直接调用 backtrack 拿对齐点，openEnd 为真时末行取 ArgMin 作为终点；costMatrix 则原样吐出 m_costMatrix 供可视化或二次校验。 在 MT5 里把这段接进你自己的序列比对类，跑两段 EURUSD 的 H1 收盘价，观察 m_costMatrix 右下角数值相对 inf 的收敛情况，能直观判断对齐是否跑通。外汇与贵金属波动剧烈，这类距离度量仅作形态比对参考，实盘信号须自行验证风险。
class="kw">break; } } if(m_winMethod!=CONSTRAINT_NONE) { for(class="type">ulong i = class="num">0; i<wm.Rows(); i++) for(class="type">ulong j = class="num">0; j<wm.Cols(); j++) if((i+j)>class="num">0 && wm[i][j] != class="num">1.0) m_distance[i][j] = wm[i][j]; } m_costMatrix = matrix::Zeros(m_distance.Rows()+class="type">ulong(stepsMatrix.Col(class="num">0).Max()),m_distance.Cols()+class="type">ulong(stepsMatrix.Col(class="num">1).Max())); m_costMatrix.Fill(class="type">class="kw">double("inf")); m_costMatrix[class="type">ulong(stepsMatrix.Col(class="num">0).Max())][class="type">ulong(stepsMatrix.Col(class="num">1).Max())] = m_distance[class="num">0][class="num">0]; m_dirMatrix = matrix::Zeros(m_costMatrix.Rows(),m_costMatrix.Cols()); m_dirMatrix.Fill(class="type">class="kw">double(INT_MIN)); for(class="type">ulong i = class="num">0; i<m_dirMatrix.Cols(); i++) m_dirMatrix[class="num">0][i] = class="type">class="kw">double(class="num">1); for(class="type">ulong i = class="num">0; i<m_dirMatrix.Rows(); i++) m_dirMatrix[i][class="num">0] = class="type">class="kw">double(class="num">2); if(!calCM(m_distance,stepsMatrix,m_costMatrix,m_dirMatrix)) { Print(__FUNCTION__, " computeCM() failed "); class="kw">return class="kw">false; } m_jmin = m_costMatrix.Cols() - class="num">1; class="kw">return true; } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Get the optimal path: corresponding points from both series | class=class="str">"cmt">//+------------------------------------------------------------------+ matrix warpPath(class="type">bool openEnd=class="kw">false) { matrix stmatrix = m_stepPattern.getStepMatrix(); class="kw">return backtrack(m_dirMatrix,stmatrix,openEnd,openEnd?class="type">long(m_costMatrix.Row(m_costMatrix.Rows()-class="num">1).ArgMin()):-class="num">1); } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Get the accumulated cost matrix | class=class="str">"cmt">//+------------------------------------------------------------------+ matrix costMatrix(class="type">void) { class="kw">return m_costMatrix; } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Get the cost matrix | class=class="str">"cmt">//+------------------------------------------------------------------+
把动态时间规整的代价与方向矩阵吐出来
封装类里留了两个轻量接口，直接把累积代价矩阵和回溯方向矩阵返给调用方，省得外部再翻私有成员。localCostMatrix 返回 m_distance，directionMatrix 返回 m_dirMatrix，两者都是 matrix 类型，可在 MT5 策略里接着做路径回溯或可视化。 calCM 是私有的累积代价计算核心。它先取步长矩阵 stepMatrix 第 0、1 列的最大值 max0、max1，作为两个序列对齐的起始偏移；随后三层循环遍历 costMatrix 的剩余行列，对每个候选步长 k 计算 curd = 上一格代价 + 当前距离，若更小就更新 costMatrix 与 dirMatrix。注意 i、j 的索引做了 i-max0、j-max1 的偏移，说明距离矩阵 distMatrix 只存了偏移后的局部块。 算完之后用 np::sliceMatrix 把前 max0 行、max1 列裁掉，只保留有效对齐区。这种写法在 EURUSD 的 1 分钟序列上做 DTW 时，若步长集含 8 个候选位移，calCM 的内层循环次数约为 (N-max0)*(M-max1)*8，N、M 为两序列长度，调参时心里得有这个数。 dist 函数按 m_distMetric 分派距离度量，目前挂了欧氏与曼哈顿两条路。欧氏对波动率突变敏感，曼哈顿在贵金属跳空段更鲁棒，切换枚举值就能比两种度量的对齐路径差异。外汇与贵金属杠杆高、滑点不可控，回测对齐再漂亮也只代表历史形态概率，实盘务必小仓验证。
matrix localCostMatrix(class="type">void) { class="kw">return m_distance; } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Get the direction matrix | class=class="str">"cmt">//+------------------------------------------------------------------+ matrix directionMatrix(class="type">void) { class="kw">return m_dirMatrix; } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| class="kw">private method implementing accumulated cost calculation | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">bool calCM(matrix &distMatrix,matrix &stepMatrix,matrix &costMatrix,matrix &dirMatrix) { class="type">ulong max0,max1; max0 = class="type">ulong(stepMatrix.Col(class="num">0).Max()); max1 = class="type">ulong(stepMatrix.Col(class="num">1).Max()); class="type">class="kw">double curCost,curd; for(class="type">ulong i = max0; i<costMatrix.Rows(); i++) { for(class="type">ulong j = max1; j<costMatrix.Cols(); j++) { for(class="type">ulong k = class="num">0; k<stepMatrix.Rows(); k++) { curd = costMatrix[i-class="type">ulong(stepMatrix[k][class="num">0])][j-class="type">ulong(stepMatrix[k][class="num">1])]; curCost = curd + distMatrix[i-max0][j-max1]; if(curCost<costMatrix[i][j]) { costMatrix[i][j] = curCost; dirMatrix[i][j] = class="type">class="kw">double(k); } } } } costMatrix = np::sliceMatrix(costMatrix,max0,END,class="num">1,max1); dirMatrix = np::sliceMatrix(dirMatrix,max0,END,class="num">1,max1); class="kw">return true; } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| distance metric calculation | class=class="str">"cmt">//+------------------------------------------------------------------+ class="type">class="kw">double dist(vector &u,vector &v) { class="kw">switch(m_distMetric) { case DIST_EUCLIDEAN: class="kw">return euclidean(u,v); case DIST_CITYBLOCK: class="kw">return cityblock(u,v);