Warrior_EA/research/test_liquidity.py

"""The stop-run / liquidity-sweep hypothesis, tested as a COMPLETE trade.

Every earlier test in this project asked "does X predict direction" with barriers bolted on
afterwards at a fixed ATR. That is not how a trade is specified. Here entry, stop and target
come from ONE structure, which is also what makes the hypothesis falsifiable:

  Retail stops sit in a KNOWN place - just beyond the recent swing high/low. Carol Osler's
  work on currency order flow is the published version of this: stop-loss orders cluster,
  and their execution cascades price; take-profit orders cluster too and reverse it. So the
  claim is not "a pattern repeats" but "there is a reservoir of forced orders at a location
  we can compute in advance".

  The setup, therefore:
    - price takes out the N-bar extreme (the stops fire, price spikes)
    - the break is MARGINAL, not a real breakout (overshoot <= max_over ATR)
    - price closes back INSIDE the range (the spike was absorbed, not continuation)
    - enter the OPPOSITE way on the next open
    - stop goes just beyond the sweep extreme - i.e. where the liquidity actually was, not
      at an arbitrary ATR multiple. If the level does not hold, the premise is wrong and the
      trade should die immediately, which is what makes the stop tight and the R large.
    - target is a multiple of that stop distance

WHAT WOULD MAKE THIS FAKE, and is therefore controlled for:
  - Selection: sweeps happen more in volatile regimes. The null permutes DIRECTION across
    the same firing bars, so regime is held fixed and only the directional claim is tested.
  - Cost: per-bar spread charged on entry and on both barriers.
  - Multiple testing: many (N, overshoot, R) combinations, so a family-wise max-statistic
    bar over the whole grid, and split-half on anything that clears it.
  - The obvious trap: requiring "closes back inside" uses bar i's CLOSE, so entry must be at
    bar i+1's open. Using bar i's close as the entry price would be lookahead.
"""
import numpy as np, sys, os, datetime as dt
sys.stdout.reconfigure(encoding='utf-8', errors='replace')

BARS = 'c:/Users/admin/Documents/Workspaces/Market Data/bars/'


def load(sym, tf):
    z = np.load(f"{BARS}{sym}_{tf}_ticks.npz", allow_pickle=True)
    a = z['bars']
    cols = [str(c) for c in z['columns']]
    return a, {c: k for k, c in enumerate(cols)}


def atr_of(h, l, c, n=14):
    pc = np.roll(c, 1); pc[0] = c[0]
    tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc)))
    out = np.convolve(tr, np.ones(n) / n, mode='full')[:len(tr)]
    out[:n] = tr[:n].mean()
    return out


def rolling_extreme(x, N, kind='max'):
    """Extreme of the N bars ENDING AT i-1 (never includes bar i itself)."""
    n = len(x)
    out = np.full(n, np.nan)
    f = np.maximum if kind == 'max' else np.minimum
    #--- simple O(n*log) via strides is overkill; N is small and n <= 1.7M
    from numpy.lib.stride_tricks import sliding_window_view
    if n > N:
        w = sliding_window_view(x, N)
        agg = w.max(axis=1) if kind == 'max' else w.min(axis=1)
        out[N:] = agg[:-1]
    return out


def outcomes(o, h, l, entry_i, direction, stop_px, targ_px, H):
    """Walk each trade forward bar by bar. Trades have INDIVIDUAL stop/target prices here,
    so the vectorised block scan used elsewhere does not apply. Stop is checked before
    target within a bar (a bar spanning both books the loss)."""
    win = np.zeros(len(entry_i), bool)
    tout = np.zeros(len(entry_i), bool)
    for k in range(len(entry_i)):
        e = entry_i[k]
        hi = min(e + H, len(o) - 1)
        d = direction[k]
        w = False; done = False
        for j in range(e, hi + 1):
            if d > 0:
                if l[j] <= stop_px[k]:
                    done = True; break
                if h[j] >= targ_px[k]:
                    w = True; done = True; break
            else:
                if h[j] >= stop_px[k]:
                    done = True; break
                if l[j] <= targ_px[k]:
                    w = True; done = True; break
        win[k] = w
        tout[k] = not done
    return win, tout


def run(sym, tf='H1', Ns=(20, 50), overs=(0.25, 0.5), Rs=(1.0, 2.0, 3.0),
        H=48, nperm=2000, seed=5, verbose=True):
    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)
    t = g('time')
    hrs = np.array([dt.datetime.fromtimestamp(x / 1000, dt.UTC).hour for x in t])
    rng = np.random.default_rng(seed)
    rows, acc = [], []

    for N in Ns:
        ph = rolling_extreme(h, N, 'max')
        pl = rolling_extreme(l, N, 'min')
        for ov in overs:
            #--- SHORT setup: swept the prior high marginally, closed back below it
            sw_hi = (h > ph) & ((h - ph) <= ov * atr) & (c < ph) & np.isfinite(ph)
            #--- LONG setup: swept the prior low marginally, closed back above it
            sw_lo = (l < pl) & ((pl - l) <= ov * atr) & (c > pl) & np.isfinite(pl)
            fire = np.nonzero(sw_hi | sw_lo)[0]
            fire = fire[(fire > N + 2) & (fire < n - H - 2)]
            if len(fire) < 150:
                continue
            d = np.where(sw_hi[fire], -1, 1)
            e = fire + 1
            #--- stop just beyond the sweep extreme: that IS the level being tested
            buf = 0.05 * atr[fire]
            stop = np.where(d < 0, h[fire] + buf, l[fire] - buf)
            entry = o[e]
            risk = np.abs(entry - stop) + spb[e]
            ok = risk > 0
            for R in Rs:
                targ = np.where(d < 0, entry - R * risk, entry + R * risk)
                #--- sequential: no overlapping trades
                keep, busy = [], -1
                for k in range(len(e)):
                    if not ok[k] or e[k] <= busy:
                        continue
                    keep.append(k); busy = e[k] + H
                keep = np.array(keep, int)
                if len(keep) < 100:
                    continue
                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)
                nT = len(keep); wr = w.mean()
                expR = wr * R - (1 - wr)
                be = 1.0 / (1.0 + R)
                #--- null: same bars, permuted directions (regime held fixed)
                wl, ws = None, None
                dd = d[keep]
                #--- recompute both-direction outcomes once for the null
                stop_L = np.where(True, l[fire[keep]] - buf[keep], 0) - spb[e[keep]]
                stop_S = h[fire[keep]] + buf[keep] + spb[e[keep]]
                entL = o[e[keep]]; riskL = np.abs(entL - stop_L) + spb[e[keep]]
                riskS = np.abs(entL - stop_S) + spb[e[keep]]
                wL, _ = outcomes(o, h, l, e[keep], np.ones(len(keep), int),
                                 stop_L, entL + R * riskL, H)
                wS, _ = outcomes(o, h, l, e[keep], -np.ones(len(keep), int),
                                 stop_S, entL - R * riskS, H)
                long_ = dd > 0
                pw = np.empty(nperm)
                B = max(1, 2000000 // max(nT, 1))
                for b0 in range(0, nperm, B):
                    b1 = min(b0 + B, nperm)
                    pl_ = rng.permuted(np.broadcast_to(long_, (b1 - b0, nT)), axis=1)
                    pw[b0:b1] = np.where(pl_, wL, wS).mean(axis=1)
                nmu, nsd = pw.mean(), max(pw.std(ddof=1), 1e-12)
                z = (wr - nmu) / nsd
                rows.append((N, ov, R, nT, 100 * wr, 100 * nmu, z, expR, 100 * to.mean(),
                             keep, w, dd))
                acc.append(np.abs((pw - nmu) / nsd))

    if not rows:
        print(f"{sym} {tf}: no setup fired often enough")
        return []
    crit = float(np.quantile(np.maximum.reduce(acc), 0.95))
    if verbose:
        print(f"\n=== {sym} {tf} liquidity sweep  H={H}  {len(rows)} configs  "
              f"family-wise |z| > {crit:.2f} ===")
        print(f"{'N':>4}{'over':>6}{'R':>5}{'trades':>8}{'win%':>7}{'null%':>7}{'z':>7}"
              f"{'expR':>8}{'t/o%':>7}")
        for r in sorted(rows, key=lambda x: -x[6])[:10]:
            star = ' *' if abs(r[6]) > crit else ''
            print(f"{r[0]:>4}{r[1]:>6.2f}{r[2]:>5.1f}{r[3]:>8}{r[4]:>7.2f}{r[5]:>7.2f}"
                  f"{r[6]:>+7.2f}{r[7]:>+8.3f}{r[8]:>7.1f}{star}")
    return [(r, crit) for r in rows if abs(r[6]) > crit]


if __name__ == '__main__':
    syms = [a for a in sys.argv[1:] if not a.startswith('-')] or \
           ['EURUSD', 'USDJPY', 'XAUUSD', 'SP500']
    tf = 'H1'
    for a_ in sys.argv[1:]:
        if a_.startswith('--tf='):
            tf = a_.split('=', 1)[1]
    survivors = []
    for s in syms:
        if os.path.exists(f"{BARS}{s}_{tf}_ticks.npz"):
            survivors += [(s,) + x for x in run(s, tf)]
    print(f"\n{'='*70}\nconfigs clearing their family-wise bar: {len(survivors)}")
    for s in survivors:
        r = s[1]
        print(f"  {s[0]:>7} N={r[0]} over={r[1]} R={r[2]}  {r[3]} trades  "
              f"win {r[4]:.2f}% vs null {r[5]:.2f}%  z {r[6]:+.2f}  expR {r[7]:+.3f}")