Warrior_EA/research/wyckoff_scan.py

"""
Wyckoff structure scanner + paper-trade ledger (2026-10-05).  Tick volume only (real volume ignored by decision).

PRE-REGISTERED (written before any result was seen) - thresholds are the playbook's proposals
(docs/Wyckoff/books/PLAYBOOK.md section 5, marked [P]), NOT fitted:
  ATR14; wide bar >= 1.3 ATR; high volume RV >= 1.5 where RV = tick volume / median of the SAME HOUR over the prior
  20 sessions (time-of-day removed); touch tolerance 0.25 ATR; swing = 3-bar fractal CONFIRMED 3 bars late (causal);
  range = >= 2 swing highs within tol of the top AND >= 2 swing lows within tol of the bottom among swings confirmed
  in the last 60 bars, height 2..8 ATR, efficiency ratio < 0.35.   prior trend = move over the 60 bars before the
  box start, > 3 ATR.   Everything dated >= 2025-01-01 is sealed.
Setups (entry at the OPEN of the next bar, cost deducted, label = triple barrier TP 2 / SL 1.5 ATR / 48 bars):
  SPRING_L : box + prior trend down + a bar pierced the bottom and closed back inside within 3 bars (this bar completes it)
  UTAD_S   : mirror
  SOS_L    : box + SOS bar (wide, RV>=1.5, close in top third, close > box top by 0.25 ATR)
  SOW_S    : mirror
Control for each: the same direction on EVERY bar where a box exists (no trigger).  Also the unconditional all-bars mean.
Volume sanity: a window is 'bad' when >5% of its last 120 bars have zero volume or its 99th-percentile RV > 15.
  python wyckoff_scan.py            # all 9 instruments, prints the ledger summary, writes wyckoff_ledger.csv
"""
from __future__ import annotations
import os, sys
import numpy as np
import pandas as pd

sys.path.insert(0, os.path.dirname(__file__))
import discover as D

TOL, WIDE, HIGHV = 0.25, 1.3, 1.5
TP, SL, T = 2.0, 1.5, 48
SYMS = list(D.START)


def rel_volume(b: pd.DataFrame) -> np.ndarray:
    hr = b.index.hour
    rv = np.full(len(b), np.nan)
    tv = b.tv.to_numpy(float)
    for h in range(24):
        idx = np.where(hr == h)[0]
        if len(idx) < 5:
            continue
        s = pd.Series(tv[idx]).rolling(20, min_periods=5).median().shift(1).to_numpy()
        rv[idx] = tv[idx] / np.where(s > 0, s, np.nan)
    return rv


def swings(h: np.ndarray, l: np.ndarray, k: int = 3):
    n = len(h)
    sh = np.zeros(n, bool); sl_ = np.zeros(n, bool)
    for i in range(k, n - k):
        if h[i] == h[i - k:i + k + 1].max() and h[i] > h[i - 1]: sh[i] = True
        if l[i] == l[i - k:i + k + 1].min() and l[i] < l[i - 1]: sl_[i] = True
    return sh, sl_                                   # swing at i is KNOWN only at bar i+k


def scan(sym: str) -> pd.DataFrame:
    b = D.load_h1(sym)
    b = b[b.index < D.SEAL]
    b = b[b.index.year >= D.START[sym]]
    o, h, l, c = (b[x].to_numpy(float) for x in "ohlc")
    n = len(b)
    pc = np.r_[c[0], c[:-1]]
    tr = np.maximum(h - l, np.maximum(abs(h - pc), abs(l - pc)))
    atr = pd.Series(tr).rolling(14).mean().to_numpy()
    rv = rel_volume(b)
    zero = pd.Series(b.tv.to_numpy() == 0).rolling(120).mean().to_numpy()
    rv99 = pd.Series(rv).rolling(120, min_periods=60).quantile(0.99).to_numpy()
    sh, sl_ = swings(h, l)
    cost = D.COST_BP[sym] * 1e-4
    rows = []
    for t in range(200, n - T - 2):
        a = atr[t]
        if not a > 0:
            continue
        # swings confirmed by t, in the last 60 bars
        lo = t - 60
        ih = np.where(sh[lo:t - 2])[0] + lo
        il = np.where(sl_[lo:t - 2])[0] + lo
        if len(ih) < 2 or len(il) < 2:
            continue
        top, bot = h[ih].max(), l[il].min()
        ht = (top - bot) / a
        if not 2 <= ht <= 8:
            continue
        nt = (h[ih] >= top - TOL * a).sum(); nb = (l[il] <= bot + TOL * a).sum()
        if nt < 2 or nb < 2:
            continue
        w = c[lo:t + 1]
        er = abs(w[-1] - w[0]) / (np.abs(np.diff(w)).sum() + 1e-12)
        if er > 0.35:
            continue
        start = min(ih.min(), il.min())
        pre = c[start] - c[max(0, start - 60)]
        prior = 1 if pre > 3 * a else -1 if pre < -3 * a else 0
        # shakeout completed on THIS bar: some bar in t-2..t pierced an edge, bar t closes back inside
        pierced_lo = l[t - 2:t + 1].min() < bot
        pierced_hi = h[t - 2:t + 1].max() > top
        spring = pierced_lo and c[t] > bot and (c[t - 1] <= bot or l[t] < bot)   # back inside only now
        utad = pierced_hi and c[t] < top and (c[t - 1] >= top or h[t] > top)
        rng = h[t] - l[t]; cl = (c[t] - l[t]) / rng if rng > 0 else .5
        wide = rng >= WIDE * a; hv = rv[t] >= HIGHV
        sos = wide and hv and cl >= .67 and c[t] > top + TOL * a
        sow = wide and hv and cl <= .33 and c[t] < bot - TOL * a
        # outcome, long and short, entry next open
        e = o[t + 1]
        res = {}
        for side in (1, -1):
            tp_p, sl_p = e + side * TP * a, e - side * SL * a
            out = side * (c[t + T] - e) / a - cost * e / a * 2   # vertical exit, fallback
            for j in range(t + 1, t + 1 + T):
                hit_sl = (l[j] <= sl_p) if side == 1 else (h[j] >= sl_p)
                hit_tp = (h[j] >= tp_p) if side == 1 else (l[j] <= tp_p)
                if hit_sl:
                    out = -SL - cost * e / a * 2; break
                if hit_tp:
                    out = TP - cost * e / a * 2; break
            res[side] = out
        rows.append((sym, b.index[t], prior, ht, int(spring), int(utad), int(sos), int(sow), rv[t],
                     int(zero[t] > .05 or rv99[t] > 15), res[1], res[-1]))
    return pd.DataFrame(rows, columns=["sym", "t", "prior", "ht", "spring", "utad", "sos", "sow", "rv",
                                      "badvol", "rL", "rS"])


def thin(df: pd.DataFrame, gap: int = 48) -> pd.DataFrame:
    d = df.sort_values("t")
    hrs = d.t.to_numpy().astype("datetime64[h]").astype("int64")
    keep, last = [], -10**12
    for i, hh in enumerate(hrs):
        if hh - last >= gap:
            keep.append(i); last = hh
    return d.iloc[keep]


def tstat(x: np.ndarray) -> float:
    return float(x.mean() / (x.std(ddof=1) / np.sqrt(len(x)))) if len(x) > 2 and x.std() > 0 else float("nan")


def main():
    parts = []
    for s in SYMS:
        p = os.path.join(D.CACHE, f"{s}_wyscan.pkl")
        d = pd.read_pickle(p) if os.path.exists(p) else scan(s)
        d.to_pickle(p); parts.append(d); print(s, len(d), "box-bars", flush=True)
    L = pd.concat(parts)
    L = L[L.badvol == 0]
    L.to_csv(os.path.join(os.path.dirname(__file__), "wyckoff_ledger.csv"), index=False)
    print("\nR units = ATR multiples net of cost (TP 2 / SL 1.5 / 48 bars).  trades thinned to >=48h apart per symbol.")
    print(f"{'setup':10s}{'n':>6s}{'meanR':>8s}{'t':>6s}{'ctl meanR':>11s}{'ctl n':>7s}{'lift':>7s}{'pos yrs':>9s}")
    setups = [("SPRING_L", (L.spring == 1) & (L.prior == -1), "rL"), ("SOS_L", L.sos == 1, "rL"),
              ("UTAD_S", (L.utad == 1) & (L.prior == 1), "rS"), ("SOW_S", L.sow == 1, "rS")]
    for name, m, col in setups:
        tr = pd.concat([thin(g) for _, g in L[m].groupby("sym")]) if m.sum() else L[m]
        ctl = pd.concat([thin(g) for _, g in L.groupby("sym")])
        if len(tr) < 3:
            print(f"{name:10s}{len(tr):6d}  too few"); continue
        x = tr[col].to_numpy(); y = ctl[col].to_numpy()
        yrs = tr.groupby(tr.t.dt.year)[col].mean()
        print(f"{name:10s}{len(x):6d}{x.mean():8.3f}{tstat(x):6.1f}{y.mean():11.3f}{len(y):7d}{x.mean() - y.mean():7.3f}"
              f"{(yrs > 0).sum():>5d}/{len(yrs)}")
    print("\nper symbol (n, meanR):")
    for name, m, col in setups:
        t = L[m]
        print(name, {s: (int(len(g)), round(float(g[col].mean()), 2)) for s, g in t.groupby("sym")})


if __name__ == "__main__":
    main()