"""Download Kalshi price history for every settled soccer match since May 2025. For the ML experiment (2026-09-29): pairs with SoFIFA player ratings, Understat and Transfermarkt, so each match has the market's prices alongside the team/player data. Per market (each match has three: home, tie, away): * hourly candles, 72h before kickoff -> kickoff * 1-minute candles, 3h before kickoff -> 2.5h after (covers lineup release and the whole match) Each candle: yes bid/ask open/high/low/close, trade price mean/close, volume, open interest. Run with the LIVE Python (it needs kalshi_trader's authenticated client): python research\\kalshi_history.py --test --series KXEPLGAME # 4 matches python research\\kalshi_history.py # everything (resumable) HOW IT TREATS THE ACCOUNT. Read-only GETs, capped at MAX_RPS. The same API key serves the live xlow monitor, so this stays well under Kalshi's read limit to leave it headroom. Every response is cached under raw/, so a re-run resumes and never refetches. WHERE SETTLED DATA LIVES. Older settled markets move to Kalshi's /historical/ endpoints: /markets and /series/.../candlesticks return nothing or 404 for them. This tries /historical/ first and falls back to the live endpoints for recently settled markets that have not moved yet. KICKOFF = expected_expiration_time - 3h. Kalshi does not publish kickoff; historical markets carry no occurrence_datetime. The 3h offset matched 22/22 fixtures against API-Football kickoffs (see mm_maker_live MAKER_ONLY_SERIES) and matched Liverpool v Bournemouth 2025-08-15 exactly. """ import argparse import csv import datetime as dt import hashlib import json import sys import time from pathlib import Path sys.path.insert(0, r'E:\soccer\main') import kalshi_trader as kt # noqa: E402 from kalshi_live import _all_kalshi_series, _series_to_div # noqa: E402 import requests # noqa: E402 # ONE keep-alive connection for the whole run. kalshi_trader calls # requests.get() per request, i.e. a fresh socket (and DNS lookup) each time; # over tens of thousands of calls that is exactly the port churn behind the # recurring outages (memory: udp-port-exhaustion). This swaps it for a Session # in THIS process only -- the live monitor imports its own copy. kt.requests = requests.Session() OUT = Path(r'E:\soccer\research\ml_data\kalshi_history') RAW = OUT / 'raw' SINCE = dt.datetime(2025, 5, 1, tzinfo=dt.timezone.utc) EXTRA_SERIES = ['KXUEFANLGAME'] # maker-only series, not in maps KO_FROM_EXPIRY_H = 3.0 MAX_RPS = 5.0 HOURLY = (dt.timedelta(hours=72), dt.timedelta(0)) MINUTE = (dt.timedelta(hours=3), dt.timedelta(hours=2, minutes=30)) _last = [0.0] _n = {'api': 0, 'cached': 0} def get(path, params=None, cache=True): """GET with an on-disk cache. Returns parsed JSON, or None on 404. 429 / 5xx / network errors back off and retry (2s, 4s, ... 64s) instead of crashing a multi-hour run; anything still failing after that raises. """ key = RAW / (hashlib.sha1((path + json.dumps(params or {}, sort_keys=True)) .encode()).hexdigest() + '.json') if cache and key.exists(): _n['cached'] += 1 return json.loads(key.read_text(encoding='utf-8')) for attempt in range(7): wait = 1.0 / MAX_RPS - (time.time() - _last[0]) if wait > 0: time.sleep(wait) _last[0] = time.time() _n['api'] += 1 try: r = kt._get(path, params or {}) break except Exception as e: msg = str(e) if '-> 404' in msg: r = None break if attempt == 6 or not ('-> 429' in msg or '-> 5' in msg or 'Connection' in type(e).__name__ or 'Timeout' in type(e).__name__): raise time.sleep(2 ** (attempt + 1)) if cache: RAW.mkdir(parents=True, exist_ok=True) key.write_text(json.dumps(r), encoding='utf-8') return r def settled_events(series): out, cur = [], '' for _ in range(100): p = {'series_ticker': series, 'status': 'settled', 'limit': 200} if cur: p['cursor'] = cur r = get('/events', p, cache=False) or {} # new settlements appear out += r.get('events') or [] cur = r.get('cursor') or '' if not cur: break return out def markets_for(event): """(markets, source) -- source is 'historical' or 'live'.""" r = get('/historical/markets', {'event_ticker': event}) or {} ms = r.get('markets') or [] if ms: return ms, 'historical' r = get('/markets', {'event_ticker': event}) or {} return (r.get('markets') or []), 'live' def candles(series, ticker, source, start, end, interval): path = (f'/historical/markets/{ticker}/candlesticks' if source == 'historical' else f'/series/{series}/markets/{ticker}/candlesticks') r = get(path, {'start_ts': int(start.timestamp()), 'end_ts': int(end.timestamp()), 'period_interval': interval}) or {} return r.get('candlesticks') or [] def _f(x): try: return float(x) except (TypeError, ValueError): return '' def _v(d, k): """Field k, whichever naming the endpoint used. /historical/ candles say 'close' and 'volume'; the live endpoint says 'close_dollars' and 'volume_fp' for the same numbers.""" for name in (k, k + '_dollars', k + '_fp'): if d.get(name) is not None: return _f(d[name]) return '' def candle_row(ticker, c): b, a, p = c.get('yes_bid') or {}, c.get('yes_ask') or {}, c.get('price') or {} return {'ticker': ticker, 'ts_utc': dt.datetime.fromtimestamp(c['end_period_ts'], dt.timezone.utc).isoformat(), 'bid_open': _v(b, 'open'), 'bid_high': _v(b, 'high'), 'bid_low': _v(b, 'low'), 'bid_close': _v(b, 'close'), 'ask_open': _v(a, 'open'), 'ask_high': _v(a, 'high'), 'ask_low': _v(a, 'low'), 'ask_close': _v(a, 'close'), 'price_mean': _v(p, 'mean'), 'price_close': _v(p, 'close'), 'volume': _v(c, 'volume'), 'open_interest': _v(c, 'open_interest')} MKT_FIELDS = ['series', 'event', 'ticker', 'side', 'title', 'result', 'kickoff_utc', 'volume', 'open_time', 'close_time', 'settlement_ts', 'source'] CANDLE_FIELDS = ['ticker', 'ts_utc', 'bid_open', 'bid_high', 'bid_low', 'bid_close', 'ask_open', 'ask_high', 'ask_low', 'ask_close', 'price_mean', 'price_close', 'volume', 'open_interest'] def run(test=False, only=None): OUT.mkdir(parents=True, exist_ok=True) # Leagues we hold odds/stats files for go first: they are the ML join. # The rest (MLS, Argentina, ...) are market-only and fill in afterwards. have = {f.name.split('_')[0] for f in Path(r'E:\soccer\league_data').glob('*_*.csv')} s2d = _series_to_div() series_list = sorted(set(_all_kalshi_series()) | set(EXTRA_SERIES), key=lambda x: (s2d.get(x) not in have, x)) if only: series_list = [x for x in series_list if x in only] files = {k: open(OUT / f, 'w', newline='', encoding='utf-8') for k, f in (('m', 'markets.csv'), ('h', 'candles_1h.csv'), ('1', 'candles_1m.csv'))} wm = csv.DictWriter(files['m'], fieldnames=MKT_FIELDS); wm.writeheader() wh = csv.DictWriter(files['h'], fieldnames=CANDLE_FIELDS); wh.writeheader() w1 = csv.DictWriter(files['1'], fieldnames=CANDLE_FIELDS); w1.writeheader() done = no_time = 0 try: for s in series_list: evs = settled_events(s) if test: # newest two (live endpoints) + oldest two (historical) evs = evs[:2] + evs[-2:] kept = 0 for e in evs: ms, src = markets_for(e['event_ticker']) for m in ms: exp = m.get('expected_expiration_time') if not exp: no_time += 1 continue ko = (dt.datetime.fromisoformat(exp.replace('Z', '+00:00')) - dt.timedelta(hours=KO_FROM_EXPIRY_H)) if ko < SINCE: continue wm.writerow({'series': s, 'event': e['event_ticker'], 'ticker': m['ticker'], 'side': m['ticker'].rsplit('-', 1)[-1], 'title': e.get('title', ''), 'result': m.get('result', ''), 'kickoff_utc': ko.isoformat(), 'volume': m.get('volume_fp', m.get('volume', '')), 'open_time': m.get('open_time', ''), 'close_time': m.get('close_time', ''), 'settlement_ts': m.get('settlement_ts', ''), 'source': src}) for c in candles(s, m['ticker'], src, ko - HOURLY[0], ko + HOURLY[1], 60): wh.writerow(candle_row(m['ticker'], c)) for c in candles(s, m['ticker'], src, ko - MINUTE[0], ko + MINUTE[1], 1): w1.writerow(candle_row(m['ticker'], c)) kept += 1 done += 1 if done % 100 == 0: print(f' {done} matches ({_n["api"]} API calls, {_n["cached"]} cached)') for fh in files.values(): fh.flush() print(f'{s:28} {len(evs):5} settled events, {kept} markets since {SINCE:%Y-%m-%d}') finally: for fh in files.values(): fh.close() print(f'done: {done} matches, {_n["api"]} API calls, {_n["cached"]} from cache, ' f'{no_time} markets without an expiry time skipped -> {OUT}') if __name__ == '__main__': ap = argparse.ArgumentParser() ap.add_argument('--test', action='store_true', help='newest 2 + oldest 2 matches per series') ap.add_argument('--series', nargs='*', help='limit to these series tickers') a = ap.parse_args() run(a.test, set(a.series) if a.series else None)