基于转移熵的时间序列因果分析（基础篇）
用转移熵看清谁在带动谁
在 MT5 里做多品种联动，大多数人只看相关系数，但相关系数只说「一起动」，说不清「谁先动、谁被带着动」。转移熵（Transfer Entropy）补的就是这个方向性因果缺口：它衡量从序列 X 到序列 Y 的信息流向，数值越高，越倾向说明 X 的过去对预测 Y 的现在更有用。 外汇与贵金属这类高杠杆品种波动耦合快，用错因果方向可能把滞后指标当领先信号，实盘风险很高。下面这段 MQL5 片段演示了如何对两个收盘价序列计算基础转移熵，开 MT5 新建脚本粘进去就能跑出具体数值验证。 转移熵不是「稳赚信号」，只是把「谁带动谁」的概率结构量化出来；真要下单仍得结合价格行为结构，且外汇贵金属波动可能在极端行情下让历史信息流失效。
class="type">class="kw">double TransferEntropy(class="type">class="kw">double &x[], class="type">class="kw">double &y[], class="type">int k) { class="type">int n = ArraySize(x); if(n <= k+class="num">1) class="kw">return class="num">0; class="type">class="kw">double h_y=class="num">0, h_y_cond=class="num">0; class="type">int states[class="num">2][class="num">2] = {{class="num">0,class="num">0},{class="num">0,class="num">0}}; for(class="type">int i=k+class="num">1; i<n; i++) { class="type">int sy = (y[i]>y[i-class="num">1])?class="num">1:class="num">0; class="type">int sy_p = (y[i-class="num">1]>y[i-class="num">2])?class="num">1:class="num">0; class="type">int sx_p = (x[i-class="num">1]>x[i-class="num">2])?class="num">1:class="num">0; states[sy_p][sy]++; if(sx_p==class="num">1) states[sy_p][sy]++; } for(class="type">int a=class="num">0;a<class="num">2;a++) for(class="type">int b=class="num">0;b<class="num">2;b++) if(states[a][b]>class="num">0) h_y -= (class="type">class="kw">double)states[a][b]/n*MathLog((class="type">class="kw">double)states[a][b]/n); class="kw">return h_y - h_y_cond; }
「用传递熵看清谁在影响谁」
传递熵本质是一个统计度量，算的是一个时间序列把多少信息「推」给了另一个序列。它不只在意相关性，而是试图把因果影响的方向也量化出来——A 对 B 有驱动，还是 B 跟着 A 走，在耦合的时间序列里这点很关键。 在 MT5 里把这套方法落地，意味着你可以拿两段行情（比如黄金与美指小时线）跑一遍，看信息究竟从哪头传过来。识别出的「施动变量」可能比滞后指标更早暴露拐点倾向，对预测任务有增强作用，但外汇与贵金属高杠杆下任何信号都只是概率，需自行验证。 我们接下来会给出 MQL5 侧的具体实现，让你能直接复制进编辑器测自己的品种对。
◍ 别把同步涨跌当成因果
实证数据常带欺骗性。两个变量同步波动只说明相关，不证明谁驱动谁；冰淇淋销量与某股夏季同涨，背后隐藏因子是季节而非消费联动。外汇与贵金属同受通胀或避险情绪牵引时，也容易出现这种伪联系，交易者若直接据此开仓，属于把相关当因果。 格兰杰（Granger）在维纳基础上给出可操作的判据：若用 X 与 Y 的过去值比单用 Y 滞后值更能解释 Y 未来，则 X 对 Y 有预测性因果。两条原则很硬——效果不能早于原因，且原因须携带传递给结果的独特信息。这使得 MT5 里的时间序列回测天然适合做这类检验，但外汇/贵金属高杠杆高风险，结论只代表概率倾向。 数学上，设最大滞后 p，Y(t) 先仅对 Y(t−k) 回归；再把 X(t−k) 加入得非受限模型。若 X 过去值显著压低预测误差，即拒绝“X 不导致 Y”的零假设。F 统计量用受限/非受限残差平方和构建：F = [(RSS_r − RSS_ur)/p] / [RSS_ur/(n−2p−1)]，自由度 p 与 n−2p−1；大于临界值则判 X Granger-cause Y。该检验基于自回归，擅长抓线性因果，非线性关系需另寻方法。
用传递熵抓跨品种信息流
互信息只告诉你两个序列有没有共享信息，但分不清方向。想判断是 X 把不确定性传给了 Y，还是反过来，得看传递熵（TE）：它基于 Kullback-Leibler 散度，量化在已知 Y 自身过去和 X 过去的前提下，Y 下一步概率分布还变了多少。
- 年 Barnett、Barrett 与 Seth 证明过：当序列是高斯分布时，TE 恰好等于格兰杰因果 F 统计量的一半。我们代码里先落这条线性定义，再借 Schreiber 的思路把序列当马尔可夫过程，用条件互信息扩到非线性——单个滞后 k 的 TE 可拆成四个联合熵项，只依赖概率分布，不预设数据生成过程。
代价也很直接：每多一个滞后，状态空间维度翻两倍，有限样本下概率密度估计误差指数级放大。外汇与贵金属这种高噪声品种里，算出的熵值很可能是伪信号，必须做显著性检验。 常用两招：一是把序列随机打乱重算 TE，原始值低于打乱版的比例就是 p 值，越接近 0 越显著；二是算 z 分数，即原始 TE 偏离打乱均值多少个标准差，打乱破坏了时间结构均值近 0，z 高于 3.0 才倾向认为信息传递真实存在。z 分数通常比 p 值少打乱几次，MT5 上跑起来更省时间。
「在MT5里跑通传递熵分析的类骨架」
把两个品种的价格序列塞进一个至少两列的 matrix，用 CTransEntropy 的 Initialize() 把内生、外生列索引和滞后参数交进去，就能拿到一套双向因果检验框架。非平稳数据建议先差分，再把 winsize 设成非零值走滑动窗口——窗口化和步长由你定，能缓解非平稳带来的伪因果。 类里同时给了线性格兰杰因果（linear_transfer）和非线性直方图熵（nonlinear_entropy）两条路。Calculate_Linear_TE() 若不传 n_shuffles 默认 0，即不做显著性；想拿 p 值和 z 分数就给个打乱次数，它会在保留边缘分布下重排后给出 TE_XY / TE_YX 两侧的 z_score 与 p_value。 下面这段是无参构造和 Initialize 的前半段，几个硬约束直接写在里面对 lag 和索引做了拦截：lag 必须 >0 且 < 总行数一半，endog 不能等于 exog，否则打印错误并返回 false。m_dataset 被强制Resize成 [行数,2]，只抽两列进来。 调用完计算函数，用 get_results() 取向量结构，最后一窗的双向熵也可经 get_transfer_entropies() 直接拿——向量顺序严格跟随原始 matrix 的列序，第一列序列对应第一个熵。外汇与贵金属波动大、跳空频繁，窗口参数设错可能把噪声当因果，上线前务必用历史数据复算一遍。
class="kw">public: CTransEntropy(class="type">void) { if(!m_transfer_entropies.Resize(class="num">2)) Print(__FUNCTION__, " error ", GetLastError()); } class="type">bool Initialize(matrix &in, class="type">ulong endog_index, class="type">ulong exog_index, class="type">ulong lag, class="type">bool maxLagOnly=true, class="type">ulong winsize=class="num">0,class="type">ulong winstride=class="num">0) { if(!lag || lag>in.Rows()/class="num">2) { Print(__FUNCTION__, " Invalid parameter(s) : lag must be > class="num">0 and < rows/class="num">2"); class="kw">return false; } if(endog_index==exog_index) { Print(__FUNCTION__, " Invalid parameter(s) : endog cannot be = exog "); class="kw">return false; } if(!m_dataset.Resize(in.Rows(),class="num">2)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return false; } if(!m_dataset.Col(in.Col(endog_index),class="num">0) || !m_dataset.Col(in.Col(exog_index),class="num">1)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return false; } if(!m_wins.Initialize(m_dataset,lag,maxLagOnly,winsize,winstride)) class="kw">return false; m_tlag = lag; m_endog = endog_index; m_exog = exog_index; m_maxlagonly = maxLagOnly;
◍ 数据集滑窗的滞后展开逻辑
CDataWindows 这个类负责把原始矩阵切成带滞后的分析窗口。核心在 applylags 私有方法：先按 m_lag 把数据行切片，再决定只保留最大滞后还是展开全部滞后阶。 当 m_max_lag_only 为真，输出矩阵只扩 2 列（原列数+2），循环 i 从 2 到 3，把 m_data 的第 0、1 列做去头切片后塞进新列。这意味着只取 lag=1 的偏移量，适合快速看最近一期滞后特征。 若 m_max_lag_only 为假，输出列数变成原列数 + m_lag*2，双层循环把前两列各自展开成 1 到 m_lag 阶滞后。比如 m_lag=5 时，前两列会多出 10 列滞后特征，样本行数相应减少 5 行。 任何 Resize 或 Col 写入失败都会打印函数名和 GetLastError 并返回 1x1 零矩阵，调用方得自行判断返回尺寸来识别异常。外汇与贵金属行情序列用这套滑窗前，建议先小 lag 跑通再放大，杠杆品种回测误差可能被放大。
class CDataWindows { class="kw">private: matrix m_dwins[], m_data; class="type">ulong m_lag, m_win_size, m_stride_size; class="type">bool m_max_lag_only, m_has_windows; matrix applylags(class="type">void) { matrix out=np::sliceMatrixRows(m_data,m_lag); if(m_max_lag_only) { if(!out.Resize(out.Rows(),m_data.Cols()+class="num">2)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return matrix::Zeros(class="num">1,class="num">1); } for(class="type">ulong i = class="num">2; i<class="num">4; i++) { vector col = m_data.Col(i-class="num">2); col = np::sliceVector(col,class="num">0,col.Size()-m_lag); if(!out.Col(col,i)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return matrix::Zeros(class="num">1,class="num">1); } } } else { if(!out.Resize(out.Rows(),m_data.Cols()+(m_lag*class="num">2))) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return matrix::Zeros(class="num">1,class="num">1); } for(class="type">ulong i = class="num">0,k = class="num">2; i<class="num">2; i++) { for(class="type">ulong t = class="num">1; t<(m_lag+class="num">1); t++,k++) { vector col = m_data.Col(i); col = np::sliceVector(col,m_lag-t,col.Size()-t); if(!out.Col(col,k)) { Print(__FUNCTION__, " error ", GetLastError()); class="kw">return matrix::Zeros(class="num">1,class="num">1); } } } } class="kw">return true; } }