| """Tick Forge audit runner and report (free companion script, MIT).
|
|
|
| The Tick Forge window builds an audit set of custom symbols and writes a manifest
|
| (MQL5\\Files\\TickForge\\audit_<name>.json). MQL5 programs cannot start the Strategy Tester, so this
|
| script does that part on a machine YOU choose: it runs one Expert over every symbol of the set,
|
| collects the tester reports and writes a plain comparison.
|
|
|
| run tf_audit.py run --manifest M.json --expert Folder\\Expert --terminal T\\terminal64.exe
|
| --data <terminal data folder> [--set my.set] [--period M5] [--model 1]
|
| [--deposit 10000] [--leverage 100] [--out audit_out]
|
| report tf_audit.py report --manifest M.json --reports DIR (re-read reports already collected)
|
|
|
| SAFETY
|
| * Use a TESTING terminal install, never a terminal that is logged in to a live-money account.
|
| The script refuses to start if that terminal is already running (it never kills anything).
|
| * --terminal and --data are mandatory: nothing is guessed.
|
| * The Expert runs only in the Strategy Tester; no orders reach any account.
|
|
|
| Models: 0 every tick, 1 1-minute OHLC, 2 open prices only, 4 every tick based on real ticks.
|
| Use 4 for symbols generated as 'real ticks', 1 for 'M1 bars only'.
|
| """
|
| import argparse, datetime as dt, html, json, os, re, shutil, subprocess, sys, time
|
| from pathlib import Path
|
|
|
|
|
| # --------------------------------------------------------------------- helpers
|
| def w16(path, text):
|
| Path(path).write_bytes(b"\xff\xfe" + text.replace("\n", "\r\n").encode("utf-16-le"))
|
|
|
|
|
| def running(exe):
|
| ps = ("Get-Process terminal64 -EA SilentlyContinue | Where-Object {$_.Path -eq '%s'} | "
|
| "Select-Object -First 1 -ExpandProperty Id" % exe.replace("'", "''"))
|
| out = subprocess.run(["powershell", "-NoProfile", "-Command", ps], capture_output=True, text=True).stdout.strip()
|
| return bool(out)
|
|
|
|
|
| def date_add(s, days=0, secs=0):
|
| d = dt.datetime.strptime(s, "%Y.%m.%d") + dt.timedelta(days=days, seconds=secs)
|
| return d.strftime("%Y.%m.%d")
|
|
|
|
|
| # --------------------------------------------------------------------- report parsing
|
| METRICS = (("net", "Total Net Profit:"), ("pf", "Profit Factor:"), ("trades", "Total Trades:"),
|
| ("ddrel", "Equity Drawdown Relative:"), ("dd", "Equity Drawdown Maximal:"), ("rf", "Recovery Factor:"),
|
| ("win", "Profit Trades (% of total):"), ("avgw", "Average profit trade:"), ("avgl", "Average loss trade:"),
|
| ("quality", "History Quality:"))
|
|
|
|
|
| def num(s):
|
| m = re.search(r"-?\d[\d\s]*\.?\d*", s.replace("\xa0", " "))
|
| return float(m.group(0).replace(" ", "")) if m else None
|
|
|
|
|
| def parse_report(path):
|
| raw = Path(path).read_bytes()
|
| txt = raw.decode("utf-16", "ignore") if raw[:2] in (b"\xff\xfe", b"\xfe\xff") else raw.decode("utf-8", "ignore")
|
| flat = re.sub(r"<[^>]+>", "|", txt)
|
| flat = html.unescape(re.sub(r"[\|\s]*\|[\|\s]*", "|", flat))
|
| out = {}
|
| for key, label in METRICS:
|
| m = re.search(re.escape(label) + r"\|([^|]*)", flat)
|
| out[key] = m.group(1).strip() if m else ""
|
| ent = []
|
| for m in re.finditer(r"(\d{4}\.\d\d\.\d\d \d\d:\d\d:\d\d)\|\d+\|[^|]*\|(buy|sell)\|in\|", flat):
|
| ent.append((m.group(1), m.group(2)))
|
| out["entries"] = ent
|
| return out
|
|
|
|
|
| # --------------------------------------------------------------------- comparison
|
| def to_dt(s):
|
| return dt.datetime.strptime(s, "%Y.%m.%d %H:%M:%S")
|
|
|
|
|
| def compare(manifest, runs, cap=None):
|
| """runs: {symbol: parsed report}. Returns rows + flags."""
|
| syms = manifest["symbols"]
|
| ctrl = next((s for s in syms if s["offset_sec"] == 0 and s["expr"].replace(" ", "") == "p"), None)
|
| rows, flags = [], []
|
| c = runs.get(ctrl["name"]) if ctrl else None
|
| cset = {to_dt(t) for t, _ in c["entries"]} if c else set()
|
| cap = cap or dt.datetime.now()
|
| end0 = dt.datetime.strptime(manifest["to"], "%Y.%m.%d") + dt.timedelta(days=1)
|
| for s in syms:
|
| r = runs.get(s["name"])
|
| row = {"name": s["name"], "label": s["label"], "offset": s["offset_sec"], "expr": s["expr"], "ok": r is not None}
|
| if r:
|
| row.update({k: r[k] for k, _ in METRICS})
|
| row["entries"] = len(r["entries"])
|
| off = dt.timedelta(seconds=s["offset_sec"])
|
| # the tester cannot use data dated after today: compare only the part of the control that fits
|
| row["truncated"] = (end0 + off) > cap
|
| sub = {t for t in cset if t + off <= cap} if row["truncated"] else cset
|
| row["ctrl_n"] = len(sub)
|
| back = {to_dt(t) - off for t, _ in r["entries"]}
|
| row["match"] = (len(back & sub) / len(sub)) if sub else None
|
| row["missing"] = sorted(sub - back)
|
| rows.append(row)
|
| if not c:
|
| flags.append("No control run found: compare against the control first, nothing below is reliable without it.")
|
| return rows, flags
|
| ctr = num(c["trades"]) or 0
|
| cnet = num(c["net"]) or 0.0
|
| if ctr < 20:
|
| flags.append("The control run has only %d trades. A few trades cannot separate logic from luck: use a longer range." % ctr)
|
| for row in rows:
|
| if not row["ok"] or row["name"] == ctrl["name"]:
|
| continue
|
| tr = num(row["trades"]) or 0
|
| net = num(row["net"]) or 0.0
|
| if row["offset"] and row["offset"] % (7 * 86400) == 0 and row["expr"].replace(" ", "") == "p":
|
| cn = row["ctrl_n"] or 0
|
| drop = (cn - tr) / cn * 100 if cn else 0
|
| if row.get("truncated"):
|
| flags.append("%s: the shifted data runs past today, so only the first part could be tested; compared on that overlap only (%d of %d control entries)." % (row["name"], cn, ctr))
|
| if abs(drop) >= 25:
|
| flags.append("%s: trade count moved %+.0f%% under a pure date shift of whole weeks (weekday and time of day unchanged)." % (row["name"], -drop))
|
| if row["match"] is not None and row["match"] < 0.7:
|
| flags.append("%s: only %.0f%% of the control entries reappear at the shifted time. Look at the missing entries below: do they sit on particular dates?" % (row["name"], row["match"] * 100))
|
| if cnet > 0 and net <= 0:
|
| flags.append("%s: the control made money, this shifted run did not (net %.2f vs %.2f)." % (row["name"], net, cnet))
|
| elif row["expr"].replace(" ", "") in ("p*0.5", "p*2"):
|
| if cnet > 0 and net <= 0:
|
| flags.append("%s: profitable on the control, not after rescaling the prices. Check for hard-coded price levels or distances." % row["name"])
|
| if not flags:
|
| flags.append("No rule-of-thumb flag fired. That means the Expert behaved similarly on these runs. It is not a guarantee about anything else.")
|
| return rows, flags
|
|
|
|
|
| CSS = ("body{font:15px/1.55 Arial,Helvetica,sans-serif;max-width:900px;margin:0 auto;padding:16px;color:#222}"
|
| "table{border-collapse:collapse;width:100%;margin:8px 0 16px}td,th{border:1px solid #bbb;padding:5px 8px;text-align:right}"
|
| "th{background:#f1f1f1}td:first-child,th:first-child{text-align:left}.col{max-width:680px}"
|
| ".flag{border-left:4px solid #b36b00;background:#fff8ec;padding:6px 12px;margin:6px 0}")
|
|
|
|
|
| def write_html(path, manifest, rows, flags, expert):
|
| def f(v):
|
| return html.escape(str(v)) if v not in (None, "") else "-"
|
| h = ["<!doctype html><html lang='en'><head><meta charset='utf-8'><title>Tick Forge audit report</title><style>%s</style></head><body>" % CSS,
|
| "<h1>Tick Forge audit report</h1><div class='col'>",
|
| "<p>Expert: <b>%s</b>. Source symbol: <b>%s</b>, %s to %s. Data: %s.</p>" % (
|
| f(expert), f(manifest.get("source")), f(manifest.get("from")), f(manifest.get("to")),
|
| "real ticks" if manifest.get("mode") == 0 else "M1 bars"),
|
| "<p>Every symbol below is the broker's own history with only the date or the price level changed. "
|
| "The control is the unchanged history. Flags are rules of thumb, not statistics, and a flag is a question, not a verdict.</p></div>",
|
| "<h2>Flags</h2>"]
|
| h += ["<div class='flag'>%s</div>" % html.escape(x) for x in flags]
|
| h.append("<h2>Results</h2><table><tr><th>Symbol</th><th>Change</th><th>Trades</th><th>Net</th><th>PF</th><th>Max equity DD</th>"
|
| "<th>Entries reproduced</th></tr>")
|
| for r in rows:
|
| if not r["ok"]:
|
| h.append("<tr><td>%s</td><td colspan='6'>no report</td></tr>" % f(r["name"]))
|
| continue
|
| m = ("%.0f%%" % (r["match"] * 100)) if r.get("match") is not None else "-"
|
| h.append("<tr><td>%s</td><td>%s</td><td>%s</td><td>%s</td><td>%s</td><td>%s</td><td>%s</td></tr>" % (
|
| f(r["name"]), f(r["label"]), f(r["trades"]), f(r["net"]), f(r["pf"]), f(r["ddrel"]), m))
|
| h.append("</table><p>Entries reproduced = share of the control run's entries that reappear at the same moment after undoing the time shift. "
|
| "It is only meaningful for the calendar-shift rows.</p>")
|
| for r in rows:
|
| if r["ok"] and r["offset"] and r.get("missing"):
|
| days = sorted({d.strftime("%Y.%m.%d %a") for d in r["missing"]})
|
| h.append("<h3>%s: control entries that did not reappear (%d)</h3><div class='col'><p>%s</p></div>" % (
|
| f(r["name"]), len(r["missing"]), html.escape(", ".join(days[:40]) + (" ..." if len(days) > 40 else ""))))
|
| h.append("<h2>What this does not show</h2><div class='col'><p>A pass means the Expert did not depend on what was changed. It does not prove "
|
| "the Expert is safe or profitable. A failure points at something to investigate (dates, weekdays, price levels). "
|
| "Backtests on altered data are stress tests, not forecasts or trading results.</p></div>")
|
| h.append("<p><small>Generated by tf_audit.py (Tick Forge, MIT).</small></p></body></html>")
|
| Path(path).write_text("\n".join(h), encoding="utf-8")
|
|
|
|
|
| # --------------------------------------------------------------------- run
|
| def run(a):
|
| manifest = json.loads(Path(a.manifest).read_text(encoding="utf-8"))
|
| exe = a.terminal
|
| if running(exe):
|
| sys.exit("That terminal is running. Close it first (this script never closes anything for you).")
|
| data = Path(a.data)
|
| out = Path(a.out)
|
| out.mkdir(parents=True, exist_ok=True)
|
| if a.set:
|
| prof = data / "MQL5" / "Profiles" / "Tester"
|
| prof.mkdir(parents=True, exist_ok=True)
|
| shutil.copy2(a.set, prof / Path(a.set).name)
|
| reports = {}
|
| for i, s in enumerate(manifest["symbols"]):
|
| tag = "tf_audit_%s" % s["name"]
|
| shift = int(s["offset_sec"])
|
| d1 = date_add(manifest["from"], secs=shift)
|
| d2 = date_add(manifest["to"], days=1, secs=shift)
|
| ini = out / (tag + ".ini")
|
| lines = ["[Tester]", "Expert=%s" % a.expert, "Symbol=%s" % s["name"], "Period=%s" % a.period, "Model=%s" % a.model,
|
| "FromDate=%s" % d1, "ToDate=%s" % d2, "ForwardMode=0", "Deposit=%s" % a.deposit, "Currency=%s" % a.currency,
|
| "Leverage=%s" % a.leverage, "Optimization=0", "Visual=0", "Report=%s" % tag, "ReplaceReport=1", "ShutdownTerminal=1"]
|
| if a.set:
|
| lines.insert(2, "ExpertParameters=%s" % Path(a.set).name)
|
| w16(ini, "\n".join(lines) + "\n")
|
| rep = data / (tag + ".htm")
|
| if rep.exists():
|
| rep.unlink()
|
| print("[%d/%d] %s (%s)" % (i + 1, len(manifest["symbols"]), s["name"], s["label"]), flush=True)
|
| p = subprocess.Popen([exe, "/config:%s" % ini])
|
| t0 = time.time()
|
| while p.poll() is None and time.time() - t0 < a.timeout:
|
| time.sleep(3)
|
| if p.poll() is None:
|
| print(" timeout after %d s; leaving the terminal for you to inspect" % a.timeout)
|
| break
|
| time.sleep(1)
|
| if rep.exists():
|
| shutil.copy2(rep, out / rep.name)
|
| reports[s["name"]] = parse_report(rep)
|
| r = reports[s["name"]]
|
| print(" trades %s net %s PF %s" % (r["trades"], r["net"], r["pf"]))
|
| else:
|
| print(" no report written (is the Expert name right, and the symbol present in Market Watch?)")
|
| finish(manifest, reports, out, a.expert)
|
|
|
|
|
| def finish(manifest, reports, out, expert):
|
| rows, flags = compare(manifest, reports)
|
| path = Path(out) / "tick_forge_audit_report.html"
|
| write_html(path, manifest, rows, flags, expert)
|
| print("\nFlags:")
|
| for x in flags:
|
| print(" -", x)
|
| print("Report:", path)
|
|
|
|
|
| def report(a):
|
| manifest = json.loads(Path(a.manifest).read_text(encoding="utf-8"))
|
| d = Path(a.reports)
|
| runs = {}
|
| for s in manifest["symbols"]:
|
| f = d / ("tf_audit_%s.htm" % s["name"])
|
| if f.exists():
|
| runs[s["name"]] = parse_report(f)
|
| finish(manifest, runs, d, a.expert or "(see reports)")
|
|
|
|
|
| def main():
|
| ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
| sub = ap.add_subparsers(dest="cmd", required=True)
|
| r = sub.add_parser("run")
|
| r.add_argument("--manifest", required=True)
|
| r.add_argument("--expert", required=True, help="path under MQL5\\Experts, e.g. MyFolder\\MyEA")
|
| r.add_argument("--terminal", required=True, help="full path of a TESTING terminal64.exe")
|
| r.add_argument("--data", required=True, help="that terminal's data folder")
|
| r.add_argument("--set", help="Expert settings file (.set); copied into Profiles\\Tester")
|
| r.add_argument("--period", default="M5")
|
| r.add_argument("--model", default="1")
|
| r.add_argument("--deposit", default="10000")
|
| r.add_argument("--currency", default="USD")
|
| r.add_argument("--leverage", default="100")
|
| r.add_argument("--timeout", type=int, default=3600)
|
| r.add_argument("--out", default="audit_out")
|
| p = sub.add_parser("report")
|
| p.add_argument("--manifest", required=True)
|
| p.add_argument("--reports", required=True)
|
| p.add_argument("--expert")
|
| a = ap.parse_args()
|
| run(a) if a.cmd == "run" else report(a)
|
|
|
|
|
| if __name__ == "__main__":
|
| main()
|
