"""Join every data source to the league_data match table (ML experiment). Run with the isolated env (re-run any time; it reads whatever the downloads have produced so far): E:\\soccer\\ml_env\\Scripts\\python.exe research\\ml_join.py Spine: league_data -- one row per match, 22 European divisions, 2019/20 on, with results, match stats and bookmaker odds (opening and closing). Attached per match: elo_* research/ml_elo.py (pre-match ratings) us_* Understat xG / npxG / PPDA / deep completions (top 5 only) tm_* Transfermarkt game and club ids (first tiers; lineups/values are joined later through these ids) kalshi_* Kalshi event, the three market tickers and kickoff time TEAM NAMES. Every source spells clubs differently ("Man United", "Manchester United", "MUN", club id 985). Nothing is hand-mapped: each source's names are mapped to league_data's by CO-OCCURRENCE -- the same division on the same date (+-1 day), counted over the whole history, so the true pairing wins by a wide margin. Unresolved or contested names are listed, never guessed. The existing data/team_map.json (used by the live system) is only a cross-check. Writes research/ml_data/: matches_joined.csv the spine plus the ids above name_map.csv every source name -> league_data name, with its votes """ import json from collections import Counter, defaultdict from pathlib import Path import pandas as pd ML = Path(r'E:\soccer\research\ml_data') LD = Path(r'E:\soccer\league_data') TOL = 1 # days either side for date matching KEEP = ['Div', 'Date', 'Time', 'HomeTeam', 'AwayTeam', 'FTHG', 'FTAG', 'FTR', 'HTHG', 'HTAG', 'HS', 'AS', 'HST', 'AST', 'HF', 'AF', 'HC', 'AC', 'HY', 'AY', 'HR', 'AR', 'PSH', 'PSD', 'PSA', 'PSCH', 'PSCD', 'PSCA', 'AvgH', 'AvgD', 'AvgA', 'AvgCH', 'AvgCD', 'AvgCA', 'MaxCH', 'MaxCD', 'MaxCA', 'B365H', 'B365D', 'B365A', 'B365CH', 'B365CD', 'B365CA', 'BFEH', 'BFED', 'BFEA', 'BFECH', 'BFECD', 'BFECA', 'Avg>2.5', 'Avg<2.5', 'AvgC>2.5', 'AvgC<2.5', 'AHh', 'AHCh', 'AvgAHH', 'AvgAHA', 'AvgCAHH', 'AvgCAHA'] US_DIV = {'ENG-Premier League': 'E0', 'ESP-La Liga': 'SP1', 'GER-Bundesliga': 'D1', 'ITA-Serie A': 'I1', 'FRA-Ligue 1': 'F1'} TM_DIV = {'GB1': 'E0', 'ES1': 'SP1', 'L1': 'D1', 'IT1': 'I1', 'FR1': 'F1', 'NL1': 'N1', 'PO1': 'P1', 'BE1': 'B1', 'SC1': 'SC0', 'TR1': 'T1', 'GR1': 'G1'} def load_spine(): frames = [] for f in sorted(LD.glob('*_[0-9][0-9][0-9][0-9].csv')): for enc in ('utf-8-sig', 'cp1252'): try: d = pd.read_csv(f, encoding=enc, on_bad_lines='skip', low_memory=False, usecols=lambda c: c in KEEP) break except UnicodeDecodeError: continue d['season'] = f.stem.split('_')[1] frames.append(d) d = pd.concat(frames, ignore_index=True).dropna(subset=['HomeTeam', 'FTHG']) d['Date'] = pd.to_datetime(d['Date'], dayfirst=True, errors='coerce') d = d.dropna(subset=['Date']).drop_duplicates(['Div', 'Date', 'HomeTeam', 'AwayTeam']) return d.sort_values(['Date', 'Div']).reset_index(drop=True) def _fold(x): import unicodedata x = unicodedata.normalize('NFKD', str(x)).encode('ascii', 'ignore').decode().lower() return ' '.join(''.join(ch if ch.isalnum() else ' ' for ch in x).split()) def name_seed(labels, spine, extra_alias=None, cut=0.88): """(div, src_name) -> spine name from a readable label ("Huddersfield", "Sudtirol Bolzano"), where one spine club in that division is clearly the closest spelling. Only used to START the co-occurrence propagation; a seed that the fixtures contradict loses to them in cooccur().""" from difflib import SequenceMatcher teams = defaultdict(set) for dv, h, a in zip(spine.Div, spine.HomeTeam, spine.AwayTeam): teams[dv].update((h, a)) out = {} for (dv, src_name), label in labels.items(): if not isinstance(label, str): continue L = _fold(label) sc = [] for t in teams.get(dv, ()): names = {t} | set((extra_alias or {}).get(t, ())) best = max(1.0 if (_fold(n) in L or L in _fold(n)) and min(len(_fold(n)), len(L)) >= 4 else SequenceMatcher(None, _fold(n), L).ratio() for n in names) sc.append((best, t)) sc.sort(reverse=True) if sc and sc[0][0] >= cut and (len(sc) == 1 or sc[0][0] - sc[1][0] >= 0.08): out[(dv, src_name)] = sc[0][1] return out def cooccur(src, spine, source, seed=None): """Map src team names to spine names, per division, from the fixtures. src: DataFrame[div, date, home, away]. Returns (mapping, report rows). Pass 1 -- same-day votes. Every spine fixture in the division on the same date (+-1 day) votes for its home club as the src home club's name, and likewise for away. On a weekend where the whole division plays at once that is nearly every club, so this only separates teams in leagues that spread their fixtures; it is kept to START the mapping, not to finish it. Pass 2 -- anchored propagation. Once one side of a src fixture is known (from pass 1 or a name seed), the spine fixture that day with that club in that slot is unique, which pins down the OTHER side exactly. Repeated until nothing new resolves. A name is accepted with >=2 anchored votes and at least twice its runner-up; each spine name goes to one src name only. """ by_day = defaultdict(list) for dv, dte, h, a in zip(spine.Div, spine.Date.dt.normalize(), spine.HomeTeam, spine.AwayTeam): by_day[(dv, dte)].append((h, a)) src_rows = list(zip(src['div'], src.date.dt.normalize(), src.home, src.away)) votes = defaultdict(Counter) for dv, dte, h, a in src_rows: for k in range(-TOL, TOL + 1): cands = by_day.get((dv, dte + pd.Timedelta(days=k)), ()) w = (1.0 if k == 0 else 0.3) / max(len(cands), 1) for H, A in cands: votes[(dv, h)][H] += w votes[(dv, a)][A] += w mapping = dict(seed or {}) for key, c in votes.items(): top = c.most_common(2) (tgt, v), second = top[0], (top[1][1] if len(top) > 1 else 0.0) m = v / max(second, 1e-9) if key not in mapping and (tgt == key[1] or (v >= 3 and m >= 2) or (v >= 5 and m >= 1.5) or (v >= 1.5 and m >= 2.5)): mapping[key] = tgt def dedupe(mp, strength): claims = defaultdict(list) for key, tgt in mp.items(): claims[(key[0], tgt)].append((strength.get(key, 0), key)) out, lost = {}, {} for (dv, tgt), lst in claims.items(): lst.sort(reverse=True) out[lst[0][1]] = tgt for v, loser in lst[1:]: # A venue can re-code a club mid-history (Kalshi: CHE and CFC # for Chelsea, OM and OLM for Marseille). Keep a second code # only when the fixtures alone anchor it firmly. c = anchored.get(loser, Counter()) runner = max([n for t, n in c.items() if t != tgt], default=0) if c.get(tgt, 0) >= 3 and c.get(tgt, 0) >= 4 * max(runner, 0.5): out[loser] = tgt else: lost[loser] = f'lost {tgt} to {lst[0][1][1]}' return out, lost anchored, lost = defaultdict(Counter), {} for _ in range(6): anchored = defaultdict(Counter) for dv, dte, h, a in src_rows: H, A = mapping.get((dv, h)), mapping.get((dv, a)) for k in range(-TOL, TOL + 1): day = by_day.get((dv, dte + pd.Timedelta(days=k)), ()) if H: opp = [x[1] for x in day if x[0] == H] if len(opp) == 1: anchored[(dv, a)][opp[0]] += 1 break if A: opp = [x[0] for x in day if x[1] == A] if len(opp) == 1: anchored[(dv, h)][opp[0]] += 1 break new = {} for key, c in anchored.items(): top = c.most_common(2) (tgt, v), second = top[0], (top[1][1] if len(top) > 1 else 0) if v >= 2 and v >= 2 * second: new[key] = tgt strength = {k: c.most_common(1)[0][1] for k, c in anchored.items()} merged, lost = dedupe({**mapping, **new}, strength) if merged == mapping: break mapping = merged rows = [] for key in set(votes) | set(anchored): c = anchored.get(key) or Counter() top = c.most_common(2) tgt = mapping.get(key) or (top[0][0] if top else votes[key].most_common(1)[0][0]) v = c.get(tgt, 0) second = max([n for t, n in c.items() if t != tgt], default=0) rows.append({'source': source, 'div': key[0], 'src_name': key[1], 'fd_name': tgt, 'votes': v, 'margin': round(v / max(second, 0.5), 1), 'ok': key in mapping, 'note': lost.get(key, '')}) return mapping, rows def attach(spine, src, mapping, cols, prefix): """Left-join src rows (already name-mapped) onto the spine by (div, home, away, date within TOL). Returns the spine with prefixed cols.""" idx = {} for i, (dv, dte, h, a) in enumerate(zip(spine.Div, spine.Date.dt.normalize(), spine.HomeTeam, spine.AwayTeam)): idx[(dv, h, a, dte)] = i hit = {} for j, (dv, dte, h, a) in enumerate(zip(src['div'], src.date.dt.normalize(), src.home, src.away)): H, A = mapping.get((dv, h)), mapping.get((dv, a)) if not (H and A): continue for k in (0, -1, 1)[:2 * TOL + 1]: i = idx.get((dv, H, A, dte + pd.Timedelta(days=k))) if i is not None and i not in hit: hit[i] = j break out = spine.copy() add = pd.DataFrame({f'{prefix}{c}': src[c].to_numpy()[list(hit.values())] for c in cols}, index=list(hit.keys())) return out.join(add), len(hit) def kalshi_events(): m = pd.read_csv(ML / 'kalshi_history' / 'markets.csv') lm = json.loads((Path(r'E:\soccer\data\league_map.json')).read_text(encoding='utf-8')) s2d = {v.get('kalshi_series'): k for k, v in lm.items() if v.get('kalshi_series')} rows = [] for ev, g in m.groupby('event'): sides = [s for s in g.side if s != 'TIE'] if len(sides) != 2 or g.series.iloc[0] not in s2d: continue suffix = ev.split('-', 1)[1][7:] # after YYMONDD a, b = sides home, away = (a, b) if suffix == a + b else (b, a) if suffix == b + a else (None, None) if not home: continue tk = dict(zip(g.side, g.ticker)) res = g.loc[g.result == 'yes', 'side'] rows.append({'div': s2d[g.series.iloc[0]], 'series': g.series.iloc[0], 'event': ev, 'date': pd.to_datetime(g.kickoff_utc.iloc[0]).tz_convert(None), 'kickoff_utc': g.kickoff_utc.iloc[0], 'title': g.title.iloc[0], 'home': home, 'away': away, 'tk_home': tk[home], 'tk_tie': tk.get('TIE'), 'tk_away': tk[away], 'result': {home: 'H', away: 'A', 'TIE': 'D'}.get(res.iloc[0]) if len(res) else None}) return pd.DataFrame(rows) def main(): spine = load_spine() print(f'spine: {len(spine)} matches, {spine.Div.nunique()} divisions, ' f'{spine.Date.min():%Y-%m-%d} -> {spine.Date.max():%Y-%m-%d}') elo = pd.read_csv(ML / 'elo_matches.csv', parse_dates=['Date']) spine = spine.merge(elo[['Div', 'Date', 'HomeTeam', 'AwayTeam', 'elo_home', 'elo_away', 'elo_diff', 'elo_exp_home']], on=['Div', 'Date', 'HomeTeam', 'AwayTeam'], how='left') report = [] # Understat us = pd.read_csv(ML / 'understat_team_matches.csv') us['div'] = us.league.map(US_DIV) us['date'] = pd.to_datetime(us.date) us = us.rename(columns={'home_team': 'home', 'away_team': 'away'}) us = us[us.date >= '2019-07-01'] mp, rep = cooccur(us, spine, 'understat') report += rep cols = ['game_id', 'home_xg', 'away_xg', 'home_np_xg', 'away_np_xg', 'home_ppda', 'away_ppda', 'home_deep_completions', 'away_deep_completions', 'home_expected_points', 'away_expected_points'] spine, n = attach(spine, us, mp, cols, 'us_') print(f'understat: {n}/{len(us)} games joined ({n / len(us):.1%})') # Transfermarkt tm = pd.read_csv(ML / 'transfermarkt' / 'games.csv.gz', low_memory=False) tm = tm[tm.competition_id.isin(TM_DIV)].copy() tm['div'] = tm.competition_id.map(TM_DIV) tm['date'] = pd.to_datetime(tm.date) tm = tm[tm.date >= '2019-07-01'] tm['home'], tm['away'] = tm.home_club_id.astype(str), tm.away_club_id.astype(str) mp, rep = cooccur(tm, spine, 'transfermarkt') names = dict(zip(tm.home, tm.home_club_name)) | dict(zip(tm.away, tm.away_club_name)) for r in rep: r['src_label'] = names.get(r['src_name']) report += rep spine, n = attach(spine, tm, mp, ['game_id', 'home_club_id', 'away_club_id'], 'tm_') print(f'transfermarkt: {n}/{len(tm)} games joined ({n / len(tm):.1%})') # API-Football (lineups, player match stats, events: research/af_lineups.py) af = pd.read_csv(ML / 'api_football' / 'fixtures.csv') af = af[af['div'].isin(set(spine.Div))].copy() af['date'] = pd.to_datetime(af.date_utc, utc=True).dt.tz_convert(None) tmap_af = json.loads(Path(r'E:\soccer\data\team_map.json').read_text(encoding='utf-8')) # team_map.json is keyed by API-Football's own club names (the live system # resolves AF fixtures with it), so it seeds the mapping directly; the # fixtures then confirm or override each seed. teams_by_div = defaultdict(set) for dv, h, a in zip(spine.Div, spine.HomeTeam, spine.AwayTeam): teams_by_div[dv].update((h, a)) seed = {} for dv, h, a in zip(af['div'], af.home, af.away): for t in (h, a): v = tmap_af.get(t) if isinstance(v, dict) and v.get('fdco') in teams_by_div[dv]: seed[(dv, t)] = v['fdco'] mp, rep = cooccur(af, spine, 'api_football', seed=seed) report += rep spine, n = attach(spine, af, mp, ['fixture_id', 'home_id', 'away_id'], 'af_') print(f'api_football: {n}/{len(af)} matches joined ({n / max(len(af), 1):.1%})') # Kalshi kev = kalshi_events() t = kev.title.str.replace(r':.*$', '', regex=True).str.split(' vs ') lab = pd.DataFrame({'div': list(kev['div']) * 2, 'abr': list(kev.home) + list(kev.away), 'name': list(t.str[0]) + list(t.str[-1])}) lab = lab.groupby(['div', 'abr']).name.agg(lambda x: x.mode().iloc[0]).to_dict() tmap = json.loads(Path(r'E:\soccer\data\team_map.json').read_text(encoding='utf-8')) alias = defaultdict(set) for k, v in tmap.items(): if isinstance(v, dict) and v.get('fdco'): alias[v['fdco']].update([k] + list(v.get('aliases') or [])) mp, rep = cooccur(kev, spine, 'kalshi', seed=name_seed(lab, spine, alias)) for r in rep: r['src_label'] = lab.get((r['div'], r['src_name'])) report += rep spine, n = attach(spine, kev, mp, ['event', 'series', 'kickoff_utc', 'tk_home', 'tk_tie', 'tk_away', 'result'], 'kalshi_') bad = ((spine.kalshi_result.notna()) & (spine.kalshi_result != spine.FTR)).sum() print(f'kalshi: {n}/{len(kev)} events joined ({n / max(len(kev), 1):.1%}); ' f'result disagrees with league_data on {bad}') # cross-check Kalshi abbreviations against the live system's team_map tmap = json.loads(Path(r'E:\soccer\data\team_map.json').read_text(encoding='utf-8')) live = {(v.get('div'), v.get('kalshi')): v.get('fdco') for v in tmap.values() if isinstance(v, dict) and v.get('kalshi') and v.get('fdco')} agree = dis = 0 for (dv, ab), fd in mp.items(): if (dv, ab) in live: if live[(dv, ab)] == fd: agree += 1 else: dis += 1 print(f' team_map disagrees: {dv} {ab} -> cooccur {fd!r}, team_map {live[(dv, ab)]!r}') print(f'kalshi names vs team_map.json: {agree} agree, {dis} disagree') ks = kev.assign(joined=kev.event.isin(set(spine.kalshi_event.dropna()))) ks = ks[ks['div'].isin(set(spine.Div))].groupby('series').joined.agg(['sum', 'size']) print('kalshi events joined per series (leagues with odds files):') print(' ' + ', '.join(f'{s.replace("GAME", "")[2:]} {int(r["sum"])}/{int(r["size"])}' for s, r in ks.iterrows()) + f' -> total {int(ks["sum"].sum())}/{int(ks["size"].sum())}' + f' ({ks["sum"].sum() / ks["size"].sum():.1%})') rep = pd.DataFrame(report) rep.to_csv(ML / 'name_map.csv', index=False) unres = rep[~rep.ok] if len(unres): print(f'{len(unres)} unresolved names:') print(unres.sort_values(['source', 'div']).to_string(index=False, max_rows=60)) spine.to_csv(ML / 'matches_joined.csv', index=False) # Coverage of the evaluation window (Kalshi era) per division k = spine[spine.Date >= '2025-05-01'] cov = k.groupby('Div').agg(matches=('HomeTeam', 'size'), kalshi=('kalshi_event', lambda x: x.notna().sum()), understat=('us_game_id', lambda x: x.notna().sum()), lineups=('af_fixture_id', lambda x: x.notna().sum()), tm=('tm_game_id', lambda x: x.notna().sum()), pinnacle_close=('PSCH', lambda x: x.notna().sum()), avg_close=('AvgCH', lambda x: x.notna().sum()), shots=('HS', lambda x: x.notna().sum())) print('\nsince 2025-05-01 (matches with each source):') print(cov.to_string()) print(f'wrote {ML / "matches_joined.csv"} ({len(spine)} rows)') if __name__ == '__main__': main()