"""BACKTEST the ratchet against real Kalshi trade prints, queue-aware. The ratchet was justified by replaying ONE day (34 books, 3 completions). That is an anecdote. This scores it properly over every book since depth logging began (2026-07-28), using the fill model from queue_aware_fill.py: a resting ask at price P fills only once cumulative volume traded at >= P exceeds the depth that was already queued at P when we posted. Three policies, same books, same trades: CHASE -- what the live loop did: re-quote at the touch every cycle, so the effective offer is the LOWEST ask seen while waiting. RATCHET -- never quote below the first offer; follow a rising touch up. GATE+R -- ratchet, and only open books whose entry touches clear target. The ratchet trades fill probability for margin, so the only question that matters is whether the books it still completes are worth more than the books it loses. That is exactly what this measures, and it is not answerable from the paper log, which records every fill landing at our quoted price. Bounds are reported both ways -- queue-aware (lower) and optimistic (upper) -- because the truth is between them and a single number here would be a lie. Usage: python research\\mm_ratchet_backtest.py [--refresh] """ import csv, json, os, sys, statistics from collections import defaultdict sys.path.insert(0, r'E:\soccer\main') sys.path.insert(0, r'E:\soccer\research') import mm_settle as S import kalshi_trader as kt import mm_maker_live as ML from queue_aware_fill import trades_sized QUOTES = r'E:\soccer\research\mm_paper_quotes.csv' CACHE = r'E:\soccer\research\.trades_cache.json' DEPTH_START = '2026-07-28' def load_books(): """{event: {leg: [rows sorted by ts]}} for legs with depth logged.""" legs = defaultdict(list) with open(QUOTES, encoding='utf-8', errors='replace') as fh: for r in csv.DictReader(fh): t = r.get('leg_ticker') if not t or r['ts'][:10] < DEPTH_START: continue if not (r.get('k_ask') and r.get('kickoff')): continue legs[t].append(r) books = defaultdict(dict) for t, rs in legs.items(): rs.sort(key=lambda x: x['ts']) if not (rs[0].get('ask_size') or '').strip(): continue # before depth logging ev, leg = t.rsplit('-', 1) books[ev][leg] = rs return {e: b for e, b in books.items() if len(b) == 3} def _f(x, d=0.0): try: return float(x) except (TypeError, ValueError): return d def get_trades(ticker, since, cache): key = f'{ticker}|{since.isoformat()}' if key in cache: return [(S._pt(t), p, c) for t, p, c in cache[key]] tr = trades_sized(ticker, since) cache[key] = [(t.isoformat(), p, c) for t, p, c in tr] return tr def fills(trades, price, queue, ko): """(queue_aware, optimistic) -- would a resting ask at `price` fill?""" opt = any(p >= price - 1e-9 for t, p, _ in trades if t < ko) cum = 0.0 for t, p, c in trades: if t >= ko: break if p >= price - 1e-9: cum += c if cum > queue: return True, opt return False, opt def main(): refresh = '--refresh' in sys.argv cache = {} if os.path.exists(CACHE) and not refresh: try: cache = json.load(open(CACHE, encoding='utf-8')) except Exception: cache = {} books = load_books() print(f'{len(books)} three-leg books since {DEPTH_START}\n') results = {k: [] for k in ('chase', 'ratchet', 'gate')} gated = 0 for i, (ev, legs) in enumerate(sorted(books.items()), 1): if i % 25 == 0: print(f' ...{i}/{len(books)}', flush=True) entry, ko_ok = {}, True for leg, rs in legs.items(): r0 = rs[0] ko = S._pt(r0['kickoff']) if not ko: ko_ok = False break entry[leg] = {'ask0': _f(r0['k_ask']), 'low': min(_f(r['k_ask']) for r in rs if _f(r['k_ask'])), 'queue': _f(r0.get('ask_size')), 'start': S._pt(r0['ts']), 'ko': ko, 'ticker': f'{ev}-{leg}'} if not ko_ok or any(v['ask0'] <= 0 for v in entry.values()): continue asks0 = {l: v['ask0'] for l, v in entry.items()} decision = ML.book_budget(asks0, {})['decision'] per_leg = {} for leg, v in entry.items(): tr = get_trades(v['ticker'], v['start'], cache) # CHASE fills at the lowest ask seen; RATCHET holds the first ask. per_leg[leg] = { 'chase': fills(tr, v['low'], v['queue'], v['ko']), 'ratchet': fills(tr, v['ask0'], v['queue'], v['ko']), 'px_chase': v['low'], 'px_ratchet': v['ask0']} for pol in ('chase', 'ratchet'): src = 'chase' if pol == 'chase' else 'ratchet' q = all(per_leg[l][src][0] for l in per_leg) o = all(per_leg[l][src][1] for l in per_leg) coll = sum(per_leg[l][f'px_{src}'] for l in per_leg) results[pol].append({'q': q, 'o': o, 'coll': coll}) if decision == 'gate': gated += 1 else: q = all(per_leg[l]['ratchet'][0] for l in per_leg) o = all(per_leg[l]['ratchet'][1] for l in per_leg) coll = sum(per_leg[l]['px_ratchet'] for l in per_leg) results['gate'].append({'q': q, 'o': o, 'coll': coll}) json.dump(cache, open(CACHE, 'w', encoding='utf-8')) def score(rows, key): done = [r for r in rows if r[key]] if not done: return None net = [] for r in done: fee = sum(ML.maker_fee(r['coll'] / 3.0) for _ in range(3)) net.append(r['coll'] - fee - 1.0) return {'n': len(rows), 'done': len(done), 'rate': 100.0 * len(done) / len(rows), 'coll': statistics.median(r['coll'] for r in done), 'mean': statistics.mean(net) * 100, 'total': sum(net) * 100, 'winrate': 100.0 * sum(1 for x in net if x > 0) / len(net)} for key, lbl in (('q', 'QUEUE-AWARE (lower bound)'), ('o', 'OPTIMISTIC (upper bound)')): print(f'\n=== {lbl} ===') print(f'{"policy":10} {"books":>6} {"complete":>9} {"rate":>7} ' f'{"med coll":>9} {"net/book":>9} {"profitable":>11} {"total":>9}') for pol, name in (('chase', 'chase'), ('ratchet', 'ratchet'), ('gate', 'gate+ratch')): s = score(results[pol], key) if not s: print(f'{name:10} {len(results[pol]):>6} {"0":>9} — no completions') continue print(f'{name:10} {s["n"]:>6} {s["done"]:>9} {s["rate"]:>6.1f}% ' f'{s["coll"]:>9.4f} {s["mean"]:>8.2f}c {s["winrate"]:>10.0f}% ' f'{s["total"]:>8.1f}c') print(f'\ngate refused {gated} of {len(books)} books before any leg was offered') print('\nThe ratchet is only worth it if its net/book gain outweighs the') print('completion it gives up -- compare the TOTAL column, which already') print('carries both effects.') if __name__ == '__main__': main()