Warrior_EA/research/combine_charts.py

"""
Rebuild the ACCOUNT equity curve from per-chart tester runs.

The EA now trades one symbol per chart, so each tester run sees only its own
symbol and its report's drawdown is a per-symbol drawdown. The account's
drawdown is what the prop limit is measured on. Each run writes a per-bar log
(realised P&L so far, floating P&L at the bar close, worst floating inside the
bar); this sums them on a common clock (forward-filled) into:

  close curve  : realised + floating-at-close, summed   -> realistic
  worst curve  : realised + worst-in-bar floating, summed -> pessimistic bound
                 (per-symbol worsts need not coincide, so this overstates DD)

    python combine_charts.py claude_pc_SP500 claude_pc_NAS100 ...
"""

from __future__ import annotations

import sys
import numpy as np

COMMON = r"C:\Users\admin\AppData\Roaming\MetaQuotes\Terminal\Common\Files"
DEPOSIT = 100000.0


def load_eq(name):
    raw = np.genfromtxt(rf"{COMMON}\dipz_eq_{name}.csv", delimiter=",", skip_header=1,
                        dtype=str, encoding="ansi")
    t = np.array([np.datetime64(r[0][:10].replace(".", "-") + "T" + r[0][11:] + ":00") for r in raw])
    return t, raw[:, 1].astype(float), raw[:, 2].astype(float), raw[:, 3].astype(float)


def load_trades(name):
    raw = np.genfromtxt(rf"{COMMON}\dipz_trades_{name}.csv", delimiter=",", skip_header=1,
                        dtype=str, encoding="ansi")
    if raw.ndim == 1:
        raw = raw[None, :]
    ts = lambda s: np.datetime64(s[:10].replace(".", "-") + "T" + s[11:] + ":00")  # noqa: E731
    return [dict(sym=r[1], t=ts(r[2]), x=ts(r[5]), net=float(r[7]), swap=float(r[10]),
                 gross=float(r[9])) for r in raw]


def dd_of(curve):
    peak = np.maximum.accumulate(curve)
    return ((peak - curve) / peak).max()


def combine(names):
    series = [load_eq(n) for n in names]
    clock = np.unique(np.concatenate([s[0] for s in series]))
    close = np.full(len(clock), DEPOSIT)
    worst = np.full(len(clock), DEPOSIT)
    for t, rl, fc, fm in series:
        idx = np.searchsorted(t, clock, side="right") - 1      # forward-fill
        ok = idx >= 0
        close[ok] += rl[idx[ok]] + fc[idx[ok]]
        worst[ok] += rl[idx[ok]] + fm[idx[ok]]
    return clock, close, worst


def summary(names, label=""):
    clock, close, worst = combine(names)
    trades = sum((load_trades(n) for n in names), [])
    yrs = (clock[-1] - clock[0]) / np.timedelta64(365, "D")
    net = close[-1] / DEPOSIT - 1
    d_c, d_w = dd_of(close), dd_of(worst)
    swap = sum(t["swap"] for t in trades)
    gross = sum(t["gross"] for t in trades)
    permo = len(trades) / (yrs * 12)
    fails = [k for k, bad in (("cadence", permo < 2), ("DD", d_c > 0.05), ("ret/DD", net / d_c < 2)) if bad]
    print(f"{label:<22}{len(trades):>5}{permo:>6.1f}{net:>+8.1%}{((1 + net) ** (1 / yrs) - 1):>+7.1%}"
          f"{d_c:>8.2%}{d_w:>8.2%}{net / d_c:>8.2f}{swap / gross if gross else 0:>7.0%}  "
          f"{'PASS' if not fails else 'fail:' + ','.join(fails)}")
    return clock, close, trades


HDR = (f"{'portfolio':<22}{'n':>5}{'/mo':>6}{'net':>8}{'CAGR':>7}{'DD':>8}{'DDworst':>8}"
       f"{'ret/DD':>8}{'swap%':>7}  screen")

if __name__ == "__main__":
    names = sys.argv[1:]
    print(HDR)
    clock, close, trades = summary(names, "combined")
    for n in names:
        summary([n], "  " + n.replace("claude_", ""))
    yr = {}
    for t in trades:
        yr[str(t["x"])[:4]] = yr.get(str(t["x"])[:4], 0) + t["net"]
    print("\nnet by year:", {k: round(v) for k, v in sorted(yr.items())})
    print("DD = summed close-of-bar equity (realistic); DDworst = summed worst-in-bar (pessimistic bound)")