基于转移熵的时间序列因果分析·进阶篇
（2/3）· 同步涨跌不等于因果，进阶篇教你用转移熵量化信息流向并落地MQL5代码
滑窗与滞后矩阵的初始化落点
这段 CDataWindows 类的收尾逻辑，负责把外部传入的 matrix 数据按滞后阶数（lag）和滑动窗口（window_size / window_stride）切成可训练样本。Initialize 明确要求 data.Cols() 至少为 2，否则直接打印报错并返回 false——单列序列在这套封装里跑不起来。 applywindows 里用 for 循环从 (m_stride_size+m_win_size) 开始按步长抽取行切片，ArrayResize 的第三参设为 100，意味着每次扩容预留 100 个元素的缓冲，避免高频 resize 拖慢 MT5 回测。若 resize 返回小于 0，立刻 Print(__FUNCTION__," error ", GetLastError()) 并 return false，错误定位很直接。 无窗口模式走 else 分支：m_has_windows 置 false，清空 m_dwins 后仅 ArrayResize 到 1，相当于把整段数据当作单个样本。外汇与贵金属行情的高波动特性下，window_size 设 0 会丢失局部时序结构，样本构建前建议先在 MT5 里打印 m_dwins.Size() 确认切分数量符合预期。
class="kw">return matrix::Zeros(class="num">1,class="num">1); } } } } class="kw">return out; } class="type">bool applywindows(class="type">void) { if(m_dwins.Size()) ArrayFree(m_dwins); for(class="type">ulong i = (m_stride_size+m_win_size); i<m_data.Rows(); i+=class="type">ulong(MathMax(m_stride_size,class="num">1))) { if(ArrayResize(m_dwins,class="type">int(m_dwins.Size()+class="num">1),class="num">100)<class="num">0) { Print(__FUNCTION__," error ", GetLastError()); class="kw">return false; } m_dwins[m_dwins.Size()-class="num">1] = np::sliceMatrixRows(m_data,i-m_win_size,(i-m_win_size)+m_win_size); } class="kw">return true; } class="kw">public: CDataWindows(class="type">void) { } ~CDataWindows(class="type">void) { } class="type">bool Initialize(matrix &data, class="type">ulong lag, class="type">bool max_lag_only=true, class="type">ulong window_size=class="num">0, class="type">ulong window_stride =class="num">0) { if(data.Cols()<class="num">2) { Print(__FUNCTION__, " matrix should contain at least class="num">2 columns "); class="kw">return false; } m_data = data; m_max_lag_only = max_lag_only; if(lag) { m_lag = lag; m_data = applylags(); } if(window_size) { m_win_size = window_size; m_stride_size = window_stride; m_has_windows = true; if(!applywindows()) class="kw">return false; } else { m_has_windows = false; if(m_dwins.Size()) ArrayFree(m_dwins); if(ArrayResize(m_dwins,class="num">1)<class="num">0) { Print(__FUNCTION__," error ", GetLastError());
「窗口容器与线性传递熵的计算落点」
这段实现把滑动窗口存进 m_dwins 矩阵数组，并用 getWindowAt(ind) 做越界保护：索引超出 Size() 时打印函数名与「Index out of bounds」，返回 1×1 零矩阵而不是崩掉。numWindows() 直接返回数组长度，hasWindows() 暴露 m_has_windows 标志，调用方靠这两个方法判断是否有可用样本。 Calculate_Linear_TE 是核心：先按窗口数 c 建 4 个 c×2 矩阵（TE、sTE、pvals、zscores），分别对应双向传递熵、shuffle 均值、p 值与 z 分数。循环里对第 i 个窗口取 df，算 linear_transfer(df,0,1) 与 (df,1,0) 填进 m_transfer_entropies，再写回 TE 的第 i 行；若 Row() 失败立即返回 false 并报 GetLastError()。 当 n_shuffles 非 0 时才跑 significance() 做置换检验，把 mean / pvalue / zscore 三行分别写进 sTE、pvals、zscores，任一行写入失败同样返回 false。最后把 TE 的 0、1 列拆成 m_results.TE_XY 与 TE_YX，p 值和 z 分数也按列提取——外汇与贵金属价格序列的高风险在于：样本窗口若含跳空，传递熵方向可能假性反转，MT5 里建议先用真实 tick 数据跑 n_shuffles≥500 看 p 值稳定性。
class="kw">return false; } m_dwins[class="num">0]=m_data; } class="kw">return true; } matrix getWindowAt(class="type">ulong ind) { if(ind < class="type">ulong(m_dwins.Size())) class="kw">return m_dwins[ind]; else { Print(__FUNCTION__, " Index out of bounds "); class="kw">return matrix::Zeros(class="num">1,class="num">1); } } class="type">ulong numWindows(class="type">void) { class="kw">return class="type">ulong(m_dwins.Size()); } class="type">bool hasWindows(class="type">void) { class="kw">return m_has_windows; } }; class="type">bool Calculate_Linear_TE(class="type">ulong n_shuffles=class="num">0) { class="type">ulong c = m_wins.numWindows(); matrix TE(c,class="num">2); matrix sTE(c,class="num">2); matrix pvals(c,class="num">2); matrix zscores(c,class="num">2); for(class="type">ulong i=class="num">0; i<m_wins.numWindows(); i++) { matrix df = m_wins.getWindowAt(i); m_transfer_entropies[class="num">0] = linear_transfer(df,class="num">0,class="num">1); m_transfer_entropies[class="num">1] = linear_transfer(df,class="num">1,class="num">0); if(!TE.Row(m_transfer_entropies,i)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return false; } SigResult rlts; if(n_shuffles) { significance(df,m_transfer_entropies,m_endog,m_exog,m_tlag,m_maxlagonly,n_shuffles,rlts); if(!sTE.Row(rlts.mean,i) || !pvals.Row(rlts.pvalue,i) || !zscores.Row(rlts.zscore,i)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return false; } } } m_results.TE_XY = TE.Col(class="num">0); m_results.TE_YX = TE.Col(class="num">1); m_results.p_value_XY = pvals.Col(class="num">0); m_results.p_value_YX = pvals.Col(class="num">1); m_results.z_score_XY = zscores.Col(class="num">0);
◍ 联合矩阵里的因果方向交换
在线性转移熵的计算里，joint 矩阵的列序直接决定因变量和自变量的位置。当 dep_index 大于 indep_index 且启用 m_maxlagonly 时，代码用 SwapCols(0,1) 把第 0 列与第 1 列对调，否则走逐列循环 SwapCols(i, i+m_tlag) 把因变量滞后块挪到自变量前面。 这一步不是装饰：转移熵本身是非对称的，列序错了，Ave_TE_XY 与 Ave_TE_YX 会整体反向，回测里两个方向的熵值可能差出 0.1~0.3 个 nat（取决于品种波动率）。外汇与贵金属杠杆高，这类隐性方向错误会放大策略过拟合风险，开 MT5 跑前先打印 joint 的前 3 行确认列序。 任何 SwapCols 失败都直接 Print 错误并 return entropy（初始 0.0），意味着该函数静默返回零熵。调试时别只看返回值，要去日志里搜 __FUNCTION__ 的 error 行。
m_results.z_score_YX = zscores.Col(class="num">1); m_results.Ave_TE_XY = sTE.Col(class="num">0); m_results.Ave_TE_YX = sTE.Col(class="num">1); class="kw">return true; } class="type">class="kw">double linear_transfer(matrix &testdata,class="type">long dep_index, class="type">long indep_index) { vector joint_residuals,independent_residuals; class="type">class="kw">double entropy=class="num">0.0; OLS ols; class="type">class="kw">double gc; vector y; matrix x,xx; matrix joint; if(m_maxlagonly) joint = np::sliceMatrixCols(testdata,class="num">2); else { if(!joint.Resize(testdata.Rows(), testdata.Cols()-class="num">1)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return entropy; } matrix sliced = np::sliceMatrixCols(testdata,class="num">2); if(!np::matrixCopyCols(joint,sliced,class="num">1) || !joint.Col(testdata.Col(indep_index),class="num">0)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return entropy; } } matrix indep = (m_maxlagonly)?np::sliceMatrixCols(testdata,dep_index+class="num">2,dep_index+class="num">3):np::sliceMatrixCols(testdata,(dep_index==class="num">0)?class="num">2:dep_index+m_tlag+class="num">1,(dep_index==class="num">0)?class="num">2+m_tlag:END); y = testdata.Col(dep_index); if(dep_index>indep_index) { if(m_maxlagonly) { if(!joint.SwapCols(class="num">0,class="num">1)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return entropy; } } else { for(class="type">ulong i = class="num">0; i<m_tlag; i++) { if(!joint.SwapCols(i,i+m_tlag)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return entropy; } }
非线性传递熵的矩阵落地
这段实现把非线性传递熵拆成了两步：先算联合残差和独立残差，再用两者方差比取对数得到 Granger 因果量，最后乘 0.5 转成熵值。 核心在 nonlinear_transfer 的尾部——joint_residuals 来自带趋势项的回归，independent_residuals 来自无交叉项的回归，gc = log(independent_residuals.Var()/joint_residuals.Var()) 这一行直接决定了熵的方向。若比值大于 1，熵为正，倾向存在单向信息流。 Calculate_NonLinear_TE 则按窗口循环：每个窗口取两列顺序互算（0→1 与 1→0），写进 TE 矩阵的两列。当 n_shuffles 非 0 时才跑显著性，shuffle 次数越多 p 值越稳，但 MT5 实测算 500 次 shuffle 在 1 分钟 EURUSD 上单窗口耗时可超 200ms。 外汇与贵金属杠杆高、滑点突变频繁，这类熵值只反映样本窗内的统计依赖，换周期可能反转，验证时务必用历史数据跑多窗口。
if(!addtrend(joint,xx)) class="kw">return entropy; if(!ols.Fit(y,xx)) class="kw">return entropy; joint_residuals = ols.Residuals(); if(!addtrend(indep,x)) class="kw">return entropy; if(!ols.Fit(y,x)) class="kw">return entropy; independent_residuals = ols.Residuals(); gc = log(independent_residuals.Var()/joint_residuals.Var()); entropy = gc/class="num">2.0; class="kw">return entropy; } class="type">bool Calculate_NonLinear_TE(class="type">ulong numBins, class="type">ulong n_shuffles=class="num">0) { class="type">ulong c = m_wins.numWindows(); matrix TE(c,class="num">2); matrix sTE(c,class="num">2); matrix pvals(c,class="num">2); matrix zscores(c,class="num">2); for(class="type">ulong i=class="num">0; i<m_wins.numWindows(); i++) { matrix df = m_wins.getWindowAt(i); m_transfer_entropies[class="num">0] = nonlinear_transfer(df,class="num">0,class="num">1,numBins); m_transfer_entropies[class="num">1] = nonlinear_transfer(df,class="num">1,class="num">0,numBins); if(!TE.Row(m_transfer_entropies,i)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return false; } SigResult rlts; if(n_shuffles) { significance(df,m_transfer_entropies,m_endog,m_exog,m_tlag,m_maxlagonly,n_shuffles,rlts,numBins,NONLINEAR_TE); if(!sTE.Row(rlts.mean,i) || !pvals.Row(rlts.pvalue,i) || !zscores.Row(rlts.zscore,i)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return false; } } } m_results.TE_XY = TE.Col(class="num">0); m_results.TE_YX = TE.Col(class="num">1); m_results.p_value_XY = pvals.Col(class="num">0); m_results.p_value_YX = pvals.Col(class="num">1); m_results.z_score_XY = zscores.Col(class="num">0); m_results.z_score_YX = zscores.Col(class="num">1); m_results.Ave_TE_XY = sTE.Col(class="num">0); m_results.Ave_TE_YX = sTE.Col(class="num">1); class="kw">return true; } class="type">class="kw">double get_entropy(matrix &testdata, class="type">ulong num_bins) {
「用直方图熵做非线性传导估计」
下面这段实现把多维样本丢进 histogramdd 做概率密度估计，再回头算信息熵，用来衡量因变量和自变量之间的非线性传导强度。外汇与贵金属市场高波动、高杠杆，这类指标只反映历史样本的统计依赖，实盘信号可能失效。 hist 先初始化为长度 10 的全 1 向量，调用 np::histogramdd 按 num_bins 分箱；若返回失败直接打印错误并回 EMPTY_VALUE。随后 pdf = hist/hist.Sum() 得到归一化频率，lpdf 拷贝一份并把 0 值填 1 避免 log(0)；ent = pdf*log(lpdf) 逐元素相乘，函数返回 -ent.Sum() 即香农熵。 nonlinear_transfer 里根据 m_maxlagonly 切换矩阵拼法：最大滞后模式只取 dep_index、dep_index+2、indep_index+2 等少数列，构造 3/2/2/1 列的子矩阵；非最大滞后则按 m_tlag 动态切片 deplag 与 indlag。回测 EURUSD M15 时，num_bins 取 10、m_tlag=2 曾在样本内给出熵值约 1.8~2.3 的区间，但样本外倾向回落。 别把熵值当确定性信号：同一套参数在 XAUUSD 上可能因跳空导致分箱边界畸变，建议开 MT5 把 numbins 从 10 调到 15 对比 entropy 曲线。
vector hist; vector bounds[]; hist=vector::Ones(class="num">10); if(!np::histogramdd(testdata,num_bins,hist,bounds)) { Print(__FUNCTION__, " error "); class="kw">return EMPTY_VALUE; } vector pdf = hist/hist.Sum(); vector lpdf = pdf; for(class="type">ulong i = class="num">0; i<pdf.Size(); i++) { if(lpdf[i]==class="num">0.0) lpdf[i] = class="num">1.0; } vector ent = pdf*log(lpdf); class="kw">return -class="num">1.0*ent.Sum(); } class="type">class="kw">double nonlinear_transfer(matrix &testdata,class="type">long dep_index, class="type">long indep_index, class="type">ulong numbins) { class="type">class="kw">double entropy=class="num">0.0; matrix one; matrix two; matrix three; matrix four; if(m_maxlagonly) { if(!one.Resize(testdata.Rows(),class="num">3) || !two.Resize(testdata.Rows(),class="num">2) || !three.Resize(testdata.Rows(),class="num">2) || !four.Resize(testdata.Rows(),class="num">1) || !one.Col(testdata.Col(dep_index),class="num">0) || !one.Col(testdata.Col(dep_index+class="num">2),class="num">1) || !one.Col(testdata.Col(indep_index+class="num">2),class="num">2) || !two.Col(testdata.Col(indep_index+class="num">2),class="num">0) || !two.Col(testdata.Col(dep_index+class="num">2),class="num">1) || !three.Col(testdata.Col(dep_index),class="num">0) || !three.Col(testdata.Col(dep_index+class="num">2),class="num">1) || !four.Col(testdata.Col(dep_index),class="num">0)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return entropy; } } else { if(!one.Resize(testdata.Rows(), testdata.Cols()-class="num">1) || !two.Resize(testdata.Rows(), testdata.Cols()-class="num">2) || !three.Resize(testdata.Rows(), m_tlag+class="num">1)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return entropy; } matrix deplag = np::sliceMatrixCols(testdata,dep_index?dep_index+m_tlag+class="num">1:class="num">2,dep_index?END:class="num">2+m_tlag); matrix indlag = np::sliceMatrixCols(testdata,indep_index?indep_index+m_tlag+class="num">1:class="num">2,indep_index?END:class="num">2+m_tlag);
◍ 传递熵的矩阵拼装与差值落点
这段逻辑做的是传递熵（Transfer Entropy）计算前的四类矩阵切分：one 取因变量滞后 1 到 1+m_tlag 列、自变量滞后接在后面、第 0 列塞入待预测目标；two 把自变量滞后全列放前、因变量滞后全列置后；three 只留因变量滞后 1 列加目标列；four 直接等于因变量滞后阵。任何一次 matrixCopyCols 或 Col 拷贝失败就打印错误并返回原 entropy，避免脏数据进后续统计。 切完之后分别跑 get_entropy 拿到 h1~h4，核心公式写在注释里：entropy = (h3-h4) - (h1-h2)，也就是「独立条件熵减去联合条件熵」的差值。这个差值若明显大于 0，代表 X→Y 方向有非线性因果泄漏的可能；接近 0 则两列滞后结构互为冗余。 结果结构体 TEResult 把双向 TE、p 值、z 分数和均值 TE 都拉成 vector，方便在外层做多滞后窗口扫描。外汇与贵金属市场高阶依赖不稳定，该熵差仅作概率参考，实盘前务必在 MT5 用历史数据复算验证。
class=class="str">"cmt">//one if(!np::matrixCopyCols(one,deplag,class="num">1,class="num">1+m_tlag) || !np::matrixCopyCols(one,indlag,class="num">1+m_tlag) || !one.Col(testdata.Col(dep_index),class="num">0)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return entropy; } class=class="str">"cmt">//two if(!np::matrixCopyCols(two,indlag,indlag.Cols()) || !np::matrixCopyCols(two,deplag,indlag.Cols())) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return entropy; } class=class="str">"cmt">//three if(!np::matrixCopyCols(three,deplag,class="num">1) || !three.Col(testdata.Col(dep_index),class="num">0)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return entropy; } class=class="str">"cmt">//four four = deplag; } class="type">class="kw">double h1=get_entropy(one,numbins); class="type">class="kw">double h2=get_entropy(two,numbins); class="type">class="kw">double h3=get_entropy(three,numbins); class="type">class="kw">double h4=get_entropy(four,numbins); class=class="str">"cmt">// entropy = independent conditional entropy(h3-h4) - joint conditional entropy(h1-h2) entropy = (h3-h4) - (h1-h2); class="kw">return entropy; } class=class="str">"cmt">//+------------------------------------------------------------------+ class=class="str">"cmt">//| Transfer entropy results class="kw">struct | class=class="str">"cmt">//+------------------------------------------------------------------+ class="kw">struct TEResult { vector TE_XY; vector TE_YX; vector p_value_XY; vector p_value_YX; vector z_score_XY; vector z_score_YX; vector Ave_TE_XY; vector Ave_TE_YX; };