Warrior_EA/research/book.py

"""One substrate for signals and fills, so the two can never disagree.

Every earlier test in this project computed its signal on one set of bars (mid prices, built
by `ticks_to_bars.py`) and then raced the outcome on another (M5 mid bars), bolting on an
average spread at the end. Three of the four results retracted on 2026-08-02 lived in the
seam between those two representations.

Here there is one source of truth - the M1 BID/ASK book from `bidask.py` - and everything
else is derived from it:

    HTF bars   mid OHLC aggregated from the same M1 bars, for computing signals only
    i0[k]      the index in the M1 book of the FIRST minute of HTF bar k

so `i0[e]` is, exactly and by construction, the first minute that can be traded after HTF
bar `e-1` has closed. A test computes its signal on bars up to `e-1`, hands `i0[e]` to
`fills.simulate` as the start index, and there is no arithmetic left in which to hide a
lookahead.

WHY MID BARS FOR SIGNALS
------------------------
A signal is a statement about where the market is, and the mid is the least arbitrary answer.
Using the bid (or the ask) would make every long and short setup asymmetric by half a spread
for reasons that have nothing to do with the hypothesis. Execution is where the spread is
paid, and execution is `fills.py`'s job, on the real bid and ask.

BROKER TIME
-----------
Timestamps are broker time (UTC+2, no DST in the tick files). Hour and 15-minute buckets are
unaffected by the offset; the daily bucket is not, so D1 uses an explicit offset and any
session/hour analysis must state which clock it means.
"""
import numpy as np, os, sys

BIDASK = 'c:/Users/admin/Documents/Workspaces/Market Data/bidask/'
CACHE = 'c:/Users/admin/Documents/Workspaces/Market Data/htf/'
SYMS = ('EURUSD', 'USDJPY', 'XAUUSD', 'SP500')
TF_SEC = {'M1': 60, 'M5': 300, 'M15': 900, 'M30': 1800,
          'H1': 3600, 'H4': 14400, 'D1': 86400}
#--- broker time is UTC+2 and the trading day rolls at 00:00 broker, so no shift is needed
#--- for D1 once timestamps are already broker time. Kept explicit rather than assumed.
D1_OFFSET_SEC = 0


class Frame:
    """Mid OHLC on one timeframe, plus the M1 index each bar starts at."""

    __slots__ = ('sym', 'tf', 't', 'o', 'h', 'l', 'c', 'v', 'i0', 'n', 'spread')

    def __init__(self, sym, tf, t, o, h, l, c, v, i0, spread):
        self.sym, self.tf = sym, tf
        self.t, self.o, self.h, self.l, self.c, self.v = t, o, h, l, c, v
        self.i0 = i0
        self.spread = spread          # mean ask-bid over the bar, for cost bookkeeping only
        self.n = len(t)

    def last_i0(self, k):
        """M1 index of the LAST minute of HTF bar k.

        NOT `i0[k] + bars_per_htf - 1`. The M1 array is not dense - weekends, holidays and
        session gaps are simply absent - so adding 59 array positions to an hourly bar can
        land DAYS later. That produced a Monday->Friday trade reporting 7 nights held on
        2026-08-02, and silently exited trades at the wrong price. The next bar's start minus
        one is the only correct answer, because it is defined by the index, not by arithmetic
        on an assumed bar count.
        """
        k = np.asarray(k)
        nxt = np.minimum(k + 1, self.n - 1)
        out = np.where(k + 1 < self.n, self.i0[nxt] - 1, self.i0[k])
        return np.maximum(out, self.i0[k])

    def atr(self, n=14, shift=True):
        """ATR ending at the PREVIOUS bar by default - usable at bar i's open."""
        a = _atr(self.h, self.l, self.c, n)
        return np.concatenate([[a[0]], a[:-1]]) if shift else a


def _atr(h, l, c, n=14):
    pc = np.roll(c, 1); pc[0] = c[0]
    tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc)))
    out = np.convolve(tr, np.ones(n) / n, mode='full')[:len(tr)]
    out[:n] = tr[:n].mean()
    return out


def _resample(bk, tf):
    step = TF_SEC[tf] * 1000
    off = (D1_OFFSET_SEC * 1000) if tf == 'D1' else 0
    key = (bk.t - off) // step
    s = np.concatenate(([0], np.flatnonzero(np.diff(key)) + 1))
    e = np.concatenate((s[1:] - 1, [bk.n - 1]))
    mo = 0.5 * (bk.bo + bk.ao); mh = 0.5 * (bk.bh + bk.ah)
    ml = 0.5 * (bk.bl + bk.al); mc = 0.5 * (bk.bc + bk.ac)
    t = (key[s] * step + off).astype(np.int64)
    o = mo[s]
    h = np.maximum.reduceat(mh, s)
    l = np.minimum.reduceat(ml, s)
    c = mc[e]
    v = (e - s + 1).astype(np.float64)                      # minutes traded, a volume proxy
    sp = bk.ac - bk.bc
    spm = np.add.reduceat(sp, s) / v
    return t, o, h, l, c, v, s.astype(np.int64), spm


def frame(sym, tf, bk=None):
    """HTF mid bars for `sym`, cached on disk. Pass an open Book to avoid reloading it."""
    if tf == 'M1' and bk is not None:
        mo = 0.5 * (bk.bo + bk.ao); mh = 0.5 * (bk.bh + bk.ah)
        ml = 0.5 * (bk.bl + bk.al); mc = 0.5 * (bk.bc + bk.ac)
        return Frame(sym, tf, bk.t, mo, mh, ml, mc, np.ones(bk.n),
                     np.arange(bk.n, dtype=np.int64), bk.ac - bk.bc)
    os.makedirs(CACHE, exist_ok=True)
    fn = f"{CACHE}{sym}_{tf}_mid.npz"
    if os.path.exists(fn):
        z = np.load(fn)
        return Frame(sym, tf, z['t'], z['o'], z['h'], z['l'], z['c'], z['v'],
                     z['i0'], z['spread'])
    import fills
    bk = bk or fills.Book(sym)
    t, o, h, l, c, v, i0, sp = _resample(bk, tf)
    np.savez_compressed(fn, t=t, o=o, h=h, l=l, c=c, v=v, i0=i0, spread=sp)
    return Frame(sym, tf, t, o, h, l, c, v, i0, sp)


def rolling_span(h, l, L):
    """(max high, min low) over the L bars ENDING AT i-1. Never includes bar i."""
    from numpy.lib.stride_tricks import sliding_window_view
    n = len(h)
    hi = np.full(n, np.nan); lo = np.full(n, np.nan)
    if n > L:
        hi[L:] = sliding_window_view(h, L).max(axis=1)[:-1]
        lo[L:] = sliding_window_view(l, L).min(axis=1)[:-1]
    return hi, lo


def find_ranges(h, l, atr, Ls=(12, 18, 24, 36, 48, 72, 96, 144), theta=0.60):
    """Longest qualifying consolidation window ending at i-1, per bar. Same definition as
    `test_cause_effect.find_ranges` so results here remain comparable to the earlier work."""
    n = len(h)
    L_out = np.zeros(n, np.int32)
    hi_out = np.full(n, np.nan); lo_out = np.full(n, np.nan)
    for L in sorted(Ls, reverse=True):
        hi, lo = rolling_span(h, l, L)
        comp = (hi - lo) / np.maximum(atr * np.sqrt(L), 1e-12)
        ok = (L_out == 0) & np.isfinite(comp) & (comp <= theta)
        L_out[ok] = L; hi_out[ok] = hi[ok]; lo_out[ok] = lo[ok]
    return L_out, hi_out, lo_out


def nonoverlap(idx, span):
    """Keep a maximal set of events whose [i, i+span) windows do not intersect.

    Overlapping trades share price path, so their outcomes are correlated and a naive t on
    them is inflated - 6,197 overlapping trades once gave t +8.06 where 163 independent ones
    gave +1.61. Every headline number in this file is computed on the independent subset.
    """
    idx = np.asarray(idx); span = np.asarray(span)
    order = np.argsort(idx, kind='stable')
    keep = np.zeros(len(idx), bool)
    busy = -1
    for q in order:
        if idx[q] > busy:
            keep[q] = True
            busy = idx[q] + int(span[q])
    return keep


def tstat(x):
    x = np.asarray(x, float)
    if len(x) < 3:
        return 0.0
    se = x.std(ddof=1) / np.sqrt(len(x))
    return float(x.mean() / max(se, 1e-12))


def slope_t(y, x):
    """OLS slope of y on x and its t - the dose-response statistic."""
    x = np.asarray(x, float); y = np.asarray(y, float)
    if len(x) < 5 or x.std() < 1e-9:
        return 0.0, 0.0
    X = np.column_stack([np.ones(len(x)), x])
    beta, *_ = np.linalg.lstsq(X, y, rcond=None)
    resid = y - X @ beta
    s2 = resid @ resid / max(len(x) - 2, 1)
    se = np.sqrt(s2 * np.linalg.inv(X.T @ X)[1, 1])
    return float(beta[1]), float(beta[1] / max(se, 1e-12))


if __name__ == '__main__':
    sys.stdout.reconfigure(encoding='utf-8', errors='replace')
    import fills, datetime as dt
    only = [s for s in sys.argv[1:] if s in SYMS] or list(SYMS)
    for s in only:
        bk = fills.Book(s)
        print(f"\n=== {s} ===  M1 bars {bk.n:,}")
        for tf in ('M5', 'M15', 'H1', 'H4', 'D1'):
            f = frame(s, tf, bk)
            a = dt.datetime.fromtimestamp(f.t[0] / 1000, dt.UTC)
            b = dt.datetime.fromtimestamp(f.t[-1] / 1000, dt.UTC)
            #--- i0 must be strictly increasing and index the right minute, or every fill
            #--- in every test built on this is off by an unknown amount
            assert (np.diff(f.i0) > 0).all(), f"{s} {tf}: i0 not increasing"
            #--- f.t is the BUCKET START; the first minute that actually traded inside it
            #--- can be later (a quiet Sunday open, a holiday). Containment is the invariant
            #--- that matters: i0 must point inside its own bucket and nowhere else.
            step = TF_SEC[tf] * 1000
            off = bk.t[f.i0] - f.t
            assert (off >= 0).all() and (off < step).all(), \
                f"{s} {tf}: i0 outside its bucket (max off {off.max()/1000:.0f}s)"
            print(f"  {tf:>3} {f.n:>9,} bars  {a:%Y-%m-%d}..{b:%Y-%m-%d}"
                  f"  median spread/ATR14 "
                  f"{np.nanmedian(f.spread / np.maximum(f.atr(), 1e-12)):.4f}")