Warrior_EA/research/test_spring.py

"""The Spring and the Upthrust - the trade both Wyckoff books are actually built around.

Everything tested so far took a piece of the method (levels, volume nodes, range projection)
in isolation. This tests the ENTRY the books teach, whole, with its own structural stop and
its own structural target:

  book 1 ch.18  "the shakeout is the key event that all Wyckoff operators wait for"
  book 2 5.7.1  range boundaries are low-volume nodes; the shakeout pierces one and fails

  SPRING     inside a confirmed trading range, price pierces the RANGE LOW by a small
             amount and closes back inside. Demand absorbed the supply. Go long.
  UPTHRUST   the mirror at the range high. Go short.
  stop       just beyond the shakeout extreme - if that level fails, the premise is wrong
  target     the OPPOSITE side of the range (structural, per the books - not an R multiple)

WHY THIS IS NOT THE SWEEP TEST THAT ALREADY FAILED
--------------------------------------------------
[[project_stop_run_liquidity_edge]] faded pierces of a rolling N-bar extreme. Three things
differ here and each is a book requirement that test ignored:
  1. the extreme must bound a CONFIRMED CONSOLIDATION (compression-qualified), not just be
     the highest of the last N bars - most N-bar extremes are trend, not range
  2. the target is the range's far side, so reward scales with the structure that produced it
  3. VOLUME confirmation, which no previous test in this project could apply: the books
     require effort/result divergence at the shakeout and, crucially, a LOW-VOLUME TEST
     afterwards. Tick counts and true volume-at-price are available here.

FILL MODEL - the bug from the last round, designed out
------------------------------------------------------
Every entry is a MARKET ORDER AT THE NEXT BAR'S OPEN. No stop-entry, no limit, no level to
cross. The fill price is that open and the outcome race starts at that same bar, so there is
no window in which price is on the wrong side of the entry. The previous round's edge was
entirely an artifact of entering at a level while measuring from the bar open; it cannot
recur in this form.

NULL - analytic, no simulation
------------------------------
Entry, stop and target are all fixed at entry, so a driftless market gives
P(target first) = risk/(risk+reward) and expected R = 0 EXACTLY, at every geometry. Any
positive expR after cost is the finding. Bars spanning both barriers book the loss; spread
is charged once, round trip.
"""
import numpy as np, sys, datetime as dt
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 springs(sym, tf, theta=0.60, ov=0.75, back=3, path_tf='M5', H=200):
    """Detect springs/upthrusts and price them as complete trades."""
    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

    #--- average volume inside the range that is being pierced, for the effort filter
    W = np.lib.stride_tricks.sliding_window_view
    vavg = np.full(n, np.nan)
    for Lv in np.unique(L[L > 0]):
        m = L == Lv
        if m.sum() == 0 or n <= Lv:
            continue
        av = np.full(n, np.nan)
        av[Lv:] = W(vol, Lv).mean(axis=1)[:-1]
        vavg[m] = av[m]

    ok = (L > 0) & np.isfinite(hi) & np.isfinite(lo) & np.isfinite(vavg) & (vavg > 0)
    #--- SPRING: pierced the range low by <= ov*ATR and closed back inside
    sp_ = ok & (l < lo) & ((lo - l) <= ov * atr) & (c > lo)
    #--- UPTHRUST: mirror
    up_ = ok & (h > hi) & ((h - hi) <= ov * atr) & (c < hi)
    idx = np.nonzero(sp_ | up_)[0]
    idx = idx[(idx > 200) & (idx < n - 5)]
    if not len(idx):
        return None
    d = np.where(sp_[idx], 1, -1)

    e = idx + 1                                  # MARKET ORDER at the next bar's open
    pi = np.clip(pmap[e], 0, len(ph) - 1)
    keep = (pi + HH < len(ph)) & (e < n)
    idx, d, e, pi = idx[keep], d[keep], e[keep], pi[keep]
    if len(idx) < 60:
        return None
    ent = o[e]
    sp = spm[e]
    ext = np.where(d > 0, l[idx], h[idx])        # the shakeout extreme, already closed
    buf = 0.10 * atr[idx]
    stop = ext - d * buf
    targ = np.where(d > 0, hi[idx], lo[idx])     # structural target: the range's far side
    risk = np.abs(ent - stop)
    rew = np.abs(targ - ent)
    good = (risk > 2 * sp) & (rew > risk * 0.25)
    idx, d, e, pi, ent, sp, stop, targ, risk, rew = (
        v[good] for v in (idx, d, e, pi, ent, sp, stop, targ, risk, rew))
    if len(idx) < 60:
        return None

    #--- non-overlapping in time, so significance is not manufactured by shared paths
    keep2, busy = [], -1
    for q in range(len(e)):
        if e[q] <= busy:
            continue
        keep2.append(q); busy = e[q] + L[idx[q]]
    keep2 = np.array(keep2, int)
    idx, d, e, pi, ent, sp, stop, targ, risk, rew = (
        v[keep2] for v in (idx, d, e, pi, ent, sp, stop, targ, risk, rew))

    r = race_px(ph, pl, pi, d, stop, targ, HH)
    RR = rew / risk
    R = np.where(r > 0, RR, np.where(r < 0, -1.0, 0.0))
    un = r == 0
    if un.any():
        q = np.minimum(pi[un] + HH, len(pc) - 1)
        R[un] = (pc[q] - ent[un]) * d[un] / risk[un]
    R = R - sp / risk
    return dict(R=R, d=d, t=t1[e], RR=RR, idx=idx,
                vshake=vol[idx] / vavg[idx], L=L[idx], unres=(r == 0).mean())


def report(sym, tf, out, tag):
    if out is None:
        print(f"  {sym:>7} {tf:>4} {tag:<22} - too few events")
        return None
    R = out['R']
    if len(R) < 60:
        print(f"  {sym:>7} {tf:>4} {tag:<22} n={len(R)} too few")
        return None
    se = R.std(ddof=1) / np.sqrt(len(R))
    f = [float(x.mean()) for x in np.array_split(R, 4)]
    pos = sum(1 for x in f if x > 0)
    print(f"  {sym:>7} {tf:>4} {tag:<22} n={len(R):>5}  R:R {out['RR'].mean():>4.1f}"
          f"  expR {R.mean():+7.3f}  t {R.mean()/max(se,1e-12):+6.2f}"
          f"  folds " + "".join(f"{x:+6.2f}" for x in f) + f"  {pos}/4")
    return R.mean(), R.mean() / max(se, 1e-12), pos, len(R)


if __name__ == '__main__':
    syms = [s for s in sys.argv[1:] if s in SYMS] or list(SYMS)
    print("=== SPRING / UPTHRUST: the shakeout trade, entered at the next bar's OPEN ===")
    print("    stop beyond the shakeout extreme, target the far side of the range.")
    print("    Driftless benchmark is expR = 0 exactly, at every geometry.\n")
    for tf in ('M15', 'H1', 'H4'):
        for s in syms:
            out = springs(s, tf)
            report(s, tf, out, 'all shakeouts')
            if out is None:
                continue
            #--- the books' volume requirement: effort at the shakeout
            for lo_, hi_, lbl in ((1.5, 99., 'climactic vol >1.5x'),
                                  (0.0, 0.8, 'quiet vol <0.8x')):
                m = (out['vshake'] >= lo_) & (out['vshake'] < hi_)
                if m.sum() >= 60:
                    sub = {k: (v[m] if isinstance(v, np.ndarray) and len(v) == len(m) else v)
                           for k, v in out.items()}
                    report(s, tf, sub, lbl)