Warrior_EA/research/test_sweep_detail.py

"""Does the stop-run edge behave the way its MECHANISM says it should?

The statistics already cleared a family-wise bar and a split-half. That is necessary and
not sufficient - a mined artifact can do both if the mining was thorough enough. What a
mined artifact CANNOT do is obey a mechanism it was never fitted to.

If the edge is really stop-cluster liquidity, then it must be concentrated where the stop
reservoir is deepest and where fresh forced orders arrive:
  - around SESSION OPENS, and the London/NY handover
  - NOT uniform around the clock
A uniform effect would be evidence AGAINST the mechanism even with the same p-value, and
would make the whole thing much likelier to be curve-fit. This is the test that can
disconfirm.

Also here: walk-forward by quarters rather than halves (a 2-way split can hide a single
lucky regime), and the other two instruments.

Broker time is UTC+2 (measured, see dukas.py notes) - every hour below is converted to UTC
so the session labels mean what they say.
"""
import numpy as np, sys, datetime as dt
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
from test_liquidity import load, atr_of, rolling_extreme, outcomes

BROKER_OFFSET_H = 2          # SQX conforms to the5ers: broker = UTC + 2


def build(sym, tf='H1', N=50, ov=0.5, R=3.0, H=48):
    a, I = load(sym, tf)
    g = lambda c: a[:, I[c]]
    o, h, l, c = g('open'), g('high'), g('low'), g('close')
    spb = g('spread_mean'); spb = np.concatenate([[spb[0]], spb[:-1]])
    atr = atr_of(h, l, c, 14); atr = np.concatenate([[atr[0]], atr[:-1]])
    n = len(c)
    ph = rolling_extreme(h, N, 'max'); pl = rolling_extreme(l, N, 'min')
    sh = (h > ph) & ((h - ph) <= ov * atr) & (c < ph) & np.isfinite(ph)
    sl = (l < pl) & ((pl - l) <= ov * atr) & (c > pl) & np.isfinite(pl)
    fire = np.nonzero(sh | sl)[0]
    fire = fire[(fire > N + 2) & (fire < n - H - 2)]
    d = np.where(sh[fire], -1, 1)
    e = fire + 1
    buf = 0.05 * atr[fire]
    stop = np.where(d < 0, h[fire] + buf, l[fire] - buf)
    ent = o[e]
    risk = np.abs(ent - stop) + spb[e]
    keep, busy = [], -1
    for k in range(len(e)):
        if risk[k] <= 0 or e[k] <= busy:
            continue
        keep.append(k); busy = e[k] + H
    keep = np.array(keep, int)
    targ = np.where(d < 0, ent - R * risk, ent + R * risk)
    w, to = outcomes(o, h, l, e[keep], d[keep],
                     np.where(d[keep] < 0, stop[keep] + spb[e[keep]],
                              stop[keep] - spb[e[keep]]),
                     targ[keep], H)
    t = a[:, I['time']][e[keep]]
    utc_h = np.array([(dt.datetime.fromtimestamp(x / 1000, dt.UTC).hour - BROKER_OFFSET_H) % 24
                      for x in t])
    dow = np.array([dt.datetime.fromtimestamp(x / 1000, dt.UTC).weekday() for x in t])
    return w, d[keep], utc_h, dow, t, R


def expR(w, R):
    return w.mean() * R - (1 - w.mean()) if len(w) else 0.0


def boot_ci(w, R, n=4000, seed=3):
    """Bootstrap CI on expected R. Trades are non-overlapping so a plain resample is fine."""
    if len(w) < 20:
        return (np.nan, np.nan)
    rng = np.random.default_rng(seed)
    idx = rng.integers(0, len(w), (n, len(w)))
    m = w[idx].mean(axis=1)
    e = m * R - (1 - m)
    return float(np.quantile(e, 0.05)), float(np.quantile(e, 0.95))


if __name__ == '__main__':
    SESSIONS = [('Sydney/Tokyo', lambda h: (h >= 0) & (h < 7)),
                ('London open', lambda h: (h >= 7) & (h < 10)),
                ('London', lambda h: (h >= 10) & (h < 12)),
                ('LDN/NY overlap', lambda h: (h >= 12) & (h < 16)),
                ('NY afternoon', lambda h: (h >= 16) & (h < 21)),
                ('Rollover/thin', lambda h: (h >= 21) | (h < 0))]

    for sym in ('EURUSD', 'USDJPY', 'XAUUSD', 'SP500'):
        try:
            w, d, uh, dow, t, R = build(sym)
        except Exception as ex:
            print(f"{sym}: {ex}")
            continue
        print(f"\n{'='*76}\n{sym} H1  N=50 ov=0.5 R=3  {len(w)} trades  "
              f"overall expR {expR(w,R):+.3f}\n{'='*76}")

        print("  --- by SESSION (UTC) - the mechanism test -------------------------")
        print(f"  {'session':<16}{'n':>6}{'win%':>8}{'expR':>9}{'90% CI':>20}")
        for name, f in SESSIONS:
            m = f(uh)
            if m.sum() < 30:
                continue
            lo, hi = boot_ci(w[m], R)
            print(f"  {name:<16}{int(m.sum()):>6}{100*w[m].mean():>8.2f}{expR(w[m],R):>+9.3f}"
                  f"   [{lo:+.3f}, {hi:+.3f}]{'  *' if lo > 0 else ''}")

        print("  --- WALK-FORWARD by quarter of the sample -------------------------")
        q = np.array_split(np.arange(len(w)), 4)
        line = "  "
        for i, ix in enumerate(q):
            line += f"Q{i+1} {expR(w[ix],R):+.3f} (n={len(ix)})   "
        print(line)
        pos = sum(1 for ix in q if expR(w[ix], R) > 0)
        print(f"  quarters positive: {pos}/4")

        print("  --- by DIRECTION --------------------------------------------------")
        for lbl, m in (('fade high sweep (short)', d < 0), ('fade low sweep (long)', d > 0)):
            if m.sum() < 30:
                continue
            lo, hi = boot_ci(w[m], R)
            print(f"  {lbl:<24}{int(m.sum()):>6}{100*w[m].mean():>8.2f}"
                  f"{expR(w[m],R):>+9.3f}   [{lo:+.3f}, {hi:+.3f}]")