""" FORWARD TEST (read-only, trades nothing): honest-de-vig band [0.07,0.20) vs the current live band [0.05,0.12), on genuinely out-of-sample fixtures. WHY: the band sweep (research/devig_band_sweep.py, 2026-06-27) found that under POWER de-vig (which de-biases Pinnacle's longshot over-statement), the signal lives at honest CLV >= 0.07, not 0.05 -- the marginal [0.05,0.07) bets pay vig with zero edge (WR==implied) and drag the recent half. A [0.07,0.20) band was +20% ROIturn and positive in BOTH temporal halves, while the current [0.05,0.12) band's edge had reverted (test-half +0.9%). BUT that band was selected in-sample over 12 candidates (p=0.046 -> ~0.55 Bonferroni), so it is a HYPOTHESIS, not proof. This script freezes the band choice as of FREEZE_DATE and tracks both bands on later fixtures so we get clean OOS evidence before any live change. NOTHING is traded. Selection (both bands, NON-look-ahead): first pre-KO snapshot whose mid-CLV is in the band; both-sides dedupe to the higher-CLV side; bet at that snapshot's ask; settle by ftr. The de-vig differs per arm: - 'live' : multiplicative pin_fair, band [0.05,0.12) (what the pipeline ships) - 'honest': power pin_fair, band [0.07,0.20) (the candidate) Appends one dated row per run to forward_test_devig_band_log.csv. """ import csv, glob, os, math from datetime import datetime, date from collections import defaultdict LIVE_DIR = r"E:\soccer\live_data" LOG_CSV = r"E:\soccer\research\forward_test_devig_band_log.csv" FREEZE_DATE = "2026-06-27" # band chosen on data <= this; game_date > this = OOS ARMS = { # name -> (devig, lo, hi) 'live': ('mult', 0.05, 0.12), 'honest': ('power', 0.07, 0.20), } _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 fair_power(o): r = [1.0/x for x in o] lo, hi = 0.5, 10.0 for _ in range(60): k = (lo+hi)/2 if sum(ri**k for ri in r) > 1.0: lo = k else: hi = k k = (lo+hi)/2 p = [ri**k for ri in r]; s = sum(p) return [pi/s for pi in p] _DEVIG = {'mult': fair_mult, 'power': fair_power} def load_fixtures(): fixtures = defaultdict(list) for path in glob.glob(os.path.join(LIVE_DIR, "*.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) prepped = [] for (gd, league, home, away), rows in fixtures.items(): 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 len(pre) < 2: continue ftr = _result(rows) if ftr is None: continue prepped.append(dict(gd=gd, league=league, ftr=ftr, pre=[r for _, r in pre])) prepped.sort(key=lambda x: x['gd']) return prepped def select(prepped, devig, lo, hi): fn = _DEVIG[devig]; bets = [] for fx in prepped: cands = [] 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 mid = _f(r.get(f'kalshi_{side}_mid')); ask = _f(r.get(f'kalshi_{side}_ask')) if mid is None or ask is None or not (0 < ask < 1): continue pf = fn((ph, pd_, pa))[sidx] clv = (pf - mid) / pf if pf else None if clv is None or not (lo <= clv < hi): continue won = (fx['ftr'] == 'H' and side == 'home') or (fx['ftr'] == 'A' and side == 'away') cands.append((clv, ask, won)); break if cands: clv, ask, won = max(cands, key=lambda t: t[0]) bets.append(dict(gd=fx['gd'], ask=ask, won=won)) bets.sort(key=lambda b: b['gd']) return bets def stat(bets): n = len(bets) if not n: return None wins = sum(b['won'] for b in bets) net = sum(((1-b['ask'])/b['ask']) if b['won'] else -1.0 for b in bets) mu = sum(b['ask'] for b in bets); var = sum(b['ask']*(1-b['ask']) for b in bets) z = (wins-mu)/math.sqrt(var) if var > 0 else 0.0 return dict(n=n, wr=wins/n*100, iwr=mu/n*100, roi=net/n*100, z=z) def log_snapshot(prepped=None): if prepped is None: prepped = load_fixtures() row = {'run_date': date.today().isoformat(), 'freeze_date': FREEZE_DATE} for name, (dv, lo, hi) in ARMS.items(): oos = [fx for fx in prepped if fx['gd'] > FREEZE_DATE] s = stat(select(oos, dv, lo, hi)) if oos else None row[f'{name}_n'] = s['n'] if s else 0 row[f'{name}_wr'] = round(s['wr'], 1) if s else '' row[f'{name}_roi'] = round(s['roi'], 1) if s else '' row[f'{name}_z'] = round(s['z'], 2) if s else '' new = not os.path.exists(LOG_CSV) with open(LOG_CSV, 'a', newline='', encoding='utf-8') as fh: w = csv.DictWriter(fh, fieldnames=list(row.keys())) if new: w.writeheader() w.writerow(row) return row def main(): prepped = load_fixtures() print(f"FREEZE_DATE={FREEZE_DATE} total resolved fixtures={len(prepped)}\n") for label, sub in (("IN-SAMPLE (<=freeze, reference)", [f for f in prepped if f['gd'] <= FREEZE_DATE]), ("FORWARD (OOS, >freeze)", [f for f in prepped if f['gd'] > FREEZE_DATE])): print(f"--- {label} ---") if not sub: print(" (no fixtures yet)\n"); continue for name, (dv, lo, hi) in ARMS.items(): s = stat(select(sub, dv, lo, hi)) if s: print(f" {name:7s} {dv:5s} [{lo:.2f},{hi:.2f}) n={s['n']:4d} WR={s['wr']:4.1f}% " f"impl={s['iwr']:4.1f}% ROIturn={s['roi']:+6.1f}% z={s['z']:+.2f}") else: print(f" {name:7s} (no bets)") print() log_snapshot(prepped) print(f"logged snapshot -> {LOG_CSV}") if __name__ == '__main__': main()