"""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}")