Warrior_EA/research/test_classic.py

"""Do the shipped classic patterns carry a directional edge?

Pre-registration note: the 26 conditions are NOT invented here. They are the models the
EA already ships in Signals/, written long before this test existed, with their shipped
constructor weights. Nothing about them is fitted to this data, so there is no in-sample
/ out-of-sample distinction to draw - every trade is an honest out-of-sample trade and
the whole history can be used. That is the one real advantage of testing a rule instead
of a model.

The null: by the gambler's-ruin identity, the probability of touching +k*ATR before
-m*ATR on a driftless random walk is m/(m+k) - which is exactly the break-even win rate
for a k:m payoff. So chance == break-even at every geometry, and "is this pattern better
than a coin" and "does this pattern make money" are the same question. Spread is charged
inside the barrier, which pushes the honest bar slightly above break-even.

Multiple comparisons: 15 actionable patterns x geometries. Controlled with a sign-flip
null (keep each pattern's firing TIMES, randomise its DIRECTION) and the distribution of
the MAX |z| over the whole family - the family-wise bar, not the per-test one.
"""
import numpy as np, sys, time
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
from classic import build_patterns, module_votes, combined_direction, MODULE_SLICES
from kit import load_rates, atr

TF = {5: 'M5', 15: 'M15', 16385: 'H1', 16388: 'H4', 16408: 'D1'}


def barrier_outcomes(o, h, l, a, sl, tp, H, spread):
    """For an entry at the OPEN of bar i, in both directions: did TP land before SL?

    Returns winL, resL, winS, resS  (res = resolved, i.e. not a timeout).
    A bar that spans both barriers scores as the LOSS (strict tp < sl), as in kit.py.
    """
    n = len(o)
    INF = np.iinfo(np.int32).max
    winL = np.zeros(n, bool); resL = np.zeros(n, bool)
    winS = np.zeros(n, bool); resS = np.zeros(n, bool)
    risk, rew = sl * a, tp * a
    lTp, lSl = o + rew + spread, o - risk + spread
    sTp, sSl = o - rew - spread, o + risk - spread
    CH = max(200000 // max(H, 1), 1)
    for s in range(0, n, CH):
        e2 = min(s + CH, n - H)
        if e2 <= s:
            break
        wi = np.arange(0, H)[None, :] + np.arange(s, e2)[:, None]
        wh, wl = h[wi], l[wi]

        def first(mask):
            any_ = mask.any(axis=1)
            return np.where(any_, mask.argmax(axis=1), INF)
        lsl = first(wl <= lSl[s:e2, None]); ltp = first(wh >= lTp[s:e2, None])
        ssl = first(wh >= sSl[s:e2, None]); stp = first(wl <= sTp[s:e2, None])
        winL[s:e2] = ltp < lsl; resL[s:e2] = np.minimum(ltp, lsl) < INF
        winS[s:e2] = stp < ssl; resS[s:e2] = np.minimum(stp, ssl) < INF
    return winL, resL, winS, resS


def sequential(fire_bars, dirs, winL, winS, H, n):
    """Sequential NON-OVERLAPPING trades: while a position is open, later signals are
    ignored. This is the only simulation whose confidence interval means anything,
    because it is the only one where the trades are independent."""
    out_i, out_d, out_w = [], [], []
    busy_until = -1
    for j, d in zip(fire_bars, dirs):
        if j <= busy_until or j + 1 + H >= n:
            continue
        e = j + 1                      # enter at the OPEN of the next bar
        w = winL[e] if d > 0 else winS[e]
        out_i.append(e); out_d.append(d); out_w.append(bool(w))
        busy_until = e + H
    return np.array(out_i, int), np.array(out_d, int), np.array(out_w, bool)


_CACHE = {}


def prepare(sym, tf):
    """Load + build patterns once per symbol; the divergence bit-map walk is the slow part."""
    key = (sym, tf)
    if key not in _CACHE:
        t, o, h, l, c, v, spr = load_rates(sym, tf)
        a = atr(h, l, c, 14)
        tick = np.nanmin(np.abs(np.diff(np.unique(np.round(c, 8)))))
        sp = np.nanmedian(spr) * tick
        if not np.isfinite(sp):
            sp = 0.0
        fireL, fireS, names, weights = build_patterns(o, h, l, c)
        _CACHE[key] = (o, h, l, c, a, sp, fireL, fireS, names, weights)
    return _CACHE[key]


def run(sym, tf, sl_m, tp_m, H, nperm=2000, seed=0, quiet=False):
    o, h, l, c, a, sp, fireL, fireS, names, weights = prepare(sym, tf)
    n = len(c)
    # ATR known at the signal bar; entry one bar later, so shift the ATR forward by one
    a_sig = np.concatenate([[a[0]], a[:-1]])
    winL, resL, winS, resS = barrier_outcomes(o, h, l, a_sig, sl_m, tp_m, H, sp)

    be = sl_m / (sl_m + tp_m)           # break-even == chance
    rows = []
    rng = np.random.default_rng(seed)
    perm_max = np.zeros(nperm)

    actionable = [k for k in range(len(names)) if weights[k] > 10]
    per_pattern_perm = {}

    for k in actionable:
        fb = np.nonzero(fireL[:, k] | fireS[:, k])[0]
        # a bar where both sides fire is a genuine flat vote in the EA - skip it
        both = fireL[fb, k] & fireS[fb, k]
        fb = fb[~both]
        if len(fb) == 0:
            continue
        d = np.where(fireL[fb, k], 1, -1)
        ti, td, tw = sequential(fb, d, winL, winS, H, n)
        nT = len(ti)
        if nT < 30:
            continue
        wr = tw.mean()
        se = np.sqrt(be * (1 - be) / nT)
        z = (wr - be) / se
        exp_R = wr * tp_m - (1 - wr) * sl_m
        rows.append((names[k], int(weights[k]), len(fb), nT, 100 * wr, 100 * (wr - be), z, exp_R))
        # sign-flip null for this pattern: same firing bars, randomised direction
        wl_at, ws_at = winL[ti], winS[ti]
        fl = rng.random((nperm, nT)) < 0.5
        pw = np.where(fl, wl_at[None, :], ws_at[None, :]).mean(axis=1)
        pz = (pw - be) / se
        per_pattern_perm[names[k]] = pz
        perm_max = np.maximum(perm_max, np.abs(pz))

    if not rows:
        return None
    rows.sort(key=lambda r: -r[6])
    crit = np.quantile(perm_max, 0.95)
    if not quiet:
        print(f"\n=== {sym} {TF.get(tf,tf)}  SL{sl_m}:TP{tp_m}  H={H}  "
              f"bars={n}  spread={sp:.5f} ({sp/np.nanmedian(a):.3f} ATR)  "
              f"break-even={100*be:.2f}% ===")
        print(f"{'pattern':<14}{'w':>4}{'fires':>8}{'trades':>8}{'win%':>8}"
              f"{'edge pp':>9}{'z':>7}{'exp R':>8}")
        for r in rows:
            star = ' *' if abs(r[6]) > crit else ''
            print(f"{r[0]:<14}{r[1]:>4}{r[2]:>8}{r[3]:>8}{r[4]:>8.2f}"
                  f"{r[5]:>+9.2f}{r[6]:>+7.2f}{r[7]:>+8.3f}{star}")
        print(f"  family-wise 5% bar (max|z| over {len(rows)} patterns, {nperm} sign-flips): "
              f"|z| > {crit:.2f}")
    return rows, crit


if __name__ == '__main__':
    t0 = time.time()
    GEOM = [(2, 3, 96), (2, 6, 192), (1, 2, 64)]
    for sym, tf in [('EURUSD', 16385), ('USDJPY', 16385), ('XAUUSD', 16385), ('SP500', 16385)]:
        for (s, p, H) in GEOM:
            try:
                run(sym, tf, s, p, H)
            except Exception as ex:
                print(f"{sym} {tf} {s}:{p}  FAILED  {ex}")
    print(f"\ntotal {time.time()-t0:.0f}s")