foxhunt/scripts/measure_passive_mm_queue.py

#!/usr/bin/env python3
"""Queue-aware passive market-making fill model on ES MBP-10 — the decisive
viability test (does the passive edge survive realistic, adversely-biased fills?).

The optimistic harness (measure_passive_mm.py) assumed we win every queue and got
break-even / a Q2 OFI-conditioned tilt. The whole result rides on that assumption.
This model adds FIFO queue priority:

  * Segment each side into PRICE-LEVEL EPISODES (maximal runs of constant best
    bid/ask price).
  * At each episode start we "join the back of the queue": queue_ahead = touch size.
  * Only TRADES advance us (cancels assumed behind us — the conservative FIFO
    backtest standard). We FILL when cumulative same-side aggressor volume during
    the episode >= queue_ahead.
  * Episode end:
      - price moves AWAY favorably without filling  → MISS (we missed that move)
      - price moves through our side via trades       → FILL (adverse: we bought as
                                                          it fell / sold as it rose)
  * The fills we WIN are concentrated in heavy-same-side-flow episodes = adverse.

Then re-measure markout + OFI-conditioned selective quoting on the WON fills, and
report the realistic fill rate. If the OFI-conditioned edge survives here across
quarters, passive MM is viable; if it evaporates, it was a win-every-queue artifact.

Usage:
  python3 scripts/measure_passive_mm_queue.py <file.dbn.zst> [--n-cap N] [--fee-ticks F] [--ofi-window-sec W]
  python3 scripts/measure_passive_mm_queue.py --self-test
"""
import argparse
import sys
import time

import numpy as np

TICK = 0.25
TICK_USD = 12.50
PX_SCALE = 1e9


def compute_ofi(bid_px, bid_sz, ask_px, ask_sz):
    pb_n, pb_p = bid_px[1:], bid_px[:-1]
    qb_n, qb_p = bid_sz[1:].astype(np.float64), bid_sz[:-1].astype(np.float64)
    pa_n, pa_p = ask_px[1:], ask_px[:-1]
    qa_n, qa_p = ask_sz[1:].astype(np.float64), ask_sz[:-1].astype(np.float64)
    e_bid = qb_n * (pb_n >= pb_p) - qb_p * (pb_n <= pb_p)
    e_ask = qa_p * (pa_n >= pa_p) - qa_n * (pa_n <= pa_p)
    return e_bid + e_ask


def episode_fills(level_px, touch_sz, consume_vol):
    """FIFO queue fills per price-level episode.

    level_px[i]   : best price on our side at event i (int fixed-point)
    touch_sz[i]   : size resting at the touch at event i (queue we join behind)
    consume_vol[i]: same-side aggressor trade volume at event i (advances queue)

    Returns boolean fill mask (True at the event where our episode-start order fills).
    One hypothetical order placed at each episode start (join back of queue).
    """
    n = len(level_px)
    chg = np.empty(n, bool); chg[0] = True
    chg[1:] = level_px[1:] != level_px[:-1]
    ep = np.cumsum(chg) - 1                                   # 0-based episode id per event
    ep_start = np.flatnonzero(chg)                            # first event index of each episode
    queue0_ep = touch_sz[ep_start].astype(np.float64)        # queue we join behind
    cs = np.cumsum(consume_vol.astype(np.float64))
    base = cs[ep_start] - consume_vol[ep_start]               # cumsum just BEFORE episode start
    within = cs - base[ep]                                    # same-side volume consumed since episode start (incl. current)
    q0_evt = queue0_ep[ep]
    crossed = (q0_evt > 0) & (within >= q0_evt)
    prev_crossed = np.r_[False, crossed[:-1]]
    same_ep = np.r_[False, ep[1:] == ep[:-1]]
    fill = crossed & ~(prev_crossed & same_ep)               # first crossing within each episode
    return fill, ep, ep_start, chg


def self_test():
    # one bid-level episode at 100 with queue0=5; sells of 3 then 3 → fill at idx2.
    bid = np.array([100, 100, 100, 100, 99], dtype=np.int64)
    bsz = np.array([5, 5, 5, 5, 4], dtype=np.int64)
    sell = np.array([0, 3, 3, 0, 0], dtype=np.float64)       # sell-aggressor vol at bid
    fill, ep, ep_start, chg = episode_fills(bid, bsz, sell)
    assert fill.tolist() == [False, False, True, False, False], fill.tolist()
    # second episode (99) has no sells → no fill
    assert ep.tolist() == [0, 0, 0, 0, 1], ep.tolist()
    print("queue-fill self-test PASS: fill at idx2 (cum sell 6 >= queue0 5)")
    return 0


def decode(path, n_cap):
    import zstandard
    import databento_dbn as d
    ts = np.zeros(n_cap, np.int64)
    bpx = np.zeros(n_cap, np.int64); apx = np.zeros(n_cap, np.int64)
    bsz = np.zeros(n_cap, np.int64); asz = np.zeros(n_cap, np.int64)
    inst = np.zeros(n_cap, np.int32)
    agg = np.zeros(n_cap, np.int8)          # +1 buy-agg (lifts ask), -1 sell-agg (hits bid)
    tsz = np.zeros(n_cap, np.float64)       # trade size (0 for non-trades)
    dec = d.DBNDecoder(); dctx = zstandard.ZstdDecompressor()
    i = 0; t0 = time.time(); done = False
    with open(path, "rb") as fh:
        reader = dctx.stream_reader(fh)
        while not done:
            chunk = reader.read(1 << 22)
            if not chunk:
                break
            dec.write(chunk)
            for rec in dec.decode():
                if "MBP10" not in type(rec).__name__:
                    continue
                l0 = rec.levels[0]
                ts[i] = rec.ts_event
                bpx[i] = l0.bid_px; apx[i] = l0.ask_px
                bsz[i] = l0.bid_sz; asz[i] = l0.ask_sz
                inst[i] = rec.instrument_id
                a = str(rec.action); s = str(rec.side)
                if a.endswith("TRADE") or a == "T":
                    if s.endswith("BID") or s == "B":
                        agg[i] = 1; tsz[i] = rec.size
                    elif s.endswith("ASK") or s == "A":
                        agg[i] = -1; tsz[i] = rec.size
                i += 1
                if i >= n_cap:
                    done = True
                    break
    return ts[:i], bpx[:i], apx[:i], bsz[:i], asz[:i], inst[:i], agg[:i], tsz[:i], time.time() - t0


def markout_table(side_name, fi, pos_sign, fill_px, mid, ts, ofi_w, fee_ticks, n_episodes):
    mid_fill = mid[fi]
    half = pos_sign * (mid_fill - fill_px) / TICK
    aligned = np.sign(ofi_w[fi]) == pos_sign
    print(f"\n[{side_name}] won fills={len(fi)}  fill_rate={100*len(fi)/max(n_episodes,1):.1f}% of episodes  "
          f"half-spread={half.mean():.3f}t  OFI-aligned={100*aligned.mean():.1f}%")
    print(f"{'Δ(s)':>6} {'markout':>9} {'net_all':>9} {'net_$':>8} {'net_aligned':>12} {'net_adverse':>12} {'sel_$/fill':>11}")
    for d_s in [0.1, 1.0, 5.0, 10.0, 60.0]:
        idx2 = np.clip(np.searchsorted(ts, ts[fi] + int(d_s * 1e9), side="left"), 0, len(mid) - 1)
        markout = pos_sign * (mid[idx2] - mid_fill) / TICK
        net = half + markout - fee_ticks
        na = net[aligned].mean() if aligned.any() else float("nan")
        nadv = net[~aligned].mean() if (~aligned).any() else float("nan")
        sel = (na * TICK_USD) if aligned.any() else float("nan")
        print(f"{d_s:>6} {markout.mean():>9.4f} {net.mean():>9.4f} {net.mean()*TICK_USD:>7.2f}$ "
              f"{na:>12.4f} {nadv:>12.4f} {sel:>10.2f}$")


def analyze(path, n_cap, fee_ticks, ofi_window_sec):
    ts, bpx, apx, bsz, asz, inst, agg, tsz, dt = decode(path, n_cap)
    n = len(ts)
    ids, cnts = np.unique(inst, return_counts=True)
    dom = ids[cnts.argmax()]
    m = inst == dom
    ts, bpx, apx, bsz, asz, agg, tsz = (a[m] for a in (ts, bpx, apx, bsz, asz, agg, tsz))
    valid = (bpx > 0) & (apx > 0) & (bpx < apx)
    ts, bpx, apx, bsz, asz, agg, tsz = (a[valid] for a in (ts, bpx, apx, bsz, asz, agg, tsz))
    mid = (bpx + apx) / 2.0 / PX_SCALE
    print(f"\n=== {path.split('/')[-1]} (QUEUE-AWARE) ===")
    print(f"decoded {n} events in {dt:.1f}s | dominant={dom} → {valid.sum()} clean front-month events")

    ofi = np.empty(len(ts)); ofi[0] = 0.0
    ofi[1:] = compute_ofi(bpx, bsz, apx, asz)
    cum_ofi = np.cumsum(ofi)
    w_ns = int(ofi_window_sec * 1e9)
    j0 = np.searchsorted(ts, ts - w_ns, side="left")
    ofi_w = cum_ofi - cum_ofi[j0]                            # trailing-window OFI per event

    # BUY side: episodes of constant bid_px; sell-aggressor (agg==-1) volume consumes bid queue
    sell_vol = np.where(agg == -1, tsz, 0.0)
    buy_fill, ep_b, ep_start_b, _ = episode_fills(bpx, bsz, sell_vol)
    fib = np.flatnonzero(buy_fill)
    markout_table("BUY (rest at bid)", fib, +1.0, bpx[fib] / PX_SCALE, mid, ts, ofi_w,
                  fee_ticks, len(ep_start_b))

    # SELL side: episodes of constant ask_px; buy-aggressor (agg==+1) volume consumes ask queue
    buy_vol = np.where(agg == 1, tsz, 0.0)
    sell_fill, ep_a, ep_start_a, _ = episode_fills(apx, asz, buy_vol)
    fia = np.flatnonzero(sell_fill)
    markout_table("SELL (rest at ask)", fia, -1.0, apx[fia] / PX_SCALE, mid, ts, ofi_w,
                  fee_ticks, len(ep_start_a))

    print(f"\nfee={fee_ticks}t/side (${fee_ticks*TICK_USD:.2f}); markout marked at mid (no flatten cost).")
    print("Compare to optimistic (measure_passive_mm.py): if markout here is MORE adverse and")
    print("net_aligned no longer > 0, the Q2 OFI edge was a win-every-queue artifact.")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("file", nargs="?")
    ap.add_argument("--n-cap", type=int, default=8_000_000)
    ap.add_argument("--fee-ticks", type=float, default=0.1)
    ap.add_argument("--ofi-window-sec", type=float, default=1.0)
    ap.add_argument("--self-test", action="store_true")
    args = ap.parse_args()
    if args.self_test:
        return self_test()
    if not args.file:
        ap.error("file required (or --self-test)")
    analyze(args.file, args.n_cap, args.fee_ticks, args.ofi_window_sec)
    return 0


if __name__ == "__main__":
    sys.exit(main())