Warrior_EA/research/fractal_nn.py

"""Can a clean NN predict pivots? The user's request, run on the bug-free offline stack.

PRE-REGISTERED DESIGN (written before any result was seen):

Target - "predict the pivot" in its tradeable form: at every H1 bar close, the sign of the
mid-price move from here to the NEXT confirmed Bill Williams fractal (strict 5-bar local
extreme, confirmed 2 bars after its bar). Fractals are the user's suggested high-frequency
swing markers (~10x more label events than ZigZag pivots); the next confirmation is the
nearest causally-knowable swing event, so this asks exactly "which side of the current price
does the next swing marker land on" - direction, at swing scale, with an adaptive horizon.

Features - kit.py's scale-free causal set (returns/ATR, donchian position, SMA distances,
wick/body shape, vol ratio, RSI, session clock). Computed on closed bars only.

Training - the same numpy MLP + per-symbol standardization + pooling that just demonstrated
it CAN extract real conditional structure (+2.6pp on XAUUSD meta-labels): 4 symbols
(EURUSD/USDJPY/XAUUSD/SP500 M1-book H1 mids), chronological 55/15/30 with 50-bar purges,
threshold fitted on CALIB only (both sides: long if p>=thr, short if p<=1-thr, 25% coverage
floor, objective = coverage x mean net pnl), TEST touched once.

Readout - real ask/bid fills from the validated book at the first M1 of the next bar; exit
at the first M1 after the next fractal confirmation. GROSS (mid-mid, zero cost) printed next
to NET so "is it only the spread?" is answered by subtraction. Gate: test mean NET pnl/trade
> 0 at 2 sigma, per symbol. Threshold dose-response curve printed for the stability read.

Context this experiment lives in: exact-pivot labels were the EA's original target
(replaced b4a704d); model-free MI found no per-feature directional info at any lag 0-20;
direction-at-fixed-horizon is closed. What has NEVER been run is a pivot-form target on the
offline stack, pooled cross-sectionally. This is that test - the last clean look at
direction, at the scale the user believes in.
"""
import numpy as np
import sys

sys.stdout.reconfigure(encoding="utf-8", errors="replace")
import book
import fills
import kit
from meta_pool import MLP, _standardize

SYMS = ("EURUSD", "USDJPY", "XAUUSD", "SP500")
PURGE_BARS = 50
MIN_COVERAGE = 0.25
SEED = 20260814


def fractal_events(h, l):
    """Bar indices where a strict 5-bar fractal CONFIRMS (extreme bar + 2)."""
    n = len(h)
    ev = []
    for j in range(2, n - 2):
        up = h[j] > h[j - 1] and h[j] > h[j - 2] and h[j] > h[j + 1] and h[j] > h[j + 2]
        dn = l[j] < l[j - 1] and l[j] < l[j - 2] and l[j] < l[j + 1] and l[j] < l[j + 2]
        if up or dn:
            ev.append(j + 2)
    return np.unique(np.array(ev, np.int64))


def build(sym):
    bk = fills.Book(sym)
    f = book.frame(sym, "H1", bk)
    X, names, atr = kit.features(f.t / 1000.0, f.o, f.h, f.l, f.c, f.v)  # book t is ms
    ev = fractal_events(f.h, f.l)
    # for every bar t, the next confirmation strictly after t
    nxt = np.searchsorted(ev, np.arange(f.n), "right")
    ok = (nxt < len(ev)) & (np.arange(f.n) >= 300)
    t_idx = np.flatnonzero(ok)
    e_idx = ev[nxt[t_idx]]
    ok2 = e_idx + 1 < f.n
    t_idx, e_idx = t_idx[ok2], e_idx[ok2]
    y = (f.c[e_idx] > f.c[t_idx]).astype(np.int64)     # up move to next swing marker
    # real-fill price paths: enter first M1 of t+1, exit first M1 of e+1
    ein = f.i0[t_idx + 1]
    eout = f.i0[e_idx + 1]
    ok3 = eout > ein
    t_idx, e_idx, y, ein, eout = t_idx[ok3], e_idx[ok3], y[ok3], ein[ok3], eout[ok3]
    mid_in = 0.5 * (bk.ao[ein] + bk.bo[ein])
    mid_out = 0.5 * (bk.ao[eout] + bk.bo[eout])
    pnl_l_gross = mid_out - mid_in
    pnl_l_net = bk.bo[eout] - bk.ao[ein]
    pnl_s_net = bk.bo[ein] - bk.ao[eout]
    atr_t = atr[t_idx]
    return {"sym": sym, "t": t_idx, "y": y, "X": X[t_idx].astype(np.float32),
            "gl": pnl_l_gross, "nl": pnl_l_net, "ns": pnl_s_net, "atr": atr_t,
            "hold": e_idx - t_idx, "n_bars": f.n}


def splits(d):
    n = len(d["t"])
    i1 = int(n * 0.55)
    i2 = int(n * 0.70)
    tr = np.arange(n) < i1
    ca = (np.arange(n) >= i1) & (np.arange(n) < i2)
    te = np.arange(n) >= i2
    # purge: drop calib/test rows whose bar index is within PURGE_BARS of the previous slice
    ca &= d["t"] > d["t"][i1 - 1] + PURGE_BARS
    te &= d["t"] > d["t"][i2 - 1] + PURGE_BARS
    return tr, ca, te


def main():
    data = [build(s) for s in SYMS]
    for d in data:
        print(f"{d['sym']}: {len(d['y'])} samples, base up {100.0 * d['y'].mean():.1f}%, "
              f"median hold {np.median(d['hold']):.0f} bars")
    # pooled training with per-symbol normalization
    Xs, ys, norms = [], [], {}
    for d in data:
        tr, _, _ = splits(d)
        mu = d["X"][tr].mean(axis=0, dtype=np.float64)
        sd = d["X"][tr].std(axis=0, dtype=np.float64) + 1e-8
        norms[d["sym"]] = (mu, sd)
        Xs.append(_standardize(d["X"][tr], mu, sd))
        ys.append(d["y"][tr])
    X = np.concatenate(Xs)
    y = np.concatenate(ys)
    print(f"\n[pooled] training on {len(X)} rows...")
    net = MLP(X.shape[1], h1=64, h2=32, seed=SEED).fit(X, y, epochs=30, l2=1e-3, verbose=True)
    for d in data:
        tr, ca, te = splits(d)
        mu, sd = norms[d["sym"]]
        p = net.p_win(_standardize(d["X"], mu, sd))
        # fit threshold on CALIB: coverage x mean net pnl, 25% floor
        best = (-np.inf, 0.5)
        for thr in np.linspace(0.5, 0.9, 41):
            lng = ca & (p >= thr)
            sht = ca & (p <= 1.0 - thr)
            cnt = lng.sum() + sht.sum()
            cov = cnt / max(ca.sum(), 1)
            if cov < MIN_COVERAGE or cnt == 0:
                continue
            mean_net = np.concatenate([d["nl"][lng], d["ns"][sht]]).mean()
            score = cov * mean_net
            if score > best[0]:
                best = (score, thr)
        thr = best[1]
        lng = te & (p >= thr)
        sht = te & (p <= 1.0 - thr)
        netpnl = np.concatenate([d["nl"][lng], d["ns"][sht]])
        gross = np.concatenate([d["gl"][lng], -d["gl"][sht]])
        n = len(netpnl)
        if n == 0:
            print(f"\n{d['sym']}: fitted thr {thr:.2f} -> no trades on test")
            continue
        se = netpnl.std() / np.sqrt(n)
        cov = 100.0 * n / max(te.sum(), 1)
        verdict = "DEPLOYABLE" if netpnl.mean() > 2.0 * se else ""
        print(f"\n{d['sym']}: fitted thr {thr:.2f} | test trades {n} ({cov:.1f}%) | "
              f"GROSS {gross.mean():+.3f} pts | NET {netpnl.mean():+.3f} pts "
              f"(2 sigma = {2 * se:.3f}) {verdict}")
        print(f"  {'thr':>5} | {'cal n':>6} {'cal net':>8} | {'test n':>6} {'test net':>9} {'test gross':>10}")
        for tt in (0.52, 0.55, 0.60, 0.65, 0.70):
            cl = ca & (p >= tt); cs = ca & (p <= 1 - tt)
            tl = te & (p >= tt); ts = te & (p <= 1 - tt)
            cn = np.concatenate([d["nl"][cl], d["ns"][cs]])
            tn = np.concatenate([d["nl"][tl], d["ns"][ts]])
            tg = np.concatenate([d["gl"][tl], -d["gl"][ts]])
            if len(tn) < 100:
                break
            print(f"  {tt:5.2f} | {len(cn):6d} {cn.mean():+8.3f} | {len(tn):6d} {tn.mean():+9.3f} {tg.mean():+10.3f}")


if __name__ == "__main__":
    main()