""" OOS test of the "unique-bet" hypothesis: bets where exactly ONE book (Kalshi XOR Poly) fires a [0.05,0.12) signal backtested BETTER than bets where BOTH fire (Kalshi-only +63.9% z+2.10, Poly-only +59.6% z+2.89 vs both ~+8-11%). That was in-sample, hand-sliced, small n. Here we hold it out. The rule IS implementable at bet time: pre-KO you can see whether only one book is in-band (no look-ahead). So "bet only the one-book-only signals" is a real strategy. Protocol: restrict to SHARED-PRICE fixtures (both books quoted, fair test), order by game_date, 60/40 temporal split. Report TRAIN vs TEST for: - UNIQUE (Kalshi-only @ ask + Poly-only @ mid) <- the hypothesis - OVERLAP (both fired; Kalshi side @ ask) <- the baseline If the unique edge is real it should persist (stay +, ideally > overlap) in TEST. Caveat baked in: Poly-only priced at MID (no historical ask) = optimistic; also shown Poly-only @ est-ask (near-term half-spread) for honesty. Multiplicative de-vig. """ import csv, glob, os, math from datetime import datetime from collections import defaultdict DIR = r"E:\soccer\poly_backfill" BAND = (0.05, 0.12) SPLIT = 0.60 _FMTS = ['%Y-%m-%d %H:%M:%S', '%m/%d/%Y %H:%M', '%m/%d/%Y %H:%M:%S', '%Y-%m-%d %H:%M'] def _pts(s): for f in _FMTS: try: return datetime.strptime(s, f) except (ValueError, TypeError): pass return None def _f(x): try: return float(x) except (TypeError, ValueError): return None def _result(rows): for r in reversed(rows): v = (r.get('ftr') or '').upper().strip() if v in ('H', 'D', 'A'): return v return None def fair_mult(o): r = [1.0/x for x in o]; s = sum(r); return [ri/s for ri in r] def half_spread(mid): # near-term Poly sample (poly_spread_nearterm.py) if mid < 0.05: return 0.003 elif mid < 0.35: return 0.005 if mid < 0.25 else 0.025 return 0.020 def load(): fixtures = defaultdict(list) for path in glob.glob(os.path.join(DIR, "*_polybackfill.csv")): with open(path, newline='', encoding='utf-8') as fh: for r in csv.DictReader(fh): fixtures[(r.get('game_date'), r.get('league'), r.get('home'), r.get('away'))].append(r) fx = [] for key, rows in fixtures.items(): gd, league, home, away = key if not gd: continue pre = sorted([(_pts(r['timestamp']), r) for r in rows if (r.get('status') or '').upper() == 'PRE' and _pts(r['timestamp'])], key=lambda t: t[0]) if not pre: continue ftr = _result(rows) if ftr is None: continue fx.append(dict(gd=gd, league=league, ftr=ftr, pre=[r for _, r in pre])) fx.sort(key=lambda f: f['gd']) return fx def fire(fx, book): """book='kalshi' (settle @ ask) or 'poly' (settle @ mid). Returns bet dict + have_price.""" cands = []; have = False for side, sidx in (('home', 0), ('away', 2)): for r in fx['pre']: ph, pd_, pa = _f(r.get('pin_home')), _f(r.get('pin_draw')), _f(r.get('pin_away')) if None in (ph, pd_, pa) or not (ph > 1 and pd_ > 1 and pa > 1): continue if book == 'kalshi': mid = _f(r.get(f'kalshi_{side}_mid')); px = _f(r.get(f'kalshi_{side}_ask')) else: mid = _f(r.get(f'poly_{side}_price')); px = mid if mid is None or px is None or not (0 < mid < 1) or not (0 < px < 1): continue have = True pf = fair_mult((ph, pd_, pa))[sidx] clv = (pf - mid) / pf if pf else None if clv is None or not (BAND[0] <= clv < BAND[1]): continue won = (fx['ftr'] == 'H' and side == 'home') or (fx['ftr'] == 'A' and side == 'away') cands.append((clv, side, px, mid, won)); break if not cands: return (None, have) clv, side, px, mid, won = max(cands, key=lambda t: t[0]) return (dict(side=side, px=px, mid=mid, won=won, clv=clv), have) def stat(bets, price='px'): n = len(bets) if not n: return None wins = sum(b['won'] for b in bets) net = mu = var = 0.0 for b in bets: p = b[price] if price in b else b['px'] net += ((1-p)/p) if b['won'] else -1.0 mu += p; var += p*(1-p) z = (wins-mu)/math.sqrt(var) if var > 0 else 0.0 pval = 0.5*math.erfc(z/math.sqrt(2)) return dict(n=n, wr=wins/n*100, roi=net/n*100, z=z, p=pval) def line(tag, s): if not s: print(f" {tag:32s} (no bets)"); return print(f" {tag:32s} n={s['n']:3d} WR={s['wr']:4.1f}% ROIturn={s['roi']:+6.1f}% " f"z={s['z']:+.2f} p={s['p']:.3f}") def main(): fx = load() # classify each shared-price fixture recs = [] for f in fx: k, kh = fire(f, 'kalshi'); p, ph = fire(f, 'poly') if not (kh and ph): continue # shared-price only recs.append(dict(gd=f['gd'], k=k, p=p)) recs.sort(key=lambda r: r['gd']) cut = int(len(recs)*SPLIT) tr, te = recs[:cut], recs[cut:] print(f"shared-price fixtures={len(recs)} split @ {SPLIT:.0%}: " f"train={len(tr)} (<= {tr[-1]['gd']}) test={len(te)} (>= {te[0]['gd']})\n") def buckets(recs): konly = [r['k'] for r in recs if r['k'] and not r['p']] ponly = [r['p'] for r in recs if r['p'] and not r['k']] both_k = [r['k'] for r in recs if r['k'] and r['p']] # Poly-only at estimated ask ponly_ask = [dict(b, pxa=b['mid']+half_spread(b['mid'])) for b in ponly] return konly, ponly, ponly_ask, both_k for label, part in (("TRAIN (in-sample)", tr), ("TEST (held-out OOS)", te)): konly, ponly, ponly_ask, both_k = buckets(part) unique_pooled = konly + ponly # Kalshi@ask + Poly@mid print(f"== {label} ==") line("UNIQUE pooled (K@ask + P@mid)", stat(unique_pooled)) line(" Kalshi-only @ ask", stat(konly)) line(" Poly-only @ mid", stat(ponly)) line(" Poly-only @ est-ask", stat(ponly_ask, price='pxa')) line("OVERLAP both-fired (K@ask)", stat(both_k)) print() # whole-sample reference (what generated the hypothesis) konly, ponly, ponly_ask, both_k = buckets(recs) print("== FULL sample (hypothesis-generating; for reference) ==") line("UNIQUE pooled (K@ask + P@mid)", stat(konly + ponly)) line("OVERLAP both-fired (K@ask)", stat(both_k)) if __name__ == '__main__': main()