"""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()