Warrior_EA/research/backtest.py

"""
Multi-symbol strategy screen, ranked the way the account is actually judged.

RANKING RULE (non-negotiable here): drawdown first, not profit. A candidate is
only reported as viable if it clears ALL THREE of
    * cadence  >= 2 trades / month     (a strategy that never trades is not one)
    * maxDD    <= the prop limit
    * ret/DD   >= 2
Profit ranking is what produces the curve-fit winners this project keeps
re-discovering, so it is deliberately not the sort key.

COSTS ARE REAL. The exported `spread` column is in POINTS at each bar, and
point size is recovered from the decimal count the exporter wrote (it used
DoubleToString(price, digits), so the file's precision IS the symbol's digits).
A long round trip pays the full spread once: you buy the ask and sell the bid.
This matters more than any parameter -- the decade verdict on this project was
that spread is the whole gap between paper and traded results.

CONTROL. Every candidate is compared against a LONG-ONLY RANDOM ENTRY with a
matched holding period, not a two-sided coin flip. On a drifting index a
two-sided control is a straw man: it loses money by construction, so beating
it proves nothing. Long bias on an index is not a flaw, it is the asset.
"""

from __future__ import annotations

import numpy as np

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


# ------------------------------------------------------------------ loading
def load(symbol: str, period: str = "PERIOD_D1"):
    path = rf"{COMMON}\bars_{symbol}_{period}.csv"
    raw = np.genfromtxt(path, delimiter=",", skip_header=1, dtype=str, encoding="ansi")
    ts = np.array([f"{r[0][:10].replace('.', '-')}T{r[0][11:]}" for r in raw],
                  dtype="datetime64[s]")
    o, h, l, c = (raw[:, i].astype(float) for i in (1, 2, 3, 4))
    vol = raw[:, 5].astype(float)
    spread_pts = raw[:, 6].astype(float)
    # digits == decimals the exporter wrote; point == 10^-digits
    digits = max(len(s.split(".")[1]) if "." in s else 0 for s in raw[:20, 4])
    point = 10.0 ** (-digits)
    return dict(ts=ts, o=o, h=h, l=l, c=c, v=vol,
                cost=spread_pts * point, point=point, digits=digits, symbol=symbol)


# --------------------------------------------------------------- indicators
def sma(x, n):
    out = np.full(len(x), np.nan)
    cs = np.concatenate([[0.0], np.cumsum(x)])
    out[n - 1:] = (cs[n:] - cs[:-n]) / n
    return out


def rolling_std(x, n):
    out = np.full(len(x), np.nan)
    cs = np.concatenate([[0.0], np.cumsum(x)])
    cs2 = np.concatenate([[0.0], np.cumsum(x * x)])
    m = (cs[n:] - cs[:-n]) / n
    m2 = (cs2[n:] - cs2[:-n]) / n
    out[n - 1:] = np.sqrt(np.maximum(m2 - m * m, 0.0))
    return out


def atr(h, l, c, n=14):
    tr = np.maximum(h[1:] - l[1:], np.maximum(np.abs(h[1:] - c[:-1]), np.abs(l[1:] - c[:-1])))
    tr = np.concatenate([[h[0] - l[0]], tr])
    out = np.full(len(tr), np.nan)
    if len(tr) <= n:
        return out
    out[n - 1] = tr[:n].mean()
    for i in range(n, len(tr)):
        out[i] = (out[i - 1] * (n - 1) + tr[i]) / n
    return out


def rsi(c, n=2):
    d = np.diff(c, prepend=c[0])
    up = np.where(d > 0, d, 0.0)
    dn = np.where(d < 0, -d, 0.0)
    au = np.full(len(c), np.nan)
    ad = np.full(len(c), np.nan)
    if len(c) <= n:
        return au
    au[n] = up[1:n + 1].mean()
    ad[n] = dn[1:n + 1].mean()
    for i in range(n + 1, len(c)):
        au[i] = (au[i - 1] * (n - 1) + up[i]) / n
        ad[i] = (ad[i - 1] * (n - 1) + dn[i]) / n
    rs = au / np.where(ad > 0, ad, np.nan)
    out = 100 - 100 / (1 + rs)
    out[ad == 0] = 100.0
    return out


# ----------------------------------------------------------------- simulate
def simulate(d, entries, side=1, exit_ma=None, max_bars=10, stop_atr=3.0, atr_n=14):
    """Bar-close signals, NEXT-BAR-OPEN fills. Returns per-trade log.

    entries : bool array; entry decided on bar i, filled at open of i+1.
    exit_ma : if given, exit when close crosses back to this level; else time/stop.

    NO LOOKAHEAD: every exit test uses bar j's own high/low/close, and the stop
    is checked against the bar's extreme before its close, so an intrabar stop
    is honoured rather than assumed away.
    """
    o, h, l, c = d["o"], d["h"], d["l"], d["c"]
    a = atr(h, l, c, atr_n)
    cost = d["cost"]
    ts = d["ts"]
    trades = []
    i = 0
    n = len(c)
    while i < n - 2:
        if not entries[i] or not np.isfinite(a[i]) or a[i] <= 0:
            i += 1
            continue
        fill = i + 1
        entry = o[fill]
        if not np.isfinite(entry) or entry <= 0:
            i += 1
            continue
        stop = entry - side * stop_atr * a[i]
        exit_px, exit_j, reason = None, None, None
        for j in range(fill, min(fill + max_bars, n)):
            # stop first: within a bar we cannot know order, so assume the worse
            if side > 0 and l[j] <= stop:
                exit_px, exit_j, reason = stop, j, "stop"
                break
            if side < 0 and h[j] >= stop:
                exit_px, exit_j, reason = stop, j, "stop"
                break
            if exit_ma is not None and np.isfinite(exit_ma[j]):
                if (side > 0 and c[j] >= exit_ma[j]) or (side < 0 and c[j] <= exit_ma[j]):
                    exit_px, exit_j, reason = c[j], j, "target"
                    break
        if exit_px is None:
            exit_j = min(fill + max_bars - 1, n - 1)
            exit_px, reason = c[exit_j], "time"
        gross = side * (exit_px - entry)
        net = gross - cost[fill]                      # full spread once, round trip
        trades.append(dict(entry_i=fill, exit_i=exit_j, t=ts[fill],
                           ret=net / entry, gross=gross / entry, reason=reason,
                           bars=exit_j - fill + 1))
        i = exit_j + 1                                # no overlapping positions
    return trades


def metrics(trades, ts, risk_frac=0.01, stop_atr=3.0):
    """Equity curve at fixed fractional risk; drawdown on that curve.

    Each trade risks `risk_frac` of equity at the stop distance, so a trade's
    equity impact is risk_frac * (net return / stop distance) -- i.e. R
    multiples, which is how the prop limit is actually consumed.
    """
    if not trades:
        return None
    years = (ts[-1] - ts[0]) / np.timedelta64(365, "D")
    months = max(years * 12.0, 1e-9)
    rets = np.array([t["ret"] for t in trades])
    eq = [1.0]
    for t in trades:
        # R multiple: net move relative to the risked distance
        eq.append(eq[-1] * (1.0 + risk_frac * t["r"]))
    eq = np.array(eq)
    peak = np.maximum.accumulate(eq)
    dd = (peak - eq) / peak
    total = eq[-1] - 1.0
    maxdd = dd.max()
    return dict(n=len(trades), per_month=len(trades) / months,
                exp_bp=rets.mean() * 1e4, hit=float((rets > 0).mean()),
                total=total, maxdd=maxdd,
                ret_dd=(total / maxdd if maxdd > 1e-9 else np.nan),
                cagr=(eq[-1] ** (1 / max(years, 1e-9)) - 1.0),
                years=years)


def add_r(trades, d, stop_atr=3.0, atr_n=14):
    """Attach the R multiple to each trade (net return / risked distance)."""
    a = atr(d["h"], d["l"], d["c"], atr_n)
    for t in trades:
        i = t["entry_i"] - 1
        risk = stop_atr * a[i] / d["o"][t["entry_i"]]
        t["r"] = t["ret"] / risk if risk > 0 else 0.0
    return trades