"""Download lineups (and the rest of each match's detail) from API-Football. Every finished match since May 2025 in the 44 leagues of data/league_map.json (the 22 league_data divisions plus the Kalshi-only leagues). Run with the isolated env: E:\\soccer\\ml_env\\Scripts\\python.exe research\\af_lineups.py --test E:\\soccer\\ml_env\\Scripts\\python.exe research\\af_lineups.py HOW. /fixtures?league=&season= lists a league's matches (1 request per league-season); /fixtures?ids=a-b-... returns up to 20 matches per request WITH lineups, events, team statistics and per-player match stats embedded -- 20x cheaper than /fixtures/lineups one match at a time. QUOTA. The key is the live system's (Pro plan, 7,500/day), shared with the xlow monitor and the daily jobs. This stops by itself when the day's remaining count falls below QUOTA_FLOOR, whatever is left to fetch, and resumes from its cache on the next run. Finished matches are cached forever; league listings are re-read each run so new results appear. Writes research/ml_data/api_football/: fixtures.csv one row per match: ids, date, teams, score, referee, venue lineups.csv one row per player named in a lineup: starter or sub, shirt number, position, grid slot; team formation + coach players.csv per-player match stats: minutes, rating, shots, passes, tackles, duels, dribbles, fouls, cards, saves, ... events.csv goals, cards, substitutions, VAR, with minute team_stats.csv per-team match stats (shots, possession, xG where given) """ import argparse import csv import hashlib import json import re import sys import time from pathlib import Path import requests BASE = 'https://v3.football.api-sports.io' OUT = Path(r'E:\soccer\research\ml_data\api_football') RAW = OUT / 'raw' SINCE = '2025-05-01' SEASONS = (2024, 2025, 2026) # AF season = the year it starts QUOTA_FLOOR = 3000 # leave this many requests for the live system FINISHED = {'FT', 'AET', 'PEN'} MIN_GAP_S = 0.4 # Pro allows 300/min; stay well under # The key lives in the live code; read it rather than copy it here. _KEY = re.search(r"AF_KEY\s*=\s*'([0-9a-f]+)'", Path(r'E:\soccer\main\daily_signals.py').read_text(encoding='utf-8')).group(1) _S = requests.Session() _S.headers['x-apisports-key'] = _KEY _last = [0.0] _n = {'api': 0, 'cached': 0, 'remaining': None} class QuotaFloor(Exception): pass def get(path, params, cache=True): key = RAW / (hashlib.sha1((path + json.dumps(params, sort_keys=True)).encode()).hexdigest() + '.json') if cache and key.exists(): _n['cached'] += 1 return json.loads(key.read_text(encoding='utf-8')) if _n['remaining'] is not None and _n['remaining'] < QUOTA_FLOOR: raise QuotaFloor(f'{_n["remaining"]} requests left today (floor {QUOTA_FLOOR})') for attempt in range(5): wait = MIN_GAP_S - (time.time() - _last[0]) if wait > 0: time.sleep(wait) _last[0] = time.time() r = _S.get(BASE + path, params=params, timeout=30) _n['api'] += 1 rem = r.headers.get('x-ratelimit-requests-remaining') if rem is not None: _n['remaining'] = int(rem) if r.status_code == 429 or r.status_code >= 500: time.sleep(2 ** (attempt + 2)) continue r.raise_for_status() d = r.json() if d.get('errors'): # e.g. {'rateLimit': ...} or {'requests': 'reached the limit'} if 'rateLimit' in json.dumps(d['errors']): time.sleep(2 ** (attempt + 2)) continue raise RuntimeError(f'API-Football {path} {params}: {d["errors"]}') if cache: RAW.mkdir(parents=True, exist_ok=True) key.write_text(json.dumps(d), encoding='utf-8') return d raise RuntimeError(f'API-Football {path} {params}: still failing after retries') def league_fixtures(league_id): out = [] for season in SEASONS: d = get('/fixtures', {'league': league_id, 'season': season}, cache=False) for f in d.get('response') or []: fx = f['fixture'] if fx['status']['short'] in FINISHED and fx['date'][:10] >= SINCE: # skip the last day: stats and ratings are sometimes revised # for a few hours after full time, and details are cached forever if time.time() - fx['timestamp'] > 24 * 3600: out.append(f) return out def flat(prefix, d, out): """Flatten nested dicts: {'shots': {'total': 3}} -> shots_total.""" for k, v in (d or {}).items(): name = f'{prefix}{k}' if isinstance(v, dict): flat(name + '_', v, out) else: out[name] = v return out def run(test=False): lm = json.loads(Path(r'E:\soccer\data\league_map.json').read_text(encoding='utf-8')) leagues = [(div, v['api_football_id']) for div, v in lm.items() if v.get('api_football_id')] if test: leagues = [x for x in leagues if x[0] in ('E0', 'E2')] OUT.mkdir(parents=True, exist_ok=True) files, writers = {}, {} def write(name, row): if name not in writers: files[name] = open(OUT / f'{name}.csv', 'w', newline='', encoding='utf-8') writers[name] = csv.DictWriter(files[name], fieldnames=list(row), extrasaction='ignore') writers[name].writeheader() writers[name].writerow(row) # players.csv has a fixed column set: the union of every stat key seen # would change mid-file, so collect player rows and write them at the end. player_rows = [] done = missing_lineups = 0 try: for div, lid in leagues: fx = league_fixtures(lid) if test: fx = fx[:20] got = 0 for i in range(0, len(fx), 20): ids = '-'.join(str(f['fixture']['id']) for f in fx[i:i + 20]) for f in get('/fixtures', {'ids': ids}).get('response') or []: fid, x, lg = f['fixture']['id'], f['fixture'], f['league'] write('fixtures', { 'fixture_id': fid, 'div': div, 'league_id': lg['id'], 'season': lg['season'], 'round': lg.get('round'), 'date_utc': x['date'], 'status': x['status']['short'], 'home_id': f['teams']['home']['id'], 'home': f['teams']['home']['name'], 'away_id': f['teams']['away']['id'], 'away': f['teams']['away']['name'], 'home_goals': f['goals']['home'], 'away_goals': f['goals']['away'], 'ht_home': f['score']['halftime']['home'], 'ht_away': f['score']['halftime']['away'], 'referee': x.get('referee'), 'venue': (x.get('venue') or {}).get('name'), 'n_lineups': len(f.get('lineups') or [])}) if len(f.get('lineups') or []) < 2: missing_lineups += 1 for tl in f.get('lineups') or []: for kind, lst in (('start', tl.get('startXI')), ('sub', tl.get('substitutes'))): for p in lst or []: p = p['player'] write('lineups', { 'fixture_id': fid, 'team_id': tl['team']['id'], 'team': tl['team']['name'], 'formation': tl.get('formation'), 'coach_id': (tl.get('coach') or {}).get('id'), 'coach': (tl.get('coach') or {}).get('name'), 'role': kind, 'player_id': p['id'], 'player': p['name'], 'number': p.get('number'), 'pos': p.get('pos'), 'grid': p.get('grid')}) for tp in f.get('players') or []: for p in tp.get('players') or []: row = {'fixture_id': fid, 'team_id': tp['team']['id'], 'player_id': p['player']['id'], 'player': p['player']['name']} for st in p.get('statistics') or []: flat('', st, row) player_rows.append(row) for e in f.get('events') or []: write('events', { 'fixture_id': fid, 'minute': e['time']['elapsed'], 'extra': e['time']['extra'], 'team_id': e['team']['id'], 'type': e['type'], 'detail': e['detail'], 'player_id': e['player']['id'], 'player': e['player']['name'], 'assist_id': e['assist']['id'], 'assist': e['assist']['name'], 'comments': e.get('comments')}) for ts in f.get('statistics') or []: for s in ts.get('statistics') or []: write('team_stats', {'fixture_id': fid, 'team_id': ts['team']['id'], 'stat': s['type'], 'value': s['value']}) got += 1 done += 1 print(f'{div:5} league {lid:<4} {got:5} matches ' f'({_n["api"]} API calls, {_n["cached"]} cached, {_n["remaining"]} left today)') except QuotaFloor as e: print(f'STOPPED at the quota floor: {e}. Re-run tomorrow; everything so far is cached.') finally: if player_rows: cols = list(dict.fromkeys(k for r in player_rows for k in r)) with open(OUT / 'players.csv', 'w', newline='', encoding='utf-8') as fh: w = csv.DictWriter(fh, fieldnames=cols) w.writeheader() w.writerows(player_rows) for fh in files.values(): fh.close() print(f'done: {done} matches ({missing_lineups} without both lineups), ' f'{_n["api"]} API calls, {_n["cached"]} cached -> {OUT}') if __name__ == '__main__': ap = argparse.ArgumentParser() ap.add_argument('--test', action='store_true', help='EPL + League One, 20 matches each') run(ap.parse_args().test)