# -*- coding: utf-8 -*- """ GridMaster EA 回测引擎 - 数据: Exness MT5 Trial5 的 XAUUSDm M30 历史 (.hc 解码, High/Low 为真实值) - 逻辑: 忠实复刻 GridMaster.mq5 的 OnTick 决策 (RSI信号/网格加仓/ATR止损/整体TP-SL-回撤退出) - 近似说明见报告: Close=(H+L)/2, Open=前收, 点差=0, 无隔夜利息, 爆仓=权益<=0 """ import struct, math, datetime, json, glob, os # ============================================================ # 1. 解码 M30.hc -> bars [t, O, H, L, C] # ============================================================ HC_PATH = r'C:\Users\Jeebin\AppData\Roaming\MetaQuotes\Terminal\9985D9581557BB9EE387EE0E3FAA60C3\bases\Exness-MT5Trial5\history\XAUUSDm\cache\M30.hc' def load_bars(): d = open(HC_PATH, 'rb').read() n = struct.unpack_from('= l - 1e-6): dropped += 1 continue c = (h + l) / 2.0 o = prevC if prevC is not None else c bars.append([idx[k], o, h, l, c]) prevC = c print('bars=%d dropped=%d span=%s ~ %s' % ( len(bars), dropped, datetime.datetime.utcfromtimestamp(bars[0][0]).strftime('%Y-%m-%d'), datetime.datetime.utcfromtimestamp(bars[-1][0]).strftime('%Y-%m-%d'))) return bars # ============================================================ # 2. 指标 (Wilder RSI / ATR, 与 iRSI / iATR 一致) # ============================================================ def wilder_rsi(closes, period): n = len(closes) rsi = [50.0] * n if n <= period: return rsi gains, losses = [0.0]*n, [0.0]*n for i in range(1, n): dch = closes[i] - closes[i-1] gains[i] = max(dch, 0.0) losses[i] = max(-dch, 0.0) ag = sum(gains[1:period+1]) / period al = sum(losses[1:period+1]) / period rsi[period] = 100.0 if al == 0 else 100.0 - 100.0/(1.0 + ag/al) for i in range(period+1, n): ag = (ag*(period-1) + gains[i]) / period al = (al*(period-1) + losses[i]) / period rsi[i] = 100.0 if al == 0 else 100.0 - 100.0/(1.0 + ag/al) return rsi def wilder_atr(H, L, C, period): n = len(C) tr = [0.0]*n for i in range(1, n): tr[i] = max(H[i]-L[i], abs(H[i]-C[i-1]), abs(L[i]-C[i-1])) atr = [0.0]*n if n <= period: return atr atr[period] = sum(tr[1:period+1]) / period for i in range(period+1, n): atr[i] = (atr[i-1]*(period-1) + tr[i]) / period return atr # ============================================================ # 3. 参数读取 (.set, UTF-16LE) # ============================================================ def load_set(fn): raw = open(fn, 'rb').read().decode('utf-16') d = {} for line in raw.splitlines(): if '=' in line: k, v = line.split('=', 1) d[k.strip()] = v.strip() def f(k, default=0.0): try: return float(d.get(k, default)) except: return default def i(k, default=0): try: return int(float(d.get(k, default))) except: return default def b(k): return d.get(k, 'false').lower() == 'true' return { 'InitialLot': f('InpInitialLot', 0.01), 'AddMultiplier': f('InpAddMultiplier', 1.2), 'AddDistancePips': i('InpAddDistancePips', 4500), 'MaxOrders': i('InpMaxOrders', 12), 'MaxLot': f('InpMaxLot', 0.4), 'OverallTp': f('InpOverallTpAmount', 800.0), 'OverallSl': f('InpOverallSlAmount', 0.0), 'RetraceStart': f('InpRetraceStart', 100.0), 'RetraceDrawdown': f('InpRetraceDrawdown', 10.0), 'AtrPeriod': i('InpAtrPeriod', 14), 'AtrThreshold': f('InpAtrThreshold', 5.0), 'AtrMultHigh': f('InpAtrMultHigh', 20.0), 'AtrMultLow': f('InpAtrMultLow', 25.0), 'RsiPeriod': i('InpRsiPeriod', 14), 'RsiUpper': f('InpRsiUpper', 70.0), 'RsiLower': f('InpRsiLower', 30.0), 'RsiMid': f('InpAddRsiMidline', 50.0), 'UseRsiAdd': b('InpRsiFilterAdd'), 'UseAtrSl': b('InpUseAtrSl'), 'UseTrailing': b('InpUseTrailingTp'), 'EquityStopPct': f('InpEquityStopPct', 0.0), 'AutoLot': b('InpIsAutoLot'), 'AutoLotMargin': f('InpAutoLotMargin', 20000.0), 'SessMon': d.get('InpSessionMonday', '02:00-19:00'), 'SessTue': d.get('InpSessionTuesday', '02:00-19:00'), 'SessWed': d.get('InpSessionWednesday', '02:00-19:00'), 'SessThu': d.get('InpSessionThursday', '02:00-19:00'), 'SessFri': d.get('InpSessionFriday', '02:00-19:00'), 'Magic': i('InpMagicNum', 1), 'OneEntryPerBar': b('InpOneEntryPerBar'), } # ============================================================ # 4. 时段判定 (Mon-Fri 02:00-19:00, 平台时间≈UTC) # ============================================================ def in_session(sess, dt): if not sess: return False now = dt.hour*60 + dt.minute for seg in sess.split(','): seg = seg.strip() if '-' not in seg: continue a, b = seg.split('-') try: h1, m1 = int(a[:2]), int(a[3:5]); h2, m2 = int(b[:2]), int(b[3:5]) except: continue s1, s2 = h1*60+m1, h2*60+m2 if s1 == s2: continue if s1 < s2: if s1 <= now < s2: return True else: if now >= s1 or now < s2: return True return False def session_ok(P, t): dt = datetime.datetime.utcfromtimestamp(t) dow = dt.weekday() # 0=Mon sess = [P['SessMon'], P['SessTue'], P['SessWed'], P['SessThu'], P['SessFri'], '', ''][dow] return in_session(sess, dt) # ============================================================ # 5. 手数计算 (复刻 CalcNextLot) # ============================================================ def norm_lot(lot, maxlot, step=0.01): v = math.floor(lot/step + 1e-9) * step if v < step: v = step if maxlot > 0 and v > maxlot: v = maxlot return round(v, 2) def calc_next_lot(count, P, equity): base = P['InitialLot'] if P['AutoLot']: base = base * (equity / P['AutoLotMargin']) if P['AutoLotMargin'] > 0 else base mult = P['AddMultiplier'] if P['AddMultiplier'] > 0 else 1.0 lot = base for _ in range(count): lot *= mult prev = norm_lot(lot/mult, P['MaxLot']) cur = norm_lot(lot, P['MaxLot']) if cur <= prev: cur = norm_lot(prev*mult + 0.01, P['MaxLot']) return cur # ============================================================ # 6. 回测主循环 # ============================================================ CONTRACT = 100.0 # 1 lot = 100 oz def run_backtest(bars, rsi_arr, atr_arr, P, start_equity=60000.0, stop_on_blowup=False): eq = start_equity balance = start_equity positions = [] # {type:1 buy/-1 sell, open, lot, sl, time} peak_profit = 0.0 sl_hit = False equity_stopped = False blown = False round_pnl = 0.0 # 本轮已实亏(含中途止损), 收盘时累加 equity_curve = [balance] max_eq = balance max_dd = 0.0 rounds = 0 r_wins = 0 r_losses = 0 r_blowups = 0 r_deep = 0 # 触达 MaxOrders 的轮次 round_profits = [] total_orders = 0 largest_loss = 0.0 cur_round_open_orders = 0 def basket_profit(price): p = 0.0 for pos in positions: p += pos['type'] * (price - pos['open']) * CONTRACT * pos['lot'] return p def close_all(price, reason): nonlocal balance, rounds, r_wins, r_losses, r_blowups, peak_profit, sl_hit, equity_stopped nonlocal cur_round_open_orders, largest_loss, round_pnl, r_deep realized = 0.0 for pos in positions: cp = price if reason != 'SL' else pos['sl'] realized += pos['type'] * (cp - pos['open']) * CONTRACT * pos['lot'] balance += realized round_pnl += realized is_win = round_pnl > 0 is_blow = reason == 'BLOWUP' rounds += 1 round_profits.append(round_pnl) if is_blow: r_blowups += 1 r_losses += 1 elif is_win: r_wins += 1 else: r_losses += 1 if cur_round_open_orders >= P['MaxOrders']: r_deep += 1 if realized < largest_loss: largest_loss = realized positions.clear() peak_profit = 0.0 sl_hit = False equity_stopped = False cur_round_open_orders = 0 round_pnl = 0.0 n = len(bars) for k in range(n): t, O, H, L, C = bars[k] rsi = rsi_arr[k] atr = atr_arr[k] dist = P['AddDistancePips'] / 100.0 # 归一后恒为美元 ($45 / $25) # --- (1) 上根 bar 设的 SL, 本根用极值判断是否触发 --- if positions: still = [] for pos in positions: if pos['sl'] and pos['sl'] > 0: if pos['type'] == 1 and L <= pos['sl']: # 多单止损 pnl = 1 * (pos['sl'] - pos['open']) * CONTRACT * pos['lot'] balance += pnl round_pnl += pnl sl_hit = True cur_round_open_orders -= 1 continue if pos['type'] == -1 and H >= pos['sl']: pnl = -1 * (pos['sl'] - pos['open']) * CONTRACT * pos['lot'] balance += pnl round_pnl += pnl sl_hit = True cur_round_open_orders -= 1 continue still.append(pos) positions = still # --- (2) 无持仓 -> 尝试开首单 --- if not positions: if equity_stopped: continue if not session_ok(P, t): continue if rsi <= P['RsiLower']: sig = 1 elif rsi >= P['RsiUpper']: sig = -1 else: sig = 0 if sig == 0: continue lot = calc_next_lot(0, P, balance) if lot <= 0: continue positions.append({'type': sig, 'open': C, 'lot': lot, 'sl': 0.0, 'time': t}) total_orders += 1 cur_round_open_orders = 1 continue # --- (3) 有持仓 -> 全局退出判定 --- profit = basket_profit(C) # 爆仓 if balance + profit <= 0: close_all(C, 'BLOWUP') equity_curve.append(balance) if stop_on_blowup: blown = True break continue # 整体止损 if P['OverallSl'] > 0 and profit <= -P['OverallSl']: close_all(C, 'SL') equity_curve.append(balance) continue # 整体止盈 if P['OverallTp'] > 0 and profit >= P['OverallTp']: close_all(C, 'TP') equity_curve.append(balance) continue # 回撤保护 if P['RetraceStart'] > 0 and P['RetraceDrawdown'] > 0: if profit > peak_profit: peak_profit = profit if peak_profit >= P['RetraceStart'] and profit <= peak_profit - P['RetraceDrawdown']: close_all(C, 'RETRACE') equity_curve.append(balance) continue # 权益保护 if P['EquityStopPct'] > 0 and balance + profit <= start_equity*(1-P['EquityStopPct']/100.0): close_all(C, 'EQUITY') equity_curve.append(balance) continue # --- (4) 网格加仓 --- buys = sum(1 for p in positions if p['type'] == 1) sells = sum(1 for p in positions if p['type'] == -1) if (buys > 0 and sells > 0) or len(positions) >= P['MaxOrders']: pass else: if sl_hit: pass else: is_buy = (buys > 0) last_price = max(positions, key=lambda p: p['time'])['open'] if is_buy: span = last_price - L else: span = H - last_price if span > 0: n_adds = int(span // dist) n_adds = min(n_adds, P['MaxOrders'] - len(positions)) for i in range(n_adds): if P['UseRsiAdd']: if is_buy and rsi > P['RsiMid']: break if (not is_buy) and rsi < P['RsiMid']: break level = last_price - (i+1)*dist if is_buy else last_price + (i+1)*dist lot = calc_next_lot(len(positions), P, balance) if lot <= 0: break positions.append({'type': 1 if is_buy else -1, 'open': level, 'lot': lot, 'sl': 0.0, 'time': t}) total_orders += 1 cur_round_open_orders += 1 # --- (5) ApplyStops: 设置 ATR 止损 (供下根 bar 检测) --- if P['UseAtrSl'] and atr > 0: mult = P['AtrMultHigh'] if atr >= P['AtrThreshold'] else P['AtrMultLow'] for pos in positions: if pos['type'] == 1: cand = pos['open'] - atr*mult if pos['sl'] == 0 or cand > pos['sl']: pos['sl'] = cand else: cand = pos['open'] + atr*mult if pos['sl'] == 0 or cand < pos['sl']: pos['sl'] = cand # --- 权益曲线 / 回撤 --- eq = balance + basket_profit(C) equity_curve.append(eq) if eq > max_eq: max_eq = eq dd = (max_eq - eq) / max_eq if max_eq > 0 else 0.0 if dd > max_dd: max_dd = dd # 收尾: 若仍有持仓, 强制平仓(按最后价) if positions: close_all(bars[-1][4], 'EOD') ret = (balance - start_equity) / start_equity win_rate = (r_wins / rounds * 100.0) if rounds else 0.0 avg_win = (sum(p for p in round_profits if p > 0) / r_wins) if r_wins else 0.0 avg_loss = (sum(p for p in round_profits if p < 0) / (r_losses)) if r_losses else 0.0 return { 'config': P.get('__name', '?'), 'start': start_equity, 'final': balance, 'return_pct': ret*100.0, 'max_dd_pct': max_dd*100.0, 'rounds': rounds, 'wins': r_wins, 'losses': r_losses, 'blowups': r_blowups, 'deep_grids': r_deep, 'win_rate': win_rate, 'total_orders': total_orders, 'avg_win': avg_win, 'avg_loss': avg_loss, 'largest_loss': largest_loss, 'round_profits': round_profits, 'equity_curve': equity_curve, } # ============================================================ # 7. 主程序 # ============================================================ def main(): bars = load_bars() closes = [b[4] for b in bars] highs = [b[2] for b in bars] lows = [b[3] for b in bars] rsi_arr = wilder_rsi(closes, 14) atr_arr = wilder_atr(highs, lows, closes, 14) results = [] sets = sorted(glob.glob('GridMaster_*.set')) for fn in sets: P = load_set(fn) P['__name'] = fn.replace('GridMaster_', '').replace('.set', '') print('--- 回测 %s ---' % P['__name']) r = run_backtest(bars, rsi_arr, atr_arr, P) results.append(r) print(' 终值=$%.0f 收益=%.1f%% 最大回撤=%.1f%% 轮次=%d 胜率=%.1f%% 爆仓=%d 触顶网格=%d 订单数=%d' % (r['final'], r['return_pct'], r['max_dd_pct'], r['rounds'], r['win_rate'], r['blowups'], r['deep_grids'], r['total_orders'])) print(' 均赢=$%.1f 均亏=$%.1f 最大单轮亏损=$%.1f' % (r['avg_win'], r['avg_loss'], r['largest_loss'])) json.dump(results, open('backtest_results.json', 'w'), default=lambda o: o.tolist() if hasattr(o, 'tolist') else o) print('\n已保存 backtest_results.json') if __name__ == '__main__': main()