Warrior_EA/research/breadth.py

"""Five more instruments, to settle whether the context slope is a small real effect or nothing.

On the honest engine the Wyckoff context slope came in at +0.0239 R/trace, t +1.37, with 6 of
8 cells keeping the sign. That is the awkward middle: too weak to trade, too consistent to
dismiss. The only way past it is more independent instruments, and `sqxbars.py` has just made
five available that share no data path with the four already used - FTSE, UK100, WTI (two
feeds) and USDCAD.

WHAT IS APPROXIMATED HERE, AND WHY IT IS ACCEPTABLE
---------------------------------------------------
These files carry M1 bars with no ask, so the bid/ask book is SYNTHESISED by applying a
relative spread. That is a real approximation and it is only defensible for this particular
question, because the thing being measured is a SLOPE across context buckets and the cost was
already measured to be almost uncorrelated with the context score (+0.0008 R per trace,
t +0.85, against an effect that would have to be ~50x larger). A cost assumption shifts every
bucket together; it cannot manufacture or hide a slope.

The BASE level is a different matter and is quoted as approximate throughout.

Spreads are set as a FRACTION of price rather than in points, which also makes the whole
thing invariant to the decimal scale - convenient, because these symbols have no reference
series to calibrate a scale against, and every result here is in ATR units or R-multiples
anyway. Values are taken from the four instruments where the real spread IS known:
EURUSD 0.45 bp, USDJPY 0.64 bp, SP500 1.3 bp, XAUUSD 2.1 bp - and rounded UP, so the base is
pessimistic rather than flattering.
"""
import numpy as np, sys, os
import book, sqxbars

#--- relative half-spread inputs, in basis points of price, rounded up from measured peers
SPREAD_BP = {
    'GBRIDXGBP_dukascopy__the5ers': 1.5,     # FTSE 100 cash index
    'UK100_the5ers': 1.5,
    'LIGHTCMDUSD_dukascopy__the5ers': 4.0,   # WTI crude
    'XTIUSD_the5ers': 4.0,
    'USDCAD_dukascopy__the5ers': 1.0,
}
NICE = {
    'GBRIDXGBP_dukascopy__the5ers': 'FTSE100',
    'UK100_the5ers': 'UK100',
    'LIGHTCMDUSD_dukascopy__the5ers': 'WTI_d',
    'XTIUSD_the5ers': 'WTI_5',
    'USDCAD_dukascopy__the5ers': 'USDCAD',
}


class MidBook:
    """A fills.Book built from mid-only M1 bars by applying a relative spread.

    Quacks like fills.Book. The spread is multiplicative so it scales with price over a
    20-year sample instead of being a constant that is generous in 2012 and absurd in 2026.
    """

    def __init__(self, bars, bp):
        t, o, h, l, c = (bars[:, k] for k in range(5))
        self.t = t.astype(np.int64)
        self.n = len(t)
        e = 0.5 * bp * 1e-4
        self.bo, self.bh, self.bl, self.bc = o * (1 - e), h * (1 - e), l * (1 - e), c * (1 - e)
        self.ao, self.ah, self.al, self.ac = o * (1 + e), h * (1 + e), l * (1 + e), c * (1 + e)

    def index_at(self, t_ms):
        return np.searchsorted(self.t, np.asarray(t_ms, np.int64), 'left')


def get(sym, tf, decimals=6):
    """-> (MidBook, Frame). Decimals are irrelevant to every statistic computed on these."""
    bars, _ = sqxbars.load(sym, 'M1', decimals=decimals, verbose=False)
    #--- drop duplicate and out-of-order minutes; a resample assumes a sorted, unique index
    t = bars[:, 0].astype(np.int64)
    keep = np.concatenate(([True], np.diff(t) > 0))
    bars = bars[keep]
    bk = MidBook(bars, SPREAD_BP.get(sym, 1.5))
    t, o, h, l, c, v, i0, sp = book._resample(bk, tf)
    return bk, book.Frame(NICE.get(sym, sym), tf, t, o, h, l, c, v, i0, sp)


if __name__ == '__main__':
    sys.stdout.reconfigure(encoding='utf-8', errors='replace')
    import datetime as dt
    print("=== BREADTH INSTRUMENTS ===")
    print(f"  {'symbol':>9}{'M1 bars':>12}{'H1 bars':>10}  range"
          f"{'':>14}{'median px':>12}{'spread/ATR H1':>15}")
    for s in SPREAD_BP:
        try:
            bk, f = get(s, 'H1')
        except Exception as ex:
            print(f"  {NICE.get(s,s):>9}  FAILED: {ex}")
            continue
        a = dt.datetime.fromtimestamp(f.t[0] / 1000, dt.UTC)
        b = dt.datetime.fromtimestamp(f.t[-1] / 1000, dt.UTC)
        print(f"  {NICE[s]:>9}{bk.n:>12,}{f.n:>10,}  {a:%Y-%m-%d}..{b:%Y-%m-%d}"
              f"{np.median(f.c):>12.2f}"
              f"{np.nanmedian(f.spread/np.maximum(f.atr(),1e-12)):>15.4f}")