266 라인
14 KiB
Python
 | """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()
|