"""Wyckoff as the books actually teach it: the shakeout PLUS the context that qualifies it. The previous test fired on any pierce of a range edge. That is not the method. Book 2 §2.3 is explicit that the shakeout is the third of four cumulative traces, and that you read the structure's own history before deciding what a pierce means: TRACE 1 Phase A test location. "divide the vertical distance of the structure in two: if the Secondary Test develops in the lower part it indicates weakness; if it ends at the top, less resistance." TRACE 2 Phase B test + REACTION. "a test at the upper part denotes strength, at the lower part weakness"; and "an inability to visit the opposite extreme alerts us to a STRUCTURAL FAILURE, which adds strength in the opposite direction." TRACE 3 Phase C shakeout. "the dominant event... the shakeout alone should be valid enough to bias us in favour of its direction." TRACE 4 Phase D effort/result. "wide ranges and high volume in favour of the movement that follows the shakeout (SOS/SOW bar)." §7.1 context: in a range, trade the extremes; in a trend, trade only with it. THE DESIGN, AND WHY IT IS A DOSE-RESPONSE CURVE ----------------------------------------------- Conditioning on several agreeing traces shrinks the sample and multiplies the ways to slice it, which is precisely how a filter gets mined into looking profitable. So the test is NOT "find the combination that works". It is a single pre-specified prediction the books make and a mined artifact does not: if context is real, expR must RISE MONOTONICALLY with the number of agreeing traces. One number decides it - the slope across buckets - with no threshold to tune, no best cell to pick, and no way to improve it by looking. A jagged profile whose top bucket happens to be positive is exactly what mining produces and it fails this test. Entries remain MARKET ORDERS at the next bar's open, so the fill artifact that invalidated an earlier round cannot recur. Benchmark is expR = 0 exactly. """ import numpy as np, sys sys.stdout.reconfigure(encoding='utf-8', errors='replace') from test_retail import load_bars, race_px from test_cause_effect import atr_of, find_ranges SYMS = ('EURUSD', 'USDJPY', 'XAUUSD', 'SP500') def events(sym, tf, theta=0.60, ov=0.75, H=200, path_tf='M5', htf=200): a1, I1 = load_bars(sym, tf) g = lambda k: a1[:, I1[k]] o, h, l, c = g('open'), g('high'), g('low'), g('close') vol = g('ticks'); spm = g('spread_mean') t1 = a1[:, I1['time']].astype(np.int64) n = len(c) atr = atr_of(h, l, c, 14); atr = np.concatenate([[atr[0]], atr[:-1]]) L, hi_, lo_ = find_ranges(h, l, c, atr, theta=theta) a2, I2 = load_bars(sym, path_tf) ph, pl, pc = a2[:, I2['high']], a2[:, I2['low']], a2[:, I2['close']] pmap = np.searchsorted(a2[:, I2['time']], t1) step = 12 if tf == 'H1' else (3 if tf == 'M15' else 1) HH = H * step ok = (L > 0) & np.isfinite(hi_) & np.isfinite(lo_) sp_ = ok & (l < lo_) & ((lo_ - l) <= ov * atr) & (c > lo_) # spring -> long up_ = ok & (h > hi_) & ((h - hi_) <= ov * atr) & (c < hi_) # upthrust -> short idx = np.nonzero(sp_ | up_)[0] idx = idx[(idx > max(300, htf + 5)) & (idx < n - 5)] if not len(idx): return None d = np.where(sp_[idx], 1, -1) rows = [] for q in range(len(idx)): i = int(idx[q]); dd = int(d[q]); Lq = int(L[i]) s = i - Lq # the range window is [s, i-1] if s < 1: continue top, bot = hi_[i], lo_[i] mid = 0.5 * (top + bot) th = max(Lq // 3, 2) segA = slice(s, s + th) # Phase A third segB = slice(s + th, s + 2 * th) # Phase B third segC = slice(s + 2 * th, i) # the run-up to the shakeout #--- TRACE 1: did the early test reach the upper or the lower half? upA = h[segA].max() - mid dnA = mid - l[segA].min() t1_ = 1 if upA > dnA else -1 #--- TRACE 2: same for the middle third upB = h[segB].max() - mid dnB = mid - l[segB].min() t2_ = 1 if upB > dnB else -1 #--- TRACE 2b: STRUCTURAL FAILURE - after the Phase B test, did price fail to #--- reach the opposite extreme? Failure adds strength AGAINST the tested side. t3_ = 0 if len(c[segC]): if t2_ > 0: # tested the top; did it reach the low? t3_ = 1 if l[segC].min() > bot + 0.25 * (top - bot) else -1 else: # tested the low; did it reach the top? t3_ = -1 if h[segC].max() < top - 0.25 * (top - bot) else 1 #--- TRACE 4: effort/result on the shakeout bar itself - closes back decisively in #--- the shakeout's direction, on elevated volume. Known at the signal bar. rr = max(h[i] - l[i], 1e-12) clspos = (c[i] - l[i]) / rr if dd > 0 else (h[i] - c[i]) / rr vavg = vol[s:i].mean() if i > s else vol[i] t4_ = 1 if (clspos > 0.6 and vol[i] > 1.2 * max(vavg, 1e-12)) else -1 #--- §7.1 CONTEXT: is the larger move in the shakeout's favour (re-accumulation)? t5_ = 1 if np.sign(c[i] - c[i - htf]) == dd else -1 #--- traces are oriented so +1 = agrees with the shakeout's implied direction agree = sum(1 for x in (t1_ * dd, t2_ * dd, t3_ * dd, t4_, t5_) if x > 0) e = i + 1 if e >= n - 1: continue pi = int(pmap[e]) if pi + HH >= len(ph): continue ent = o[e] ext = l[i] if dd > 0 else h[i] stop = ext - dd * 0.10 * atr[i] targ = top if dd > 0 else bot risk = abs(ent - stop); rew = abs(targ - ent) if risk <= 2 * spm[e] or rew < 0.25 * risk: continue rows.append((e, pi, dd, ent, stop, targ, risk, rew, spm[e], agree, Lq, t1[e])) if len(rows) < 100: return None #--- non-overlapping, so buckets are not padded with shared price paths rows.sort(key=lambda r: r[0]) keep, busy = [], -1 for r in rows: if r[0] <= busy: continue keep.append(r); busy = r[0] + r[10] if len(keep) < 100: return None A = lambda k: np.array([r[k] for r in keep]) E, P, D, EN, ST, TG, RK, RW, SP, AG = (A(k) for k in range(10)) r = race_px(ph, pl, P, D, ST, TG, HH) R = np.where(r > 0, RW / RK, np.where(r < 0, -1.0, 0.0)) un = r == 0 if un.any(): qq = np.minimum(P[un] + HH, len(pc) - 1) R[un] = (pc[qq] - EN[un]) * D[un] / RK[un] R = R - SP / RK return R, AG, A(11) def trend_t(R, AG): """Slope of expR against the confluence count, and its t. This is the whole test.""" x = AG.astype(float) if x.std() < 1e-9: return 0.0, 0.0 X = np.column_stack([np.ones(len(x)), x]) beta, *_ = np.linalg.lstsq(X, R, rcond=None) resid = R - X @ beta s2 = resid @ resid / max(len(x) - 2, 1) se = np.sqrt(s2 * np.linalg.inv(X.T @ X)[1, 1]) return float(beta[1]), float(beta[1] / max(se, 1e-12)) if __name__ == '__main__': syms = [s for s in sys.argv[1:] if s in SYMS] or list(SYMS) print("=== WYCKOFF WITH CONTEXT: does expR rise with the number of agreeing traces? ===") print(" traces: PhaseA test / PhaseB test / structural failure / effort-result / HTF") print(" the books predict a MONOTONE rise. A jagged profile with a good top bucket") print(" is what mining produces.\n") print(f" {'symbol':>7}{'tf':>5}{'n':>6} " + "".join(f"{k:>9}" for k in range(6)) + f"{'slope':>9}{'t':>7}") pool_R, pool_A = [], [] for tf in ('M15', 'H1'): for s in syms: out = events(s, tf) if out is None: print(f" {s:>7}{tf:>5} - too few") continue R, AG, _ = out cells = [] for k in range(6): m = AG == k cells.append(f"{R[m].mean():+6.2f}({int(m.sum()):>3})" if m.sum() >= 25 else f"{'-':>9}") sl, tt = trend_t(R, AG) print(f" {s:>7}{tf:>5}{len(R):>6} " + "".join(f"{x:>9}" for x in cells) + f"{sl:>+9.3f}{tt:>+7.2f}") pool_R.append(R); pool_A.append(AG) if pool_R: R = np.concatenate(pool_R); AG = np.concatenate(pool_A) sl, tt = trend_t(R, AG) print(f"\n POOLED n={len(R):,}") for k in range(6): m = AG == k if m.sum() >= 25: se = R[m].std(ddof=1) / np.sqrt(m.sum()) print(f" {k} traces agree: n={int(m.sum()):>5} expR {R[m].mean():+7.3f}" f" +/- {se:.3f}") print(f" slope per extra agreeing trace: {sl:+.4f} R t {tt:+.2f}")