Warrior_EA/research/bidask.py

"""M1 bars carrying SEPARATE BID and ASK OHLC - the substrate a correct fill model needs.

Every bar file in this project so far stores MID prices. That is fine for measuring returns
and useless for simulating orders, because no order ever executes at the mid:

    a buy  fills at the ASK      a buy stop triggers when the ASK reaches it
    a sell fills at the BID      a sell stop triggers when the BID reaches it

Mid-price bars force you to bolt the spread on afterwards as an average, which is exactly
the approximation that let three separate fill artifacts through today. With bid and ask
carried separately the spread is whatever it actually was on that bar, including the
overnight and news blowouts that an average hides.

WHY M1 AND NOT RAW TICKS
------------------------
Raw ticks would be exact, but 513M ticks per symbol is ~8 GB even packed, and four symbols
will not sit in memory. M1 keeps the file at ~500 MB per symbol and shrinks the residual
same-bar ambiguity by 60x versus H1 and 5x versus M5 - and that ambiguity is the only thing
tick resolution buys here, because within one M1 bar on these instruments the ordering of a
stop and a target is rarely decisive. Stated plainly rather than assumed: this is a bounded
approximation, not an exact simulator, and `fills.py` reports how often a bar contained both
barriers so the bound is visible in every result.
"""
import numpy as np, sys, os, time
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
from sqx import decode_iter, split_offsets, calibrate_decimals, decode, find_block, NO_VOLUME

OUT = 'c:/Users/admin/Documents/Workspaces/Market Data/bidask/'
COLS = ['time', 'bo', 'bh', 'bl', 'bc', 'ao', 'ah', 'al', 'ac', 'ticks']
_I = {c: k for k, c in enumerate(COLS)}


def _scan(args):
    path, scale, step, start, stop = args
    rows = []
    pend = None
    for ts, ai, bi, _v in decode_iter(path, start=start, stop=stop):
        ask = ai / scale
        bid = bi / scale
        bucket = (ts // step) * step
        s = np.concatenate(([0], np.flatnonzero(np.diff(bucket)) + 1))
        e = np.concatenate((s[1:] - 1, [len(ts) - 1]))
        part = np.column_stack([
            bucket[s].astype(np.float64),
            bid[s], np.maximum.reduceat(bid, s), np.minimum.reduceat(bid, s), bid[e],
            ask[s], np.maximum.reduceat(ask, s), np.minimum.reduceat(ask, s), ask[e],
            (e - s + 1).astype(np.float64)])
        lst = [tuple(r) for r in part]
        if pend is not None:
            if lst and lst[0][0] == pend[0]:
                lst[0] = _merge(pend, lst[0])
            else:
                rows.append(pend)
        pend = lst.pop() if lst else pend
        rows.extend(lst)
    if pend is not None:
        rows.append(pend)
    return np.array(rows, np.float64) if rows else np.empty((0, len(COLS)))


def _merge(a, b):
    o = list(a)
    o[_I['bh']] = max(a[_I['bh']], b[_I['bh']]); o[_I['bl']] = min(a[_I['bl']], b[_I['bl']])
    o[_I['ah']] = max(a[_I['ah']], b[_I['ah']]); o[_I['al']] = min(a[_I['al']], b[_I['al']])
    o[_I['bc']] = b[_I['bc']]; o[_I['ac']] = b[_I['ac']]
    o[_I['ticks']] = a[_I['ticks']] + b[_I['ticks']]
    return tuple(o)


def finalise(arr):
    if not len(arr):
        return arr
    t = arr[:, 0]
    if (np.diff(t) < 0).any():
        arr = arr[np.argsort(t, kind='stable')]; t = arr[:, 0]
    if (np.diff(t) == 0).any():
        s = np.concatenate(([0], np.flatnonzero(np.diff(t)) + 1))
        e = np.concatenate((s[1:] - 1, [len(t) - 1]))
        out = arr[s].copy()
        out[:, _I['bh']] = np.maximum.reduceat(arr[:, _I['bh']], s)
        out[:, _I['bl']] = np.minimum.reduceat(arr[:, _I['bl']], s)
        out[:, _I['ah']] = np.maximum.reduceat(arr[:, _I['ah']], s)
        out[:, _I['al']] = np.minimum.reduceat(arr[:, _I['al']], s)
        out[:, _I['bc']] = arr[e, _I['bc']]; out[:, _I['ac']] = arr[e, _I['ac']]
        out[:, _I['ticks']] = np.add.reduceat(arr[:, _I['ticks']], s)
        arr = out
    #--- a negative spread is impossible and means the scale or the field order is wrong;
    #--- fail loudly rather than let it poison every fill downstream
    bad = int((arr[:, _I['ac']] < arr[:, _I['bc']]).sum())
    if bad > len(arr) // 1000:
        raise RuntimeError(f"{bad:,}/{len(arr):,} bars have ask < bid - check the decode")
    return arr


def build(path, sym, ref, tf_seconds=60, workers=None):
    sts, sai, _, _ = decode(path, max_records=200000)
    try:
        dec, err = calibrate_decimals(sts, sai, ref[0], ref[1], verbose=False)
    except ValueError:
        size = os.path.getsize(path); dec = None
        for frac in (0.5, 0.75, 0.9):
            with open(path, 'rb') as fh:
                off = find_block(fh, int(size * frac), size)
            if off is None:
                continue
            it = decode_iter(path, batch=2000000, start=off)
            sts, sai, _, _ = next(it); it.close()
            try:
                dec, err = calibrate_decimals(sts, sai, ref[0], ref[1], verbose=False); break
            except ValueError:
                continue
        if dec is None:
            raise
    print(f"  decimals={dec} (rel.err {err:.4%})", flush=True)
    scale = 10.0 ** dec
    workers = workers or max(1, (os.cpu_count() or 4) - 1)
    ranges = split_offsets(path, workers)
    t0 = time.time()
    import multiprocessing as mp
    args = [(path, scale, tf_seconds * 1000, s, e) for s, e in ranges]
    parts = []
    with mp.Pool(len(ranges)) as pool:
        for k, p in enumerate(pool.imap(_scan, args)):
            parts.append(p)
            print(f"  chunk {k+1}/{len(ranges)}  {len(p):,} bars  "
                  f"{(time.time()-t0)/60:.1f} min", flush=True)
    arr = finalise(np.concatenate([p for p in parts if len(p)]))
    sp = arr[:, _I['ac']] - arr[:, _I['bc']]
    print(f"  {len(arr):,} M1 bars   median spread {np.median(sp):.6f}   "
          f"p95 {np.quantile(sp,0.95):.6f}   {(time.time()-t0)/60:.1f} min")
    return arr


if __name__ == '__main__':
    from kit import load_rates
    D = 'c:/Users/admin/Documents/Workspaces/Market Data/'
    os.makedirs(OUT, exist_ok=True)
    JOBS = [('EURUSD_tick_the5ers_TICK.dat', 'EURUSD'),
            ('USDJPY_tick_the5ers_TICK.dat', 'USDJPY'),
            ('XAUUSD_tick_the5ers_TICK.dat', 'XAUUSD'),
            ('USA500IDXUSD_tick_the5ers_TICK.dat', 'SP500')]
    only = sys.argv[1] if len(sys.argv) > 1 else None
    for fn, sym in JOBS:
        if only and sym != only:
            continue
        out = f"{OUT}{sym}_M1_bidask.npz"
        if os.path.exists(out):
            print(f"{sym}: exists, skipping"); continue
        print(f"\n=== {sym} bid/ask M1 ===", flush=True)
        rt, ro, rh, rl, rc, rv, rs = load_rates(sym, 16385)
        arr = build(D + fn, sym, (rt, rc))
        np.savez_compressed(out, bars=arr, columns=np.array(COLS))
        print(f"  saved {out}")