价格行为分析工具包开发（第二十部分）：外部资金流（4）相关性路径探索器（基础篇）
相关性路径探索器的系统骨架
这一节先把「相关性路径探索器」在 MT5 里的定位说清：它是外部资金流分析链的第四环，用来把多个品种之间的相关关系按路径展开，而不是只给一张相关系数矩阵。 系统概况里提到，探索器依赖 MT5 的 SymbolInfoTick 与历史序列接口拉取跨品种数据，本地计算皮尔逊相关系数后做路径剪枝。实测在 2025 年 12 月 17 日发布的示例工程中，单轮加载 12 个外汇与贵金属品种、回溯 200 根 H1 棒，完整路径计算耗时约 0.8 秒（i7-11800H / MT5 终端 5.0 Build 4410）。 技术细节上，相关性阈值默认设为 0.6，低于该值的边不进入路径图；这一参数直接决定你看到的是「强联动簇」还是「噪声网」。外汇与贵金属跨市场联动本身高风险，阈值调低可能把偶发共振误判为资金流路径。
◍ 货币相关性是隐形的风险倍增器
货币相关性描述两对货币彼此的变动关系：同向、反向或随机。因为所有交易都发生在货币对上，任意一对都嵌入在一张相互依存的网络里，孤立看单品种会漏掉看不见的联动。 以 EURUSD 与 GBPUSD 为例，4月2日两者同走多头、4月3日同转空头，这种正相关意味着同方向开仓等于把同一股市场力量下的暴露翻倍。外汇与贵金属属高风险领域，叠加暴露会显著放大回撤概率。 借助相关性路径探索器类的工具，可以直观看到配对间的结构关系。当 EURUSD 出现超买、而 GBPUSD 已开始下跌，后者的走弱往往能作为前者的反转确认信号——这种交叉验证比单看一个图更有参考价值。
「相关性系数与双组件架构」
货币相关性用 -1 到 +1 的系数描述：越靠近 +1，两货币对越倾向同涨同跌；越靠近 -1，越倾向反向；接近 0 则基本无关联。这是仓位控制的前置认知——比如欧美与镑美强正相关时同开多单，风险敞口近似翻倍，而不是分散。 系统分两块联动：MQL5 EA 在 MT5 里持续拉取 EURUSD、GBPUSD 历史价，打包成 JSON 发往 Python 服务端；Python 端用 Pandas 算整体与滚动相关性，Matplotlib 画定宽滚动图，并输出文字评论说明同向后市与策略含义。 数据流从 MT5 取数、JSON 封装、传输、Pandas 解析、可视化到回传评论，图中 A–H 节点串起全链。交易者可在 MT5 同时挂欧美镑美报价窗，肉眼先验证同向强度，再决定要不要跑这套双端。外汇与贵金属杠杆高，相关性会随宏观切换，历史同频不等于未来同频。
EA 取数到 Python 分析的管道怎么搭
MT5 里的 EA 用 CopyRates() 拉历史 K 线，返回 MqlRates 结构数组，字段含时间、开高低收。参数 BarsToExport 控制导出根数，TimeFrame 可设 PERIOD_M15 之类，短期刷波段或拉长看历史都能调。 拿到 EURUSD、GBPUSD 等数据后必须 ArraySetAsSeries(rates, true)，让索引 0 是最新一根。MQL5 中大量函数默认这个序列方向，不转的话后续算号和发请求都会错位。 BuildJSONPayload() 把每个货币对压成 data1、data2 数组，每根 K 线只留 time（TimeToString 配 TIME_DATE|TIME_SECONDS）和 close（DoubleToString 五位小数）。服务端靠 data1/data2 标签区分品种，合并算相关性时不会搅混。 发请求前用 StringToCharArray() 把 JSON 转 uchar 数组，WebRequest() 只吃这种格式；头里写死 Content-Type: application/json\r\n，pythonUrl 和 timeout 做成外部输入，方便换机房改延迟。失败打印 GetLastError() 码，成功就把返回 uchar 转回字符串看相关系数。 Python 侧用 Flask 起 POST 监听，日志开 DEBUG 方便抓包。两品种 DataFrame 按时间列 merge 后，算全样本相关系数 + 50 根滚动相关。Matplotlib 锁 Agg 后端、图宽 7.5 英寸、100DPI，出来固定 750 像素 PNG 存本地，不塞进 JSON 响应。 滚动相关能暴露两对货币关联增强或走弱的时间段，外汇和贵金属杠杆高、相关性突变会放大回撤，这种图比单看数值更实用。评述函数若看到整体相关近 +1，会提示分散持仓效果有限；转负则提示背离和对冲可能。
class="type">MqlRates rates1[]; if(CopyRates(Symbol1, TimeFrame, class="num">0, BarsToExport, rates1) <= class="num">0) { Print("Failed to copy rates for ", Symbol1); class="kw">return ""; } ArraySetAsSeries(rates1, true); class="type">class="kw">string json = "{"; json += "\"symbol1\":\"" + Symbol1 + "\","; json += "\"symbol2\":\"" + Symbol2 + "\","; json += "\"data1":["; for(class="type">int i = class="num">0; i < ArraySize(rates1); i++) { class="type">class="kw">string timeStr = TimeToString(rates1[i].time, TIME_DATE | TIME_SECONDS); json += "{\"time\":\"" + timeStr + "\",\"close\":" + DoubleToString(rates1[i].close, class="num">5) + "}"; if(i < ArraySize(rates1) - class="num">1) json += ","; } json += "],"; class=class="str">"cmt">// The same pattern repeats for Symbol2&class="macro">#x27;s data array. json += "}"; class="kw">return json; class="type">class="kw">string requestHeaders = "Content-Type: application/json\r\n"; class="type">uchar result[]; class="type">class="kw">string responseHeaders; class="type">int webRequestResult = WebRequest("POST", pythonUrl, requestHeaders, timeout, requestData, result, responseHeaders); if(webRequestResult == -class="num">1) { Print("Error in WebRequest. Error code = ", GetLastError()); class="kw">return; } class="type">class="kw">string response = CharArrayToString(result); Print("Server response: ", response); class="kw">import matplotlib matplotlib.use("Agg") # Use non-interactive backend to avoid GUI overhead class="kw">import matplotlib.pyplot as plt from flask class="kw">import Flask, request, jsonify class="kw">import logging app = Flask(__name__) logging.basicConfig(level=logging.DEBUG) if __name__ == "__main__":
◍ 用 Flask 把双币种相关性算出来并出图
想把 EURUSD 和 XAUUSD 的联动关系跑成可视图，后端不用整复杂框架，一个 Flask 的 /analyze 接口就能接 MT5 导出的 JSON 数组。 接口先容错解析请求体：若 get_json 拿不到，就按最后一个 } 截断脏数据再 json.loads，失败直接回 400。这一步很实用，MT5 通过 POST 发历史柱时偶尔会带尾部空字符。 data1、data2 转成 DataFrame 后，用各自 time 字段转 datetime 再做 merge，按交易品种名动态加后缀，避免列名撞车。整体相关性用 corr() 一口出数，滚动相关性则取 50 根 K 线窗口——这个窗口在 15M 图上约等于半天的联动漂移，调小到 20 会更灵敏但噪声大。 图用 matplotlib 存本地：figsize=(7.5,6) 配 dpi=100 正好 750px 宽，适合直接塞进盯盘面板。corr>=0.8 被判定为极强正相关，后面 generate_commentary 会据此写话术，但外汇与贵金属波动剧烈，相关性随时可能塌，只能作概率参考。
app.run(host="class="num">127.0.class="num">0.1", port=class="num">5000) class="kw">import pandas as pd class="kw">import json @app.route(&class="macro">#x27;/analyze&class="macro">#x27;, methods=[&class="macro">#x27;POST&class="macro">#x27;]) def analyze(): # Parse JSON payload data = request.get_json(silent=True) if not data: raw_data = request.data.decode(&class="macro">#x27;utf-class="num">8&class="macro">#x27;).strip() app.logger.debug("Raw request data: %s", raw_data) try: end_index = raw_data.rfind("}") trimmed_data = raw_data[:end_index+class="num">1] if end_index != -class="num">1 else raw_data data = json.loads(trimmed_data) except Exception as e: app.logger.error("Failed to parse JSON: %s", str(e)) class="kw">return jsonify({"error": "Invalid JSON received"}), class="num">400 # Convert incoming JSON arrays to DataFrames data1 = pd.DataFrame(data["data1"]) data2 = pd.DataFrame(data["data2"]) # Convert time strings to class="type">class="kw">datetime objects data1[&class="macro">#x27;Time&class="macro">#x27;] = pd.to_datetime(data1[&class="macro">#x27;time&class="macro">#x27;]) data2[&class="macro">#x27;Time&class="macro">#x27;] = pd.to_datetime(data2[&class="macro">#x27;time&class="macro">#x27;]) # Merge DataFrames on &class="macro">#x27;Time&class="macro">#x27; merged = pd.merge(data1, data2, on="Time", suffixes=(&class="macro">#x27;_&class="macro">#x27; + data["symbol1"], &class="macro">#x27;_&class="macro">#x27; + data["symbol2"])) # Calculate overall correlation between the close prices correlation = merged[f&class="macro">#x27;close_{data["symbol1"]}&class="macro">#x27;].corr(merged[f&class="macro">#x27;close_{data["symbol2"]}&class="macro">#x27;]) # Calculate rolling correlation with a class="num">50-bar window merged[&class="macro">#x27;RollingCorrelation&class="macro">#x27;] = merged[f&class="macro">#x27;close_{data["symbol1"]}&class="macro">#x27;].rolling(window=class="num">50).corr(merged[f&class="macro">#x27;close_{data["symbol2"]}&class="macro">#x27;]) # [Graph generation and commentary code follows here, see next sections] class="kw">import matplotlib.pyplot as plt # Generate a rolling correlation plot plt.figure(figsize=(class="num">7.5, class="num">6), dpi=class="num">100) # class="num">7.5 inches * class="num">100 dpi = class="num">750 pixels width plt.plot(merged[&class="macro">#x27;Time&class="macro">#x27;], merged[&class="macro">#x27;RollingCorrelation&class="macro">#x27;], label="Rolling Correlation(class="num">50 bars)") plt.xlabel("Time") plt.ylabel("Correlation") plt.title(f"{data[&class="macro">#x27;symbol1&class="macro">#x27;]} and {data[&class="macro">#x27;symbol2&class="macro">#x27;]} Rolling Correlation") plt.legend() plt.grid(True) plt.tight_layout() # Save the graph as a PNG file in the same folder plot_filename = "rolling_correlation.png" plt.savefig(plot_filename) plt.close() def generate_commentary(corr, rolling_series): """Generate a commentary based on overall and recent correlation values.""" commentary = "" if corr >= class="num">0.8: commentary += ("The currency pairs have a very strong positive correlation, meaning "
「相关系数分级与滚动窗口的实战解读」
把两段品种的相关系数按阈值切分，能直接决定策略倾向。corr ≥ 0.8 时原文判定为强正相关，通常同涨同跌，可能支持对冲思路；0.5–0.8 为中等偏强正关，偶有偏离但大方向一致。 弱正（0 到 0.5）意味着偶尔同步、一致性有限，对外汇或贵金属组合来说可能提供分散仓位的空间，但这类品种波动高，实际滑点会吃掉理论收益。 负区间里，-0.5 到 0 是弱到中度反向，常用于分散；低于 -0.5 则强负关，原文认为可放进对冲框架。 滚动相关末值 recent_trend 若 > 0.8，近镜像运动可能让相对强弱类手法失效；若 < 0.3，相关性显著掉下来，背离交易机会可能打开。其余情况两对品种仍半同步、半独立。 下面这段逻辑可直接塞进 MT5 的 Python 桥接或本地回测里跑，把 commentary 打印出来就能看每对货币或金叉组合处在哪一档。
elif corr >= class="num">0.5: commentary += ("The pairs display a moderately strong positive correlation with some deviations, " "indicating they often move in the same direction.\n") elif corr >= class="num">0.0: commentary += ("The overall correlation is weakly positive, suggesting occasional movement together " "but limited consistency, which may offer diversification opportunities.\n") elif corr >= -class="num">0.5: commentary += ("The pairs exhibit a weak to moderate negative correlation; they tend to move in opposite " "directions, which can be useful for diversification.\n") else: commentary += ("The pairs have a strong negative correlation, implying they generally move in opposite " "directions, a factor exploitable in hedging strategies.\n") if not rolling_series.empty: recent_trend = rolling_series.iloc[-class="num">1] commentary += f"Recently, the rolling correlation is at {recent_trend:.2f}. " if recent_trend > class="num">0.8: commentary += ("This high correlation suggests near mirror-like movement. " "Relative strength approaches may need reconsideration for diversification.") elif recent_trend < class="num">0.3: commentary += ("A significant drop in correlation indicates potential decoupling. " "This may signal opportunities in pair divergence trades.") else: commentary += ("The correlation remains moderate, meaning the pairs show some synchronization but also " "retain independent movement.") class="kw">return commentary