"""What predicts a book's volatility BEFORE we have to quote it? Pre-quote movement raises the queue-aware fill rate from 7.4% to 20.5% (mm_volatility_clean.py), so selecting volatile books is worth real money. That is only usable if volatility can be known in advance. Two ways it could be: PERSISTENCE -- volatility measured in an early window predicts volatility in a later one. If it holds, no model is needed: watch the book for an hour and let it select itself. FEATURES -- league, price level, liquidity, market age, or Pinnacle's own movement predict it from the start. Volatility is measured from live_data throughout, the feed the xlow/xhigh strategies run on, so nothing here depends on the maker's own logs. """ import csv, collections, os, statistics, sys, datetime as dt sys.path.insert(0, r'E:\soccer\main') sys.path.insert(0, r'E:\soccer\research') import mm_settle as S LIVE_DIR = r'E:\soccer\live_data' SIDES = {'home': 'kalshi_home_mid', 'draw': 'kalshi_draw_mid', 'away': 'kalshi_away_mid'} PIN = {'home': 'pin_home', 'draw': 'pin_draw', 'away': 'pin_away'} DAYS = 40 # most recent live_data files to scan def _f(x, lo=0.0, hi=1.0): try: v = float(x) return v if lo < v < hi else None except (TypeError, ValueError): return None def _odds(x): try: v = float(x) return v if 1.01 < v < 200 else None except (TypeError, ValueError): return None def load(): """[{league, side, series:[(ts, mid)], pins:[odds], ko}] per fixture-leg.""" files = sorted(f for f in os.listdir(LIVE_DIR) if f.endswith('.csv'))[-DAYS:] legs = collections.defaultdict(lambda: {'series': [], 'pins': []}) for fn in files: with open(os.path.join(LIVE_DIR, fn), encoding='utf-8', errors='replace') as fh: for r in csv.DictReader(fh): ts = S._pt(r.get('timestamp')) el = r.get('elapsed') if ts is None or el in (None, ''): continue try: if float(el) > 0: # PRE-kickoff snapshots only continue except ValueError: continue base = (r.get('game_date'), r.get('league'), r.get('home'), r.get('away')) for side, col in SIDES.items(): m = _f(r.get(col)) if m is None: continue d = legs[base + (side,)] d['series'].append((ts, m)) p = _odds(r.get(PIN[side])) if p: d['pins'].append(p) return legs def vol(series): return statistics.pstdev([m for _, m in series]) if len(series) > 2 else None def main(): legs = load() rows = [] for key, d in legs.items(): s = sorted(d['series']) if len(s) < 40: continue half = len(s) // 2 v_early, v_late = vol(s[:half]), vol(s[half:]) if v_early is None or v_late is None: continue mids = [m for _, m in s] pins = d['pins'] rows.append({ 'league': key[1], 'side': key[4], 'early': v_early, 'late': v_late, 'price': statistics.median(mids), 'n': len(s), 'pin_move': (statistics.pstdev([1.0 / p for p in pins]) if len(pins) > 2 else None), }) print(f'{len(rows)} fixture-legs with >=40 pre-kickoff snapshots\n') if len(rows) < 100: print('too few to analyse') return # ── 1. PERSISTENCE ────────────────────────────────────────────────── print('=== 1. Does early volatility predict later volatility? ===') e = sorted(r['early'] for r in rows) q = [e[int(p * len(e))] for p in (0.5, 0.75)] print(f' {"early-window bucket":>22} {"legs":>6} {"median LATE vol":>17}') for lo, hi, name in ((-1, q[0], 'calm half'), (q[0], q[1], '3rd quartile'), (q[1], 9, 'wildest 25%')): sel = [r for r in rows if lo <= r['early'] < hi] if len(sel) < 15: continue print(f' {name:>22} {len(sel):>6} ' f'{statistics.median(r["late"] for r in sel)*100:>16.2f}c') # rank correlation, no scipy ranks = sorted(range(len(rows)), key=lambda i: rows[i]['early']) re_ = {i: k for k, i in enumerate(ranks)} ranks2 = sorted(range(len(rows)), key=lambda i: rows[i]['late']) rl = {i: k for k, i in enumerate(ranks2)} n = len(rows) d2 = sum((re_[i] - rl[i]) ** 2 for i in range(n)) rho = 1 - 6 * d2 / (n * (n * n - 1)) print(f'\n Spearman rho(early, late) = {rho:+.3f} ' f'({"strong" if abs(rho)>0.5 else "moderate" if abs(rho)>0.3 else "weak"})') # ── 2. STATIC FEATURES ────────────────────────────────────────────── print('\n=== 2. What is volatile, from the start? ===') by_lg = collections.defaultdict(list) for r in rows: by_lg[r['league']].append(r['early'] + r['late']) lgs = [(lg, statistics.median(v), len(v)) for lg, v in by_lg.items() if len(v) >= 12] lgs.sort(key=lambda x: -x[1]) print(f' {"league":>10} {"legs":>6} {"median vol":>12}') for lg, m, k in lgs[:6]: print(f' {lg:>10} {k:>6} {m*100:>11.2f}c') print(' ...') for lg, m, k in lgs[-4:]: print(f' {lg:>10} {k:>6} {m*100:>11.2f}c') if lgs: print(f' -> spread across leagues: {lgs[0][1]*100:.2f}c vs ' f'{lgs[-1][1]*100:.2f}c ({lgs[0][1]/max(1e-9,lgs[-1][1]):.1f}x)') print('\n by price level (a leg near 50c has more room to move):') for lo, hi in ((0, .15), (.15, .30), (.30, .45), (.45, .70), (.70, 1.0)): sel = [r for r in rows if lo <= r['price'] < hi] if len(sel) < 15: continue print(f' {lo:.2f}-{hi:.2f} {len(sel):>5} legs ' f'median vol {statistics.median(r["early"]+r["late"] for r in sel)*100:>6.2f}c') pm = [r for r in rows if r['pin_move'] is not None] if len(pm) > 80: print('\n by PINNACLE movement (does the sharp book moving predict ' 'Kalshi moving?):') p = sorted(r['pin_move'] for r in pm) pq = [p[int(x * len(p))] for x in (0.5, 0.75)] for lo, hi, name in ((-1, pq[0], 'pin calm half'), (pq[0], pq[1], 'pin 3rd q'), (pq[1], 9, 'pin wildest 25%')): sel = [r for r in pm if lo <= r['pin_move'] < hi] if len(sel) < 15: continue print(f' {name:>16} {len(sel):>5} legs kalshi vol ' f'{statistics.median(r["early"]+r["late"] for r in sel)*100:>6.2f}c') if __name__ == '__main__': main()