""" Read-only study: LEAD-LAG of Kalshi vs Pinnacle repricing, per league. Motivation (stale-Pinnacle theory): our edge = pin_fair - kalshi. If Kalshi reprices BEFORE our (delayed) Pinnacle feed, the measured edge is just Pinnacle's lag catching up to a price Kalshi already found -> illusory, no value to capture. If Pinnacle leads (or they move together), the edge reflects a real mispricing that persists long enough to trade. We need a per-league SPEED proxy better than raw Kalshi volume (volume only got us to corr p~=0.12). This measures who-moves-first directly from the snapshot series, with NO external data. Method (cross-correlation of one-step price changes, per fixture, pooled/league): For consecutive PRE snapshots build: dK_t = kalshi_mid(t) - kalshi_mid(t-1) dP_t = pin_fair(t) - pin_fair(t-1) Both home and away sides used as independent repricing observations. Cross-correlations of the increment series: rho(+1) = corr(dK_t, dP_{t+1}) Kalshi change predicts NEXT Pin change -> KALSHI LEADS rho(-1) = corr(dP_t, dK_{t+1}) Pin change predicts NEXT Kalshi change -> PIN LEADS rho(0) = corr(dK_t, dP_t) contemporaneous (same snapshot) Lead-lag score LL = rho(+1) - rho(-1). LL > 0 => Kalshi leads (bad for us). Then correlate league LL against: - beat-close% (reference-independent edge-quality screen, recomputed here) - realized Kalshi ROI from bet_history.json Hypothesis: Kalshi-leads (high LL) -> weaker edge (lower beat-close, lower ROI). NOTHING is traded. Reproducible / read-only. """ import csv, glob, os, json, math from datetime import datetime from collections import defaultdict LIVE_DIR = r"E:\soccer\live_data" HIST = r"E:\soccer\data\bet_history.json" THRESH_LO, THRESH_HI = 0.05, 0.12 # same signal band as live_monitor def f(x): try: return float(x) except (TypeError, ValueError): return None _FMTS = ['%Y-%m-%d %H:%M:%S', '%m/%d/%Y %H:%M', '%m/%d/%Y %H:%M:%S', '%Y-%m-%d %H:%M'] def parse_ts(s): for fmt in _FMTS: try: return datetime.strptime(s, fmt) except (ValueError, TypeError): pass return None def pearson(xs, ys): n = len(xs) if n < 3: return None, n mx, my = sum(xs)/n, sum(ys)/n sxy = sum((a-mx)*(b-my) for a, b in zip(xs, ys)) sxx = sum((a-mx)**2 for a in xs) syy = sum((b-my)**2 for b in ys) if sxx <= 0 or syy <= 0: return None, n return sxy/math.sqrt(sxx*syy), n # ---- load snapshots grouped by fixture ---- 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): key = (r.get('game_date'), r.get('league'), r.get('home'), r.get('away')) fixtures[key].append(r) # per-league increment-pair accumulators for the three lags lagp = defaultdict(lambda: {'p1':[[],[]], 'm1':[[],[]], 'c0':[[],[]]}) # beat-close accumulators bc = defaultdict(lambda: {'n':0, 'beat':0, 'move':0.0}) def result(rows): for r in reversed(rows): ftr = (r.get('ftr') or '').upper().strip() if ftr in ('H','D','A'): return ftr return None for key, rows in fixtures.items(): gd, league, home, away = key pre = sorted([(parse_ts(r.get('timestamp')), r) for r in rows if (r.get('status') or '').upper()=='PRE' and parse_ts(r.get('timestamp')) is not None], key=lambda t: t[0]) if len(pre) < 3: continue # ---- lead-lag: build increment series for each side ---- for kmid, pfair in [('kalshi_home_mid','pin_fair_home'), ('kalshi_away_mid','pin_fair_away')]: dK, dP = [], [] prevk = prevp = None for _, r in pre: kv, pv = f(r.get(kmid)), f(r.get(pfair)) if kv is None or pv is None: prevk = prevp = None continue if prevk is not None: dK.append(kv - prevk); dP.append(pv - prevp) else: # marker: keep alignment by appending None-break handled below pass prevk, prevp = kv, pv # build lag pairs within this contiguous increment run acc = lagp[league] for i in range(len(dK)): acc['c0'][0].append(dK[i]); acc['c0'][1].append(dP[i]) if i+1 < len(dK): acc['p1'][0].append(dK[i]); acc['p1'][1].append(dP[i+1]) # Kalshi leads acc['m1'][0].append(dP[i]); acc['m1'][1].append(dK[i+1]) # Pin leads # ---- beat-close: signal entry -> last pre-KO mid drift (mid-to-mid) ---- for side, clvk, midk in [('A','clv_kalshi_away','kalshi_away_mid'), ('H','clv_kalshi_home','kalshi_home_mid')]: ent = None for ts, r in pre: clv = f(r.get(clvk)) if clv is not None and THRESH_LO <= clv < THRESH_HI: ent = (ts, r); break if not ent: continue m0 = f(ent[1].get(midk)) mlast = None for ts, r in reversed(pre): mv = f(r.get(midk)) if mv is not None: mlast = mv; break if m0 is None or mlast is None: continue move = (mlast - m0) # mid rising = market moving toward a YES on this side bc[league]['n'] += 1 bc[league]['beat'] += 1 if move > 0 else 0 bc[league]['move'] += move break # one bet per fixture (first side that fires) # ---- realized ROI per league from bet_history.json ---- roi = {} try: with open(HIST, encoding='utf-8') as fh: hist = json.load(fh) agg = defaultdict(lambda: [0.0, 0.0]) for b in hist: st, pl = b.get('kalshi_stake'), b.get('kalshi_pnl') lg = b.get('league') or b.get('div') if st and pl is not None and lg: agg[lg][0] += st; agg[lg][1] += pl roi = {lg:(v[1]/v[0]*100 if v[0] else None) for lg, v in agg.items()} except Exception: pass # ---- compute per-league LL + assemble table ---- print("="*92) print("LEAD-LAG of Kalshi vs Pinnacle repricing (LL>0 => Kalshi LEADS => edge may be stale-Pin)") print("="*92) print(f"{'League':14}{'incN':>6}{'rho0':>8}{'rho+1':>8}{'rho-1':>8}{'LL':>8}" f"{'beatN':>7}{'beat%':>7}{'ROI%':>8}") table = [] for lg in sorted(lagp): acc = lagp[lg] r0,_ = pearson(acc['c0'][0], acc['c0'][1]) rp,np_ = pearson(acc['p1'][0], acc['p1'][1]) rm,_ = pearson(acc['m1'][0], acc['m1'][1]) if rp is None or rm is None: continue LL = rp - rm b = bc.get(lg, {'n':0,'beat':0}) bpct = (b['beat']/b['n']*100) if b['n'] else None rv = roi.get(lg) table.append((lg, np_, r0, rp, rm, LL, b['n'], bpct, rv)) # print sorted by LL descending (most Kalshi-leading first) for lg, np_, r0, rp, rm, LL, bn, bpct, rv in sorted(table, key=lambda t: -t[5]): bs = f"{bpct:6.1f}" if bpct is not None else " na" rs = f"{rv:+7.1f}" if rv is not None else " na" print(f"{lg:14}{np_:6d}{r0:8.3f}{rp:8.3f}{rm:8.3f}{LL:8.3f}{bn:7d}{bs}{rs}") # ---- correlate LL against edge metrics ---- def t_to_p(t, df): """two-sided p-value from a t-stat via regularized incomplete beta (no scipy).""" if df <= 0: return None x = df / (df + t*t) # incomplete beta I_x(df/2, 1/2) via continued fraction a, b = df/2.0, 0.5 def betacf(x, a, b): MAXIT, EPS, FPMIN = 200, 3e-12, 1e-300 qab, qap, qam = a+b, a+1.0, a-1.0 c = 1.0; d = 1.0 - qab*x/qap if abs(d) < FPMIN: d = FPMIN d = 1.0/d; h = d for m in range(1, MAXIT): m2 = 2*m aa = m*(b-m)*x/((qam+m2)*(a+m2)) d = 1.0 + aa*d if abs(d) < FPMIN: d = FPMIN c = 1.0 + aa/c if abs(c) < FPMIN: c = FPMIN d = 1.0/d; h *= d*c aa = -(a+m)*(qab+m)*x/((a+m2)*(qap+m2)) d = 1.0 + aa*d if abs(d) < FPMIN: d = FPMIN c = 1.0 + aa/c if abs(c) < FPMIN: c = FPMIN d = 1.0/d; de = d*c; h *= de if abs(de-1.0) < EPS: break return h lbeta = math.lgamma(a)+math.lgamma(b)-math.lgamma(a+b) if x < (a+1.0)/(a+b+2.0): ib = math.exp(a*math.log(x)+b*math.log(1-x)-lbeta)*betacf(x,a,b)/a else: ib = 1.0 - math.exp(a*math.log(x)+b*math.log(1-x)-lbeta)*betacf(1-x,b,a)/b return min(1.0, ib) # two-sided def corr_pairs(getx, gety, minn=8, drop=()): xs, ys = [], [] for row in table: lg, np_, r0, rp, rm, LL, bn, bpct, rv = row if bn < minn or lg in drop: continue x, y = getx(row), gety(row) if x is None or y is None: continue xs.append(x); ys.append(y) r, n = pearson(xs, ys) p = None if r is not None and n > 2 and abs(r) < 1: t = r*math.sqrt((n-2)/(1-r*r)); p = t_to_p(t, n-2) return r, n, p def fmt(r): return f"{r:+.3f}" if r is not None else " na" def fp(p): return f"{p:.3f}" if p is not None else "na" print("\n" + "-"*92) print("Does Kalshi-leading predict weaker edge? (leagues with >=8 settled bets)") for label, drop in [("ALL leagues", ()), ("EXCL E0", ('E0',))]: r_bc, n_bc, p_bc = corr_pairs(lambda t: t[5], lambda t: t[7], drop=drop) # LL vs beat% r_roi, n_roi, p_roi = corr_pairs(lambda t: t[5], lambda t: t[8], drop=drop) # LL vs ROI print(f"\n {label}") print(f" corr(LL, beat-close%) r={fmt(r_bc)} p={fp(p_bc)} n={n_bc} (expect NEG)") print(f" corr(LL, realized ROI) r={fmt(r_roi)} p={fp(p_roi)} n={n_roi} (expect NEG)") print("\nE0 row is the acid test: if stale-Pin theory holds, E0 should have the HIGHEST LL.")