""" Read-only study: does requiring the Kalshi edge to PERSIST 15/30/60 min after it first appears improve realized (paper) profit vs entering immediately? Mechanism under test (user, 2026-06-14): our Pinnacle feed is stale, so an edge measured as Pin_fair - Kalshi can be illusory in fast markets (Kalshi already moved). If the edge is still open N minutes later, it's more likely a real, durable mispricing rather than feed lag. Replays each fixture's 15-min snapshots from E:/soccer/live_data/*.csv. Signal gate matches live_monitor exactly: 0.05 <= Kalshi CLV < 0.12, Kelly 0.25x off the ask. Outcome from the FT score in the same CSV. Non-compounding $1000 bankroll so stakes are comparable across strategies. NOTHING is traded. """ import csv, glob, os from datetime import datetime from collections import defaultdict LIVE_DIR = r"E:\soccer\live_data" THRESH_LO, THRESH_HI = 0.05, 0.12 KELLY_MULT, MAX_BET_PCT, BANKROLL = 0.25, 0.25, 1000.0 DELAYS = [15, 30, 60] # minutes TOL = 9 # snapshot match tolerance (snapshots ~15 min apart) def f(x): try: return float(x) except (TypeError, ValueError): return None def kelly_stake(pf, ask): if not (pf and ask and 0 < ask < 1): return None odds = 1/ask; b = odds-1 fr = (pf*b - (1-pf))/b if fr <= 0: return None return max(round(min(fr*KELLY_MULT, MAX_BET_PCT)*BANKROLL, 2), 1.0) # ---- load all snapshots, grouped by fixture ---- fixtures = defaultdict(list) # (date,league,home,away) -> list of row dicts 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) _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(r): s = r.get('timestamp') or '' for fmt in _FMTS: try: return datetime.strptime(s, fmt) except ValueError: pass return None def result(rows): """Final H/D/A: prefer the 'ftr' column, else derive from FT goals.""" for r in reversed(rows): ftr = (r.get('ftr') or '').upper().strip() if ftr in ('H','D','A'): return ftr for r in reversed(rows): st = (r.get('status') or '').upper() hg, ag = f(r.get('home_goals')), f(r.get('away_goals')) if st in ('FT','AET','PEN') and hg is not None and ag is not None: return 'H' if hg>ag else ('A' if ag>hg else 'D') return None # strategy accumulators strat = {k: {'n':0,'win':0,'stake':0.0,'pnl':0.0} for k in ['immediate','p15','p30','p60']} by_league = defaultdict(lambda: {k: {'n':0,'win':0,'stake':0.0,'pnl':0.0} for k in ['immediate','p15','p30','p60']}) skipped_closed = defaultdict(int) # how many bets each delay drops actual_gaps = defaultdict(list) # realized minutes-later actually used per bucket def pnl(side, res, ask, stake): won = (side=='A' and res=='A') or (side=='H' and res=='H') return (stake*(1/ask - 1) if won else -stake), won for key, rows in fixtures.items(): gd, league, home, away = key res = result(rows) if res is None: continue pre = sorted([r for r in rows if (r.get('status') or '').upper()=='PRE' and parse_ts(r) is not None], key=parse_ts) if not pre: continue # evaluate each side; take the FIRST side that fires (first-signal rule) first = None # (t0_idx, side, clv) for i, r in enumerate(pre): for side, clvk, askk in [('A','clv_kalshi_away','kalshi_away_ask'), ('H','clv_kalshi_home','kalshi_home_ask')]: clv = f(r.get(clvk)) if clv is not None and THRESH_LO <= clv < THRESH_HI: first = (i, side, askk); break if first: break if not first: continue i0, side, askk = first t0 = parse_ts(pre[i0]) ask0 = f(pre[i0].get(askk)); pfk = f(pre[i0].get('pin_fair_away' if side=='A' else 'pin_fair_home')) st0 = kelly_stake(pfk, ask0) if not st0: continue # immediate entry p,w = pnl(side, res, ask0, st0) for acc in (strat['immediate'], by_league[league]['immediate']): acc['n']+=1; acc['win']+=w; acc['stake']+=st0; acc['pnl']+=p # delayed-confirmation entries: first PRE snapshot at least d minutes later for d, name in zip(DELAYS, ['p15','p30','p60']): target = t0.timestamp() + d*60 cand, cand_ts = None, None for r in pre[i0+1:]: ts = parse_ts(r) if ts and ts.timestamp() >= target: cand, cand_ts = r, ts; break if cand is None: skipped_closed[name+'_noobs'] += 1 # no later snapshot before KO -> can't confirm -> skip continue clv_now = f(cand.get('clv_kalshi_away' if side=='A' else 'clv_kalshi_home')) if clv_now is None or not (THRESH_LO <= clv_now < THRESH_HI): skipped_closed[name] += 1 # edge closed/changed band -> skip (the filter's whole point) continue actual_gaps[name].append((cand_ts - t0).total_seconds()/60) askd = f(cand.get(askk)); pfd = f(cand.get('pin_fair_away' if side=='A' else 'pin_fair_home')) std = kelly_stake(pfd, askd) if not std: continue p,w = pnl(side, res, askd, std) for acc in (strat[name], by_league[league][name]): acc['n']+=1; acc['win']+=w; acc['stake']+=std; acc['pnl']+=p def line(label, a): if a['n']==0: return f" {label:11} n=0" roi = a['pnl']/a['stake']*100 if a['stake'] else 0 return (f" {label:11} n={a['n']:3d} win={a['win']/a['n']*100:4.1f}% " f"stake=${a['stake']:7.2f} pnl={a['pnl']:+8.2f} ROI={roi:+6.1f}%") print("="*78) print("KALSHI EDGE-PERSISTENCE STUDY (entry only if edge still in [5%,12%) after N min)") print(f"snapshot files: {len(glob.glob(os.path.join(LIVE_DIR,'*.csv')))} fixtures w/ result: " f"{sum(1 for k,v in fixtures.items() if result(v))}") print("="*78) print("\nOVERALL") for name,label in [('immediate','immediate'),('p15','persist 15m'),('p30','persist 30m'),('p60','persist 60m')]: print(line(label, strat[name])) import statistics as _st print(f"\n dropped by filter (edge CLOSED -> the fakes we want to avoid): " f"15m={skipped_closed['p15']} 30m={skipped_closed['p30']} 60m={skipped_closed['p60']}") print(f" un-confirmable (no later snapshot before KO): 15m={skipped_closed['p15_noobs']} " f"30m={skipped_closed['p30_noobs']} 60m={skipped_closed['p60_noobs']}") print(f" median ACTUAL gap used (snapshots are ~hourly far from KO): " + " ".join(f"{n}={_st.median(actual_gaps[n]):.0f}min" if actual_gaps[n] else f"{n}=na" for n in ['p15','p30','p60'])) print("\nPER LEAGUE (immediate -> persist60)") rows_l = sorted(by_league.items(), key=lambda kv: -kv[1]['immediate']['n']) for league, accs in rows_l: if accs['immediate']['n'] < 3: continue print(f"\n {league}") for name,label in [('immediate','immediate'),('p15','persist 15m'),('p30','persist 30m'),('p60','persist 60m')]: print(line(label, accs[name]))