""" Per-league Pinnacle calibration test (read-only, trades nothing). THE QUESTION IT ANSWERS: "is a league's bad record a structural problem with that market, or just variance?" xlow measures CLV against Pinnacle fair. If Pinnacle is mis-calibrated in a league, CLV against it is meaningless there -- no amount of sample size on our BETS would reveal that, because the bets inherit the broken anchor. WHY IT'S POWERFUL: it uses EVERY fixture with Pinnacle odds + a result, not just the handful we bet. That's ~40x more data per league (BSA: 32 fixtures vs 12 bets), so calibration becomes testable long before bet-level ROI does. FIRST RUN (2026-07-25, n=1212 fixtures across 21+ leagues): - POOLED: Pinnacle implied 25.0% draws, actual 25.4%. Calibrated to within 0.4pp. The anchor the whole strategy rests on is sound -- this had never actually been verified before. - BSA (the 0-for-12 league that prompted the test): implied 47.6%H/26.7%D, actual 56.2%H/15.6%D. Biggest deviation in the table, and adverse to us since BSA bets skew AWAY. BUT home z=+0.98, draw z=-1.42 -- noise. - Every league shows large deviations at n=30-50 with RANDOMLY SIGNED direction (E0 +9.6pp draws, P1 +8.9, N1 +8.5 vs E2 -6.3, T1 -6.1, SC0 -4.7). Random signs at small n is the signature of noise; a real structural bias would be consistent across comparable leagues. - To detect BSA's observed 8.6pp home-calibration error at 95% you need ~130 BSA fixtures (~4x current, roughly a full season). HOW TO USE LATER: re-run when a league has accumulated materially more fixtures (or when any league's record looks structurally broken). A league whose deviation reaches |z| > 2 with n >= 100+ fixtures is a genuine calibration problem -> that league's CLV signal is not trustworthy and it should be dropped from the strategy (a real reason, unlike the hindsight league-culling that project_league_filter_study already killed). Usage: python research\\pin_calibration_by_league.py [min_n] """ import sys import csv import glob import math from collections import defaultdict sys.path.insert(0, r'E:\soccer\research') import kalshi_only_execution as K SOURCES = (r'E:\soccer\poly_backfill\*_polybackfill.csv', r'E:\soccer\live_data\*.csv') MIN_N = int(sys.argv[1]) if len(sys.argv) > 1 else 25 def load_fixtures(): """Every unique fixture with a result and valid Pinnacle 1X2 at the LAST pre-KO snapshot (closest to kickoff = sharpest line).""" out, seen = [], set() for pat in SOURCES: for path in glob.glob(pat): fx = defaultdict(list) try: with open(path, newline='', encoding='utf-8') as fh: for r in csv.DictReader(fh): fx[(r.get('game_date'), r.get('league'), r.get('home'), r.get('away'))].append(r) except Exception: continue for key, rows in fx.items(): if key in seen or not key[0]: continue ftr = K._result(rows) if ftr not in ('H', 'D', 'A'): continue pre = [r for r in rows if (r.get('status') or '').upper() == 'PRE'] if not pre: continue last = pre[-1] ph, pd_, pa = (K._f(last.get('pin_home')), K._f(last.get('pin_draw')), K._f(last.get('pin_away'))) if None in (ph, pd_, pa) or not (ph > 1 and pd_ > 1 and pa > 1): continue seen.add(key) pf = K.fair_mult((ph, pd_, pa)) out.append(dict(lg=key[1], ftr=ftr, pfh=pf[0], pfd=pf[1], pfa=pf[2])) return out def z_for(n, implied_p, actual_count): sd = math.sqrt(n * implied_p * (1 - implied_p)) return (actual_count - implied_p * n) / sd if sd > 0 else 0.0 def n_needed(effect_pp, base_p=0.476, conf=1.96): """Fixtures required to detect a calibration error of `effect_pp` (in pp).""" sd1 = math.sqrt(base_p * (1 - base_p)) return (conf * sd1 / (effect_pp / 100.0)) ** 2 def main(): allf = load_fixtures() by = defaultdict(list) for f in allf: by[f['lg']].append(f) print(f'Pinnacle calibration by league (last pre-KO snapshot), min_n={MIN_N}') print(f'total fixtures loaded: {len(allf)}\n') print('lg n impH actH zH impD actD zD impA actA zA flag') flagged = [] for lg, s in sorted(by.items(), key=lambda kv: -len(kv[1])): n = len(s) if n < MIN_N: continue iH = sum(x['pfh'] for x in s) / n iD = sum(x['pfd'] for x in s) / n iA = sum(x['pfa'] for x in s) / n cH = sum(x['ftr'] == 'H' for x in s) cD = sum(x['ftr'] == 'D' for x in s) cA = sum(x['ftr'] == 'A' for x in s) zH, zD, zA = z_for(n, iH, cH), z_for(n, iD, cD), z_for(n, iA, cA) worst = max(abs(zH), abs(zD), abs(zA)) flag = '' if worst > 2 and n >= 100: flag = 'STRUCTURAL?' flagged.append(lg) elif worst > 2: flag = 'watch (n low)' print(f'{lg:6s} {n:4d} {iH*100:5.1f} {cH/n*100:5.1f} {zH:+5.2f} ' f'{iD*100:5.1f} {cD/n*100:5.1f} {zD:+5.2f} ' f'{iA*100:5.1f} {cA/n*100:5.1f} {zA:+5.2f} {flag}') n = len(allf) iD = sum(x['pfd'] for x in allf) / n cD = sum(x['ftr'] == 'D' for x in allf) iH = sum(x['pfh'] for x in allf) / n cH = sum(x['ftr'] == 'H' for x in allf) print() print(f'POOLED n={n}: draws implied {iD*100:.1f}% actual {cD/n*100:.1f}% ' f'(z={z_for(n, iD, cD):+.2f}) | homes implied {iH*100:.1f}% ' f'actual {cH/n*100:.1f}% (z={z_for(n, iH, cH):+.2f})') print(' -> pooled calibration is the load-bearing check: if this is tight, ' 'the fair-value anchor is sound and per-league wobble is noise.') print() print('fixtures needed to detect a calibration error of given size (95%):') for pp in (5, 8.6, 10, 15): print(f' {pp:4.1f}pp -> n = {n_needed(pp):5.0f}') if flagged: print(f'\nFLAGGED (|z|>2 with n>=100): {flagged} -- these leagues\' CLV ' f'signal may not be trustworthy; investigate before trading them.') else: print('\nNo league flagged as structurally mis-calibrated.') if __name__ == '__main__': main()