"""Capture REAL-TIME Pinnacle odds alongside our existing API-Football feed. WHY. Measured 2026-07-28 on 147 fixtures (median 91 pre-match snapshots each): API-Football's "Pinnacle" money line changes about once every 5.8 HOURS, while Kalshi's own mid changes every ~2.3h. The sharpest book in the world repricing half as often as a retail prediction market is not a fact about markets -- it is a relay artifact. Every number this project rests on (the [5,12) CLV band, the n=286 backtest, the 144-bet forward test) was computed against that stale anchor, so an unknown share of what we call "CLV" may be pure time-lag rather than mispricing. This logs a fresh Pinnacle line next to the stale one so the question can be settled with data instead of argument. It TRADES NOTHING and changes no existing behaviour -- the live pipeline keeps using API-Football until the comparison says otherwise. COST CONTROL. pinnapi's free tier is 100 requests/day and returns EVERY soccer prematch fixture in one call, so one poll covers the whole slate. At POLL_MINUTES=20 that is 72 calls/day, comfortably inside the cap with headroom for reruns. Do NOT lower it below ~15 without a paid plan; the API answers 429 with a Retry-After once the cap is hit. SETUP. Add your key to E:\\soccer\\.env as: PINNAPI_KEY=xxxxxxxx Type it into the file directly -- do not paste keys into a chat window. Usage: python main\\pin_fresh_capture.py [--once] [--dry] """ import csv import datetime as dt import json import os import sys import time from pathlib import Path import requests from dotenv import load_dotenv sys.path.insert(0, r'E:\soccer\main') load_dotenv(r'E:\soccer\.env') import live_monitor as LM OUT = Path(r'E:\soccer\research\pin_fresh_capture.csv') BASE = 'https://pinnapi.com/kit/v1/prematch/fixtures' KEY = os.getenv('PINNAPI_KEY') POLL_MINUTES = 20 SPORT_SOCCER = 1 KO_TOLERANCE_MIN = 90 # our watchlist kickoff vs Pinnacle's, allowing for # scheduling drift and timezone rounding FIELDS = ['ts', 'lc', 'home', 'away', 'game_date', 'kickoff_utc', 'pin_event_id', 'pin_league', 'pin_home_name', 'pin_away_name', 'fresh_h', 'fresh_d', 'fresh_a', 'stale_h', 'stale_d', 'stale_a', 'fid', 'match_score', # book mids captured at the SAME instant as both lines, so the two # CLVs differ only by which Pinnacle line they were measured against 'k_home_mid', 'k_away_mid', 'p_home_mid', 'p_away_mid', # devigged fair + CLV under each line, and the xlow verdict each gives 'fair_fresh_h', 'fair_fresh_a', 'fair_stale_h', 'fair_stale_a', 'clv_fresh_k_h', 'clv_fresh_k_a', 'clv_fresh_p_h', 'clv_fresh_p_a', 'clv_stale_k_h', 'clv_stale_k_a', 'clv_stale_p_h', 'clv_stale_p_a', 'xlow_fresh', 'xlow_stale', 'verdict'] def devig(h, d, a): """Proportional devig -> (fair_home, fair_draw, fair_away). Mirrors the live pipeline's method exactly so any difference in the comparison comes from the LINE, never from the maths applied to it.""" try: imp = [1.0 / float(h), 1.0 / float(d), 1.0 / float(a)] except (TypeError, ValueError, ZeroDivisionError): return None, None, None s = sum(imp) if s <= 0: return None, None, None return imp[0] / s, imp[1] / s, imp[2] / s def _clv(fair, mid): if fair is None or mid is None or fair <= 0: return None return (fair - mid) / fair def xlow_verdict(fair_h, fair_a, kh, ka, ph, pa): """Which side/book the frozen xlow rule fires on, or '' for no signal. Reimplemented here rather than imported because live_monitor's version reads a snapshot row keyed by column name; the RULE is identical -- bet the cheaper book when its CLV is in [5%,12%) and the other book is NOT in band, home/away only. """ lo, hi = LM.EDGE_THRESH_HOME, LM.EDGE_UPPER_CAP_HOME inband = lambda c: c is not None and lo <= c < hi for side, fair, k_mid, p_mid in (('home', fair_h, kh, ph), ('away', fair_a, ka, pa)): if not (k_mid and p_mid and 0 < k_mid < 1 and 0 < p_mid < 1): continue ck, cp = _clv(fair, k_mid), _clv(fair, p_mid) if k_mid < p_mid and inband(ck) and not inband(cp): return f'{side}/kalshi' if p_mid < k_mid and inband(cp) and not inband(ck): return f'{side}/poly' return '' def norm(s): return ''.join(ch for ch in (s or '').lower() if ch.isalnum()) def tokens(s): return {t for t in ''.join( ch if ch.isalnum() else ' ' for ch in (s or '').lower()).split() if len(t) > 2} # ── Full-board capture (added 2026-08-07) ───────────────────────────────── # One call returns EVERY prematch soccer event, so restricting the capture to # watchlist fixtures threw away ~95% of a payload we had already paid for. The # board log keeps all of it, which means a fixture can be studied even if it was # never on the watchlist, and a league can be back-tested before it is wired up. # # DELTA SYNC. Each response carries a `last` cursor; passing it back as `since` # returns ONLY events whose odds changed. Same 1 call either way -- but the # payload collapses from ~2,400 events to the handful that actually moved, so # the board log records line MOVEMENTS rather than 72 identical snapshots a day. # That is precisely the series the stale-vs-fresh question needs. # # QUOTA. pinnapi returns no rate-limit headers at all (verified 2026-08-07), so # unlike API-Football's /status there is no way to ask how much is left. We # therefore count our own calls per UTC day and refuse to poll past DAILY_BUDGET. BOARD = Path(r'E:\soccer\research\pin_board_log.csv') STATE = Path(r'E:\soccer\data\pinnapi_state.json') DAILY_BUDGET = 85 # of a 100/day free tier; 15 held back for reruns BOARD_FIELDS = ['ts', 'event_id', 'league_id', 'league_name', 'starts', 'home', 'away', 'h', 'd', 'a'] def _state(): try: return json.loads(STATE.read_text(encoding='utf-8')) except Exception: return {} def _save_state(s): try: STATE.parent.mkdir(parents=True, exist_ok=True) STATE.write_text(json.dumps(s, indent=2), encoding='utf-8') except OSError: pass def _budget_ok(st): """True if we may spend a call today. Self-enforced -- see QUOTA above.""" today = dt.datetime.now(dt.timezone.utc).strftime('%Y-%m-%d') if st.get('day') != today: st['day'], st['calls'] = today, 0 return st['calls'] < DAILY_BUDGET def fetch_fresh(since=None): """(events, last_cursor). With `since`, returns only CHANGED events.""" if not KEY: raise RuntimeError('PINNAPI_KEY not set in .env') params = {'sport_id': SPORT_SOCCER} if since: params['since'] = since r = requests.get(BASE, params=params, headers={'x-portal-apikey': KEY}, timeout=45) if r.status_code == 429: raise RuntimeError(f'rate limited: {r.text[:120]}') r.raise_for_status() j = r.json() return j.get('events', []), j.get('last') def log_board(events, now, st): """Append events whose money line MOVED since we last saw them. Deduping here rather than logging every snapshot is what keeps a 20-minute full-board capture affordable: ~2,400 events x 72 polls would be ~170k rows a day of mostly identical numbers, where the movements are a small fraction of that. The line we did NOT log is recoverable -- it is whatever the last logged row for that event_id says. Returns rows written. """ seen = st.setdefault('seen', {}) # event_id -> "h/d/a" last logged new = not BOARD.exists() n = 0 with open(BOARD, 'a', newline='', encoding='utf-8') as fh: w = csv.DictWriter(fh, fieldnames=BOARD_FIELDS) if new: w.writeheader() for e in events: ml = ((e.get('periods') or {}).get('num_0') or {}).get('money_line') or {} if not all(ml.get(k) for k in ('home', 'draw', 'away')): continue eid = str(e.get('event_id')) sig = f"{ml.get('home')}/{ml.get('draw')}/{ml.get('away')}" if seen.get(eid) == sig: continue # unchanged since last poll seen[eid] = sig w.writerow({'ts': now.isoformat(), 'event_id': e.get('event_id'), 'league_id': e.get('league_id'), 'league_name': e.get('league_name'), 'starts': e.get('starts'), 'home': e.get('home'), 'away': e.get('away'), 'h': ml.get('home'), 'd': ml.get('draw'), 'a': ml.get('away')}) n += 1 # Drop events that have kicked off so `seen` cannot grow without bound. cutoff = now - dt.timedelta(hours=6) live_ids = {str(e.get('event_id')) for e in events if (_pt(e.get('starts')) or now) > cutoff} st['seen'] = {k: v for k, v in seen.items() if k in live_ids} return n def _pt(s): try: d = dt.datetime.fromisoformat(str(s).replace('Z', '+00:00')) except (ValueError, TypeError): return None return d if d.tzinfo else d.replace(tzinfo=dt.timezone.utc) def match_events(matches, events): """Pair our watchlist fixtures to Pinnacle events. Team names differ between providers (API-Football vs Pinnacle), so match on kickoff proximity FIRST -- that is the strong, provider-independent signal -- then require some token overlap on the names. A pairing that agrees on time but shares no team token is rejected rather than guessed at; a wrong pairing would silently compare two different matches' odds, which is worse than a missing row. """ idx = [] for e in events: ml = ((e.get('periods') or {}).get('num_0') or {}).get('money_line') or {} if not all(ml.get(k) for k in ('home', 'draw', 'away')): continue st = _pt(e.get('starts')) if st is None: continue idx.append((e, st, ml, tokens(e.get('home')) | tokens(e.get('away')))) out = [] for m in matches: gd = m.get('game_date') if isinstance(gd, str): gd = dt.date.fromisoformat(gd) try: hh, mi = str(m['kickoff']).split(':')[:2] ko = dt.datetime(gd.year, gd.month, gd.day, int(hh), int(mi), tzinfo=dt.timezone(dt.timedelta(hours=-4))) except Exception: continue ours = tokens(m.get('home')) | tokens(m.get('away')) best, bscore = None, 0 for e, st, ml, toks in idx: if abs((st - ko).total_seconds()) > KO_TOLERANCE_MIN * 60: continue score = len(ours & toks) if score > bscore: best, bscore = (e, st, ml), score if best and bscore >= 1: out.append((m, best[0], best[2], bscore, ko)) return out def cycle(dry=False): _, matches = LM.load_watchlist_file() st = _state() if not _budget_ok(st): print(f" [pinnapi] daily budget spent ({st.get('calls')}/{DAILY_BUDGET}) " f"-- skipping this poll") return 0 # ALWAYS a full board. `since` delta sync costs the same single call but # returns only changed events, which would leave the watchlist comparison # below with a near-empty board on most polls -- and that comparison, at # full 20-minute resolution, is the whole point of this capture. Storage is # controlled instead by deduping in log_board(): a full snapshot goes in, # only line MOVEMENTS come out. events, last = fetch_fresh() now = dt.datetime.now(dt.timezone.utc) st['calls'] = st.get('calls', 0) + 1 if last: st['last'] = last wrote = 0 if dry else log_board(events, now, st) if not dry: _save_state(st) pairs = match_events(matches, events) print(f'{now:%H:%M:%S} {len(events)} events ' f'({wrote} moved, call {st["calls"]}/{DAILY_BUDGET}), ' f'{len(matches)} watchlist fixtures -> {len(pairs)} matched') if dry: for m, e, ml, sc, ko in pairs: print(f" {m['lc']:5s} {m['home'][:18]:18s} <-> {e['home'][:18]:18s} " f"(score {sc}) {ml['home']}/{ml['draw']}/{ml['away']}") unmatched = [m for m in matches if m not in [p[0] for p in pairs]] for m in unmatched[:10]: print(f" UNMATCHED {m['lc']:5s} {m['home'][:20]} v {m['away'][:20]}") return len(pairs) new = not OUT.exists() with open(OUT, 'a', newline='', encoding='utf-8') as fh: w = csv.DictWriter(fh, fieldnames=FIELDS) if new: w.writeheader() for m, e, ml, sc, ko in pairs: stale = {} if m.get('fid'): try: # READ THE CACHE, NEVER REFRESH IT (2026-09-07). # # This column is the STALE line -- the number the live # pipeline is actually using -- so the right value is # whatever live_monitor last fetched, not a fresh pull. # # With the default TTL this ran every 20 min against a # 1-hour cache and so forced a refetch on roughly one run # in three: ~86 fixtures x 24 refreshes = ~2,070 API calls # a day, 28% of the 7,500 plan, spent re-reading a line it # wants stale. It also biased the experiment, because a # line this script refreshed is by definition less stale # than the one the pipeline sees. # # max_age is generous rather than infinite so a genuinely # ancient entry still gets replaced. _, pin = LM.fetch_prematch_odds(m['fid'], max_age=86400) stale = pin or {} except Exception: stale = {} # Book mids at THIS instant -- both CLVs must be measured against # the same prices or the comparison confounds line staleness with # book movement between two fetches. gd_ = m.get('game_date') if isinstance(gd_, str): try: gd_ = dt.date.fromisoformat(gd_) except ValueError: gd_ = None kb = pb = {} try: kb = LM.fetch_kalshi(m['home'], m['away'], m['lc'], gd_) or {} except Exception: kb = {} try: pb = LM.fetch_poly(m['home'], m['away'], m['lc'], gd_) or {} except Exception: pb = {} kh, ka = kb.get('kalshi_home_mid'), kb.get('kalshi_away_mid') ph, pa = pb.get('poly_home_mid'), pb.get('poly_away_mid') ffh, _, ffa = devig(ml.get('home'), ml.get('draw'), ml.get('away')) fsh, _, fsa = devig(stale.get('h'), stale.get('d'), stale.get('a')) x_fresh = xlow_verdict(ffh, ffa, kh, ka, ph, pa) x_stale = xlow_verdict(fsh, fsa, kh, ka, ph, pa) if x_fresh == x_stale: verdict = 'same' elif x_fresh and not x_stale: verdict = 'fresh_only' elif x_stale and not x_fresh: verdict = 'stale_only' else: verdict = 'differ' # PUBLISH THE LINE FOR THE LIVE PIPELINE (2026-09-07). # # This script already pulls the whole prematch slate in ONE pinnapi # call. Writing each fixture's Pinnacle line into the shared cache # turns that single free call into the odds source for every # fixture it covers, so live_monitor stops spending an # API-Football call per fixture on them. # # ONLY score >= 2. Single-token joins are how a line for the wrong # match gets in: scored against 1,303 settled fixtures they were # the entire source of a spurious +0.07 Brier gap, while joins at # score >= 2 are indistinguishable from API-Football's own line. # A wrong line here would silently corrupt pin_fair for a real bet. try: if m.get('fid') and sc >= 2 and ml.get('home') and ml.get('draw') and ml.get('away'): LM._shared_put(f"pinodds:{m['fid']}", {'h': float(ml['home']), 'd': float(ml['draw']), 'a': float(ml['away'])}) except Exception: pass # publishing is best-effort; never break the capture w.writerow({ 'ts': now.isoformat(), 'lc': m.get('lc'), 'home': m.get('home'), 'away': m.get('away'), 'game_date': str(m.get('game_date')), 'kickoff_utc': ko.astimezone( dt.timezone.utc).isoformat(), 'pin_event_id': e.get('event_id'), 'pin_league': e.get('league_name'), 'pin_home_name': e.get('home'), 'pin_away_name': e.get('away'), 'fresh_h': ml.get('home'), 'fresh_d': ml.get('draw'), 'fresh_a': ml.get('away'), 'stale_h': stale.get('h'), 'stale_d': stale.get('d'), 'stale_a': stale.get('a'), 'fid': m.get('fid'), 'match_score': sc, 'k_home_mid': kh, 'k_away_mid': ka, 'p_home_mid': ph, 'p_away_mid': pa, 'fair_fresh_h': ffh, 'fair_fresh_a': ffa, 'fair_stale_h': fsh, 'fair_stale_a': fsa, 'clv_fresh_k_h': _clv(ffh, kh), 'clv_fresh_k_a': _clv(ffa, ka), 'clv_fresh_p_h': _clv(ffh, ph), 'clv_fresh_p_a': _clv(ffa, pa), 'clv_stale_k_h': _clv(fsh, kh), 'clv_stale_k_a': _clv(fsa, ka), 'clv_stale_p_h': _clv(fsh, ph), 'clv_stale_p_a': _clv(fsa, pa), 'xlow_fresh': x_fresh, 'xlow_stale': x_stale, 'verdict': verdict}) return len(pairs) def main(): once = '--once' in sys.argv dry = '--dry' in sys.argv while True: try: cycle(dry=dry) except Exception as e: print(f'{dt.datetime.now():%H:%M:%S} error: {e}') if once or dry: break time.sleep(POLL_MINUTES * 60) if __name__ == '__main__': main()