Warrior_EA/research/test_spread.py

"""Does the spread series carry information about the barrier outcome?

Spread is the one microstructure channel that survived the API audit: it is FX-available,
and - unusually - it is genuinely historical in the Strategy Tester ("During testing, the
spread is not modeled but is taken from historical data"). Everything else that looked
promising is either absent on FX (signed tick flow, real volume), absent on retail FX and
never replayed in the tester (depth of market), or has no history at all (swap).

Same machinery and same discipline as test_volume.py: mutual information with the triple-
barrier label, block-permutation null sized to the barrier horizon, finite-sample bias
quoted rather than subtracted.

Note on what spread can and cannot be: it is UNSIGNED, like volume. A widening spread says
liquidity is withdrawing, not which way price will go. So the realistic hope here is a
regime/filter term - "is this a bar worth trading at all" - not a directional signal.
"""
import numpy as np, sys, time
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
from kit import load_rates, atr, sma, barrier_vec
from test_volume import rank_bin, mi_binned, block_perm_null_multi, NPERM


def build_spread_features(h, l, c, spr, a):
    s = spr.astype(float)
    base = sma(s, 50)
    prev = np.roll(s, 1); prev[0] = s[0]
    F, names = [], []
    F.append(np.clip(np.where(base > 0, s / base, 1.0), 0, 5));        names.append("sprLevel s/sma50")
    F.append(np.clip(np.where(prev > 0, (s - prev) / prev, 0.0), -5, 5)); names.append("sprChange")
    # cost relative to the volatility the trade must overcome - the term that actually decides
    # whether a setup is affordable, and the one that varies most across sessions
    F.append(np.clip(np.where(a > 0, s / a, 0.0), 0, 5));              names.append("spr/atr (cost)")
    # widening WHILE the bar travels: liquidity withdrawing into a move, a stress signature
    rng_atr = np.where(a > 0, (h - l) / a, 0.0)
    F.append(np.clip(np.where(base > 0, (s / base) * rng_atr, 0.0), 0, 5)); names.append("sprLevel x range")
    return [np.nan_to_num(f) for f in F], names


def run(sym, tf, sl_m, tp_m, H):
    t, o, h, l, c, v, spr = load_rates(sym, tf)
    a = atr(h, l, c, 14)
    an = np.where(a > 0, a, np.nan)
    tick = np.nanmin(np.abs(np.diff(np.unique(np.round(c, 8)))))
    sp = np.nanmedian(spr) * tick
    if not np.isfinite(sp):
        sp = 0.0
    lab, valid = barrier_vec(h, l, c, an, sl_m, tp_m, H, sp)
    F, names = build_spread_features(h, l, c, spr, an)
    m = valid & np.isfinite(a) & (a > 0) & np.isfinite(spr) & (spr > 0)
    m[:200] = False
    m[-(H + 2):] = False
    y = lab[m]
    n = int(m.sum())
    if n < 5000:
        print(f"\n=== {sym} SL{sl_m}:TP{tp_m} - only {n} usable bars, skipped ===")
        return
    print(f"\n=== {sym}  SL{sl_m}:TP{tp_m} H={H}  n={n}  "
          f"median spread {np.nanmedian(spr):.1f} pts ({sp/np.nanmedian(a):.3f} ATR) ===")
    print(f"  finite-sample MI bias ~ 7/n = {7.0/n:.6f} nats")
    xbs = [rank_bin(f[m]) for f in F]
    obs = [mi_binned(xb, y) for xb in xbs]
    nulls = block_perm_null_multi(xbs, y, H)
    print(f"{'feature':<22}{'MI (nats)':>12}{'null mean':>12}{'null p95':>11}{'excess':>10}{'p':>8}")
    for k, nm in enumerate(names):
        null = nulls[k]
        p = (1 + int((null >= obs[k]).sum())) / (NPERM + 1)
        star = ' *' if p < 0.05 else ''
        print(f"{nm:<22}{obs[k]:>12.6f}{null.mean():>12.6f}{np.quantile(null,0.95):>11.6f}"
              f"{obs[k]-null.mean():>+10.6f}{p:>8.3f}{star}", flush=True)


if __name__ == '__main__':
    t0 = time.time()
    for sym in ('EURUSD', 'USDJPY', 'XAUUSD', 'SP500'):
        for (s, p, H) in [(2, 3, 96), (1, 2, 64)]:
            try:
                run(sym, 16385, s, p, H)
            except Exception as ex:
                print(f"{sym} {s}:{p} FAILED {ex}")
    print(f"\ntotal {time.time()-t0:.0f}s")