Warrior_EA/research/test_spread2.py

"""Did charging one average spread INVENT the context slope?

An honest fill model should move the LEVEL of every bucket by the cost and leave the SHAPE
alone - cost is not supposed to know what the Wyckoff traces said. Yet switching engines cut
the context slope from +0.0425 (t +3.18) to +0.0239 (t +1.37). Something in the cost is
correlated with the score.

There is an obvious candidate, and it is the thing the bid/ask bars were built to expose:
EURUSD's p95 spread is 4x its median. Every earlier test charged a single average and
assumed those hours away. If high-agreement setups happen disproportionately in the wide-
spread hours - news, thin liquidity, the open - then the old model undercharged exactly the
bucket the hypothesis cared about, and manufactured part of the slope out of a billing error.

THE TEST
--------
For the same LPS events, regress the REALISED cost (spread at the fill minute / risk) on the
agreement count. If the slope of cost-on-agreement is materially non-zero, an average-spread
model biases the dose-response by that amount per trace, and the direction of the bias says
which way.

Also reported: what the OLD model would have charged (one median spread for the symbol) so
the two can be compared as money rather than as an argument.
"""
import numpy as np, sys
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
import fills, book, wyckoff, test_lps2

SYMS = ('EURUSD', 'USDJPY', 'XAUUSD', 'SP500')


def costs(sym, tf, **kw):
    """Same events as the attribution run, but returning the cost actually paid."""
    bk = fills.Book(sym)
    f = book.frame(sym, tf, bk)
    a = test_lps2.events(sym, tf, bk=bk, f=f, **kw)
    if a is None:
        return None
    #--- re-derive the fill minutes so the spread can be read where it was really paid
    return a, bk, f


def run(sym, tf):
    bk = fills.Book(sym)
    f = book.frame(sym, tf, bk)
    step = book.TF_SEC[tf] // 60
    h, l, c = f.h, f.l, f.c
    atr = f.atr(14)
    L, hi_, lo_ = book.find_ranges(h, l, atr, theta=0.60)
    up = (L > 0) & (c > hi_); dn = (L > 0) & (c < lo_)
    fire = np.nonzero(up | dn)[0]
    fire = fire[(fire > 305) & (fire < f.n - 65)]
    dirs = np.where(up[fire], 1, -1)
    rows, busy = [], -1
    for q in range(len(fire)):
        i = int(fire[q])
        if i <= busy:
            continue
        dd = int(dirs[q]); Lq = int(L[i]); s = i - Lq
        if s < 1:
            continue
        top, bot = hi_[i], lo_[i]
        lvl = top if dd > 0 else bot
        j = -1
        for k in range(1, 61):
            b_ = i + k
            if b_ >= f.n - 2:
                break
            near = (l[b_] <= lvl + 0.35 * atr[i]) if dd > 0 else (h[b_] >= lvl - 0.35 * atr[i])
            if near and ((c[b_] > lvl) if dd > 0 else (c[b_] < lvl)):
                j = b_; break
            if (c[b_] < lvl - 0.35 * atr[i]) if dd > 0 else (c[b_] > lvl + 0.35 * atr[i]):
                break
        if j < 0:
            continue
        e = j + 1
        if e >= f.n - 1:
            continue
        ext = l[j] if dd > 0 else h[j]
        stop = ext - dd * 0.10 * atr[i]
        ag = wyckoff.traces(f, s, i, top, bot, dd, int(np.sign(c[i] - c[i - 200])))
        rows.append((e, dd, stop, ag))
        busy = e + Lq
    if len(rows) < 100:
        return None
    E = np.array([r[0] for r in rows]); D = np.array([r[1] for r in rows])
    ST = np.array([r[2] for r in rows], float); AG = np.array([r[3] for r in rows])
    start = f.i0[E]
    ent = np.where(D > 0, bk.ao[start], bk.bo[start])
    risk = np.abs(ent - ST)
    ok = risk > 4 * f.spread[E]
    E, D, ST, AG, start, risk = (v[ok] for v in (E, D, ST, AG, start, risk))
    sp_real = (bk.ac - bk.bc)[start]                 # what was actually paid, that minute
    sp_med = np.median(bk.ac - bk.bc)                # what the old model charged, always
    return dict(ag=AG, cost_real=sp_real / risk, cost_old=sp_med / risk,
                sp_real=sp_real, sp_med=sp_med, n=len(AG))


if __name__ == '__main__':
    syms = [s for s in sys.argv[1:] if s in SYMS] or list(SYMS)
    print("=== IS THE SPREAD CORRELATED WITH THE CONTEXT SCORE? ===")
    print("    if cost rises with agreement, an average-spread model undercharges the")
    print("    bucket the hypothesis cares about and inflates the dose-response.\n")
    print(f"  {'sym':>7}{'tf':>5}{'n':>6}{'spread p50':>12}{'p95':>9}{'p95/p50':>9}"
          f"{'cost real':>11}{'cost old':>10}{'d(cost)/trace':>15}{'t':>7}")
    PA, PC, PO = [], [], []
    for tf in ('M15', 'H1'):
        for s in syms:
            r = run(s, tf)
            if r is None:
                print(f"  {s:>7}{tf:>5}     - too few"); continue
            sl, tt = book.slope_t(r['cost_real'], r['ag'])
            print(f"  {s:>7}{tf:>5}{r['n']:>6}{r['sp_med']:>12.5f}"
                  f"{np.quantile(r['sp_real'],0.95):>9.5f}"
                  f"{np.quantile(r['sp_real'],0.95)/r['sp_med']:>9.2f}"
                  f"{r['cost_real'].mean():>11.4f}{r['cost_old'].mean():>10.4f}"
                  f"{sl:>+15.5f}{tt:>+7.2f}")
            PA.append(r['ag']); PC.append(r['cost_real']); PO.append(r['cost_old'])
    if PA:
        ag = np.concatenate(PA); cr = np.concatenate(PC); co = np.concatenate(PO)
        sl, tt = book.slope_t(cr, ag)
        so, to = book.slope_t(co, ag)
        print(f"\n  POOLED n={len(ag):,}")
        print(f"    realised cost  slope {sl:+.5f} R/trace  t {tt:+.2f}   mean {cr.mean():.4f}")
        print(f"    old flat cost  slope {so:+.5f} R/trace  t {to:+.2f}   mean {co.mean():.4f}")
        print(f"    BIAS an average-spread model puts into the dose-response:"
              f" {so - sl:+.5f} R per trace")
        for k in range(6):
            m = ag == k
            if m.sum() >= 25:
                print(f"      {k} agree n={int(m.sum()):>5}  realised cost {cr[m].mean():.4f}"
                      f"   flat {co[m].mean():.4f}   undercharged by {cr[m].mean()-co[m].mean():+.4f}")