网格和马丁格尔交易系统中的机器学习。 您敢为其打赌吗？（基础篇）
给网格马丁上机器学习，先看清赌注
网格和马丁格尔系统在 MT5 上跑了很多年，核心逻辑是用仓位递增加仓来摊平成本。这类系统最怕单边行情，一旦价格不回头，浮亏会按几何级数膨胀。 给它们接上机器学习，目的通常是让模型判断「当前该不该加仓」「网格间距该放大还是收紧」。但模型训练用的历史样本，永远覆盖不了未来所有极端波动。 外汇和贵金属市场高杠杆、高波动，马丁类策略在实盘里爆仓的概率始终存在。真要试，先在 MT5 策略测试器用 2020 年 3 月原油负价、黄金单日 80 美元波段的行情跑一遍压力测试，别直接上真金白银。
◍ 网格马丁遇上机器学习的冷门现实
在动手写这套实验之前，我们先在圈子里做了一轮摸底：向社区成员发问，问大家会不会把网格或马丁格尔策略接上现代机器学习算法。结果多数人的反馈是——根本不知道从哪下手，但觉得方向有意思。 更反常的是，公开网络里几乎搜不到把网格/马丁与ML结合的系统化资料，这个思路看似简单却极少被实做。外汇与贵金属本身高杠杆、高风险，这类加仓型策略一旦遇上单边行情，回撤可能被放大数倍。 我们做这一系列实验只奔两个落点：一是证明接ML没想象中难，二是用数据看它到底实不实用、有没有正期望。你打开MT5，新建一个EA框架就能跟着复现第一步。
「给成交和网格打方向标签」
为成交贴标签的核心，是判断一段行情里这笔单子该归为“买”还是“卖”。单仓情形很好处理：设一个成交边际，比如 15 根柱线，若行情在这 15 根内整体上行，就标“买入”，否则标“卖出”，过渡波动直接忽略。 到了订单网格，逻辑要改——不能看单笔，得算整组挂单触发后的总盈亏。假设边际同样是 15 根柱线（传统时间轴上红色竖线标记），若只有一笔持仓且价格从 A 点到 B 点上涨，标“买入”；若铺了红绿双向水平线网格，就要把自初始订单起所有触发的单子利润加总。 贴标签有三条可落地的准则，先记前两条：一是按最大总利润选边，卖网格赚得多就标“卖”；二是在持单数和总利润间做加权，若某侧每笔均利高于逆向，就选那侧。第三条按触发订单数最多且总仓获利来定方向，适合网格机器人。 ML 的任务是在滑动窗口里对全历史深度持续打标，再归纳特征预测新数据。外汇与贵金属杠杆高、回撤快，这类标签只是概率倾向，开 MT5 用 15 根边际先跑通单仓标法再扩到网格。
给网格成交打标签的数据结构
把常规网格和马丁格尔网格统一抽象，靠三个全局变量就能切换形态：GRID_SIZE 控制双向挂单数量，GRID_DISTANCES 设定各订单间距（固定或逐单不同），GRID_COEFFICIENTS 存每笔手数乘数。若乘数是常数就是规则网格，若从 1 线性拉到 3 就是马丁格尔倾向的递增仓。 numpy 里 np.full(10, 0.00200) 会生成 10 个 0.00200 的等距数组；np.linspace(1, 3, num=10) 则均匀铺出 [1, 1.222, 1.444 … 3] 的乘数序列。改这两个调用，就能在 MT5 外的研究脚本里重配网格，不用动主逻辑。 统计已触发和未触发订单用一本字典 grid_stats 装着每次样本的开平信息，比 pandas 或结构化数组在此处更直白。up_range / dwn_range 记价格摆动边界，up_state / dwn_state 记最后触发级别，利润初始值来自首单，后续挂单触发就累加。 遍历 GRID_DISTANCES 时，只要价格落进 up_range 或 dwn_range 就判定触发，并推高对应计数器；下一轮把新间距叠上去继续比。买入利润 =（平仓价 - 开仓价 + 选中挂单距离）× 该单乘数，卖出反向算。利润大于零且为最大才把样本贴 1.0/0.0 进训练集，否则贴 2.0 在清洗时丢掉。 下面这段代码把标签逻辑落了地：随机抽未来第 rand 根收盘价，相对当下加 MARKUP 判方向，2.0 的样本直接 drop 掉。你可以把 MARKUP 调成点差倍数，在 EURUSD 的 M1 上跑一遍看留存率。外汇与贵金属网格策略杠杆高，样本失效时回撤可能超预期，仅作技术验证。
def add_labels(dataset, min, max): labels = [] for i in range(dataset.shape[class="num">0]-max): rand = random.randint(min, max) curr_pr = dataset[&class="macro">#x27;close&class="macro">#x27;][i] future_pr = dataset[&class="macro">#x27;close&class="macro">#x27;][i + rand] if future_pr + MARKUP < curr_pr: labels.append(class="num">1.0) elif future_pr - MARKUP > curr_pr: labels.append(class="num">0.0) else: labels.append(class="num">2.0) dataset = dataset.iloc[:len(labels)].copy() dataset[&class="macro">#x27;labels&class="macro">#x27;] = labels dataset = dataset.dropna() dataset = dataset.drop( dataset[dataset.labels == class="num">2].index).reset_index(drop=True) class="kw">return dataset GRID_SIZE = class="num">10 GRID_DISTANCES = np.full(GRID_SIZE, class="num">0.00200) GRID_COEFFICIENTS = np.linspace(class="num">1, class="num">3, num= GRID_SIZE) array([class="num">1. , class="num">1.22222222, class="num">1.44444444, class="num">1.66666667, class="num">1.88888889, class="num">2.11111111, class="num">2.33333333, class="num">2.55555556, class="num">2.77777778, class="num">3. ]) def add_labels(dataset, min, max, distances, coefficients): labels = [] for i in range(dataset.shape[class="num">0]-max): rand = random.randint(min, max) all_pr = dataset[&class="macro">#x27;close&class="macro">#x27;][i:i + rand + class="num">1] grid_stats = {&class="macro">#x27;up_range&class="macro">#x27;: all_pr[class="num">0] - all_pr.min(), &class="macro">#x27;dwn_range&class="macro">#x27;: all_pr.max() - all_pr[class="num">0], &class="macro">#x27;up_state&class="macro">#x27;: class="num">0, &class="macro">#x27;dwn_state&class="macro">#x27;: class="num">0, &class="macro">#x27;up_orders&class="macro">#x27;: class="num">0, &class="macro">#x27;dwn_orders&class="macro">#x27;: class="num">0, &class="macro">#x27;up_profit&class="macro">#x27;: all_pr[-class="num">1] - all_pr[class="num">0] - MARKUP,
◍ 网格双向盈亏的向量化遍历与标签裁剪
这段逻辑把距离序列 distances 用 np.nditer 逐个喂进双向网格状态机：每步先判 up_state+i 是否仍在 up_range 内，是则累加状态、订单数，并按 coefficients 对应档位加权计算向上利润，再扣掉该档 MARKUP 成本。 向下分支同理，dwn_state 累加后用 all_pr[0]-all_pr[-1]+dwn_state 乘系数减点差，得到下行网格的浮动盈亏。注意 coefficients 用 int(orders-1) 取索引，意味着第 1 张单对应 coefficients[0]，仓位或手数系数必须从 0 开始对齐。 遍历结束后用 up_profit 与 dwn_profit 比大小：上行利厚且为正标 0.0，下行为正标 1.0，其余标 2.0。随后 dataset 按 labels 长度截断、赋标签、dropna，并直接删掉 label==2 的所有行——也就是砍掉「双向都不赚」的样本，只留有明确偏向的训练数据。 在 MT5 里复刻时，MARKUP 建议直接读 SymbolInfoDouble(_Symbol,SYMBOL_TRADE_TICK_VALUE) 近似，或用手数×合约乘数×点值硬算；若 coefficients 长度短于最大订单数，会抛 index 越界，这是实盘前必须卡住的参数边界。外汇与贵金属杠杆高，网格回测盈利不代表实盘能扛住跳空。
for i in np.nditer(distances): if grid_stats[&class="macro">#x27;up_state&class="macro">#x27;] + i <= grid_stats[&class="macro">#x27;up_range&class="macro">#x27;]: grid_stats[&class="macro">#x27;up_state&class="macro">#x27;] += i grid_stats[&class="macro">#x27;up_orders&class="macro">#x27;] += class="num">1 grid_stats[&class="macro">#x27;up_profit&class="macro">#x27;] += (all_pr[-class="num">1] - all_pr[class="num">0] + grid_stats[&class="macro">#x27;up_state&class="macro">#x27;]) \ * coefficients[class="type">int(grid_stats[&class="macro">#x27;up_orders&class="macro">#x27;]-class="num">1)] grid_stats[&class="macro">#x27;up_profit&class="macro">#x27;] -= MARKUP * coefficients[class="type">int(grid_stats[&class="macro">#x27;up_orders&class="macro">#x27;]-class="num">1)] if grid_stats[&class="macro">#x27;dwn_state&class="macro">#x27;] + i <= grid_stats[&class="macro">#x27;dwn_range&class="macro">#x27;]: grid_stats[&class="macro">#x27;dwn_state&class="macro">#x27;] += i grid_stats[&class="macro">#x27;dwn_orders&class="macro">#x27;] += class="num">1 grid_stats[&class="macro">#x27;dwn_profit&class="macro">#x27;] += (all_pr[class="num">0] - all_pr[-class="num">1] + grid_stats[&class="macro">#x27;dwn_state&class="macro">#x27;]) \ * coefficients[class="type">int(grid_stats[&class="macro">#x27;dwn_orders&class="macro">#x27;]-class="num">1)] grid_stats[&class="macro">#x27;dwn_profit&class="macro">#x27;] -= MARKUP * coefficients[class="type">int(grid_stats[&class="macro">#x27;dwn_orders&class="macro">#x27;]-class="num">1)] if grid_stats[&class="macro">#x27;up_profit&class="macro">#x27;] > grid_stats[&class="macro">#x27;dwn_profit&class="macro">#x27;] and grid_stats[&class="macro">#x27;up_profit&class="macro">#x27;] > class="num">0: labels.append(class="num">0.0) class="kw">continue elif grid_stats[&class="macro">#x27;dwn_profit&class="macro">#x27;] > class="num">0: labels.append(class="num">1.0) class="kw">continue labels.append(class="num">2.0) dataset = dataset.iloc[:len(labels)].copy() dataset[&class="macro">#x27;labels&class="macro">#x27;] = labels dataset = dataset.dropna() dataset = dataset.drop( dataset[dataset.labels == class="num">2].index).reset_index(drop=True) class="kw">return dataset