"""Can we predict, BEFORE kickoff, which games will swing in play? In-play moves are 16x pre-match at the p99 (mm_inplay_swings.py). The question is whether the swinging is foreseeable from things known before the ball is kicked, since everything a strategy can act on has to be known by then. Target: in-play price movement per fixture -- mean absolute snapshot-to- snapshot move across its three legs, from live_data with elapsed > 0. Features: all strictly pre-kickoff -- expected goals, from API-Football rolling gf/ga over the last 5 the favourite's price (a 75c favourite has less room to travel) how evenly matched the book is (spread between top and bottom leg) league recent shot volume, as a proxy for chaotic end-to-end games A goal is what moves a soccer book, so if anything predicts the swinging it ought to be how many goals the two teams have been involved in. """ import csv, collections, os, statistics, sys sys.path.insert(0, r'E:\soccer\main') sys.path.insert(0, r'E:\soccer\research') import mm_settle as S LIVE_DIR = r'E:\soccer\live_data' FEATS = r'E:\soccer\raw_api\match_features.csv' SIDES = {'home': 'kalshi_home_mid', 'draw': 'kalshi_draw_mid', 'away': 'kalshi_away_mid'} DAYS = 30 def _f(x): try: return float(x) except (TypeError, ValueError): return None def load_live(): """{(gd, home, away): {'pre': {side: [mid]}, 'live': {side: [(ts, mid)]}}}""" files = sorted(f for f in os.listdir(LIVE_DIR) if f.endswith('.csv'))[-DAYS:] out = collections.defaultdict( lambda: {'pre': collections.defaultdict(list), 'live': collections.defaultdict(list)}) for fn in files: with open(os.path.join(LIVE_DIR, fn), encoding='utf-8', errors='replace') as fh: for r in csv.DictReader(fh): ts = S._pt(r.get('timestamp')) el = r.get('elapsed') if ts is None or el in (None, ''): continue try: el = float(el) except ValueError: continue key = (r.get('game_date'), r.get('home'), r.get('away'), r.get('league')) for side, col in SIDES.items(): m = _f(r.get(col)) if m is None or not (0 < m < 1): continue if el > 0: out[key]['live'][side].append((ts, m)) else: out[key]['pre'][side].append(m) return out def load_feats(): """{(date, team): {gf5, ga5, shots5}} from API-Football rolling features.""" out = {} if not os.path.exists(FEATS): return out with open(FEATS, encoding='utf-8', errors='replace') as fh: for r in csv.DictReader(fh): d, t = (r.get('date') or '')[:10], r.get('team') if not d or not t: continue out[(d, t)] = {'gf': _f(r.get('api_gf_w5')), 'ga': _f(r.get('api_ga_w5')), 'shots': _f(r.get('api_total_shots_w5'))} return out def inplay_move(live): vals = [] for side, pts in live.items(): s = sorted(pts) vals += [abs(b - a) * 100 for (_, a), (_, b) in zip(s, s[1:])] return statistics.mean(vals) if len(vals) >= 10 else None def bucket(rows, key, label, n=4): vals = sorted(r[key] for r in rows if r.get(key) is not None) if len(vals) < 60: return cuts = [vals[int(len(vals) * p)] for p in (0.25, 0.5, 0.75)] names = ['lowest 25%', '2nd', '3rd', 'highest 25%'] edges = [(-1e9, cuts[0]), (cuts[0], cuts[1]), (cuts[1], cuts[2]), (cuts[2], 1e9)] print(f'\n by {label}:') got = [] for (lo, hi), nm in zip(edges, names): sel = [r for r in rows if r.get(key) is not None and lo <= r[key] < hi] if len(sel) < 12: continue m = statistics.median(r['move'] for r in sel) got.append((nm, m)) print(f' {nm:>12} {len(sel):>4} games median in-play move ' f'{m:>5.2f}c') if len(got) >= 2: print(f' -> {got[-1][0]} vs {got[0][0]}: ' f'{got[-1][1]-got[0][1]:+.2f}c ' f'({got[-1][1]/max(0.01,got[0][1]):.2f}x)') def main(): live = load_live() feats = load_feats() print(f'{len(live)} fixtures in live_data, {len(feats):,} team-match ' f'feature rows') rows = [] for (gd, home, away, lg), d in live.items(): mv = inplay_move(d['live']) if mv is None: continue pre = {s: statistics.median(v) for s, v in d['pre'].items() if v} if len(pre) != 3: continue fh_, fa_ = feats.get((gd, home)), feats.get((gd, away)) exp_goals = None shots = None if fh_ and fa_ and all(x is not None for x in (fh_['gf'], fh_['ga'], fa_['gf'], fa_['ga'])): exp_goals = (fh_['gf'] + fa_['ga'] + fa_['gf'] + fh_['ga']) / 2.0 if fh_['shots'] is not None and fa_['shots'] is not None: shots = fh_['shots'] + fa_['shots'] rows.append({'move': mv, 'league': lg, 'fav': max(pre.values()), 'spread': max(pre.values()) - min(pre.values()), 'draw': pre.get('draw'), 'exp_goals': exp_goals, 'shots': shots}) print(f'{len(rows)} fixtures with in-play movement measured, ' f'{sum(1 for r in rows if r["exp_goals"] is not None)} with ' f'API-Football features\n') if len(rows) < 80: print('too few') return print(f'median in-play move overall: ' f'{statistics.median(r["move"] for r in rows):.2f}c') bucket(rows, 'fav', "the favourite's pre-match price") bucket(rows, 'spread', 'how lopsided the book is (top leg - bottom leg)') bucket(rows, 'draw', 'the draw price') fg = [r for r in rows if r['exp_goals'] is not None] if len(fg) >= 60: bucket(fg, 'exp_goals', 'expected goals (rolling gf/ga, last 5)') bucket([r for r in fg if r['shots'] is not None], 'shots', 'recent shot volume (both teams, last 5)') by_lg = collections.defaultdict(list) for r in rows: by_lg[r['league']].append(r['move']) lgs = sorted(((lg, statistics.median(v), len(v)) for lg, v in by_lg.items() if len(v) >= 8), key=lambda x: -x[1]) if lgs: print('\n by league (top and bottom):') for lg, m, k in lgs[:4]: print(f' {lg:>8} {k:>4} games {m:>5.2f}c') print(' ...') for lg, m, k in lgs[-3:]: print(f' {lg:>8} {k:>4} games {m:>5.2f}c') print(f' -> {lgs[0][1]/max(0.01,lgs[-1][1]):.1f}x across leagues') if __name__ == '__main__': main()