"""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}")