#!/usr/bin/env python3 """A chronological baseline for NBA points from retained game samples. Independent service, not affiliated with or endorsed by DraftKings. No wagering. Run: python3 nba-points-baseline.py --season 2027 --min-games 5 --max-pages 5 This measures a prior-games average against a constant, using only games before the one predicted. Samples are small and were captured after the games, so the output is a code check, not evidence of skill. """ import argparse import hashlib import json import os import sys import time from datetime import datetime, timezone from pathlib import Path from urllib.error import HTTPError, URLError from urllib.parse import urlencode from urllib.request import Request, urlopen import sqlite3 API_KEY = os.environ.get("DRAFTKINGS_API_KEY") BASE = os.environ.get("DRAFTKINGS_API_BASE", "https://draftkingsapi.com").rstrip("/") CACHE_DIR = Path(os.environ.get("DRAFTKINGS_CACHE_DIR", ".draftkings-cache")) # Collection targets five minutes. Polling faster spends quota without newer data. CACHE_TTL_SECONDS = 300 class ApiError(Exception): def __init__(self, status, message): super().__init__(f"HTTP {status}: {message}") self.status = status class Client: """Bounded client: a request budget, a timeout, a file cache and limited retries.""" def __init__(self, max_requests, use_cache=True): if not API_KEY: sys.exit("Set DRAFTKINGS_API_KEY in your environment. Create a key at https://draftkingsapi.com/dashboard") self.max_requests = max_requests self.requests = 0 self.use_cache = use_cache def get(self, path, **params): query = {k: v for k, v in params.items() if v is not None} url = f"{BASE}{path}?{urlencode(query)}" if query else f"{BASE}{path}" cached = self._read_cache(url) if cached is not None: return cached body = self._fetch(url) self._write_cache(url, body) return body def _fetch(self, url): for attempt in range(3): if self.requests >= self.max_requests: raise ApiError(0, f"Local budget of {self.max_requests} requests reached. Raise it deliberately.") self.requests += 1 request = Request(url, headers={"Authorization": f"Bearer {API_KEY}", "Accept": "application/json"}) try: with urlopen(request, timeout=20) as response: return json.load(response) except HTTPError as error: message = _error_message(error) can_retry = self.requests < self.max_requests if error.code == 429 and attempt == 0 and can_retry: # Minute limits reset at the next UTC minute. A second 429 usually means the monthly allowance. time.sleep(61 - datetime.now(timezone.utc).second) continue if error.code >= 500 and attempt < 2 and can_retry: time.sleep(2 ** attempt) continue raise ApiError(error.code, message) from None except (URLError, TimeoutError) as error: if attempt < 2 and self.requests < self.max_requests: time.sleep(2 ** attempt) continue raise ApiError(0, f"Network failure: {error}") from None raise ApiError(429, "Account request limit reached. Check the allowance in your dashboard.") def _cache_path(self, url): return CACHE_DIR / (hashlib.sha256(url.encode()).hexdigest() + ".json") def _read_cache(self, url): path = self._cache_path(url) if not self.use_cache or not path.exists() or time.time() - path.stat().st_mtime > CACHE_TTL_SECONDS: return None return json.loads(path.read_text()) def _write_cache(self, url, body): if self.use_cache: CACHE_DIR.mkdir(exist_ok=True) self._cache_path(url).write_text(json.dumps(body)) def _error_message(error): try: return json.load(error)["error"]["message"] except Exception: return error.reason def points(row): for stat in row["data"].get("stats", []): if stat["key"] == "points" and stat.get("scope") == "game" and stat.get("numeric") is not None: return float(stat["numeric"]) return None def load_rows(client, db, season, max_pages): db.execute("CREATE TABLE IF NOT EXISTS player_games(event_id TEXT, athlete_id TEXT, name TEXT, starts_at TEXT, points REAL, PRIMARY KEY(event_id, athlete_id))") after = None for _ in range(max_pages): page = client.get("/api/v1/stats/player-games", sport="NBA", season=season, completed="true", limit=200, after=after) for row in page["data"]: value, game = points(row), row.get("game") or {} if value is None or row["data"].get("did_not_play"): continue db.execute("INSERT OR REPLACE INTO player_games VALUES(?,?,?,?,?)", (game.get("event_id"), row["data"]["athlete_id"], row["data"]["name"], game.get("starts_at"), value)) db.commit() after = page["pagination"]["next"] if not after: break def main(): parser = argparse.ArgumentParser() parser.add_argument("--season", type=int, required=True) parser.add_argument("--min-games", type=int, default=5) parser.add_argument("--max-pages", type=int, default=5) parser.add_argument("--db", default="nba-baseline.sqlite") args = parser.parse_args() client = Client(max_requests=args.max_pages) db = sqlite3.connect(args.db) load_rows(client, db, args.season, args.max_pages) rows = db.execute("SELECT athlete_id, name, starts_at, points FROM player_games WHERE starts_at IS NOT NULL ORDER BY starts_at").fetchall() if not rows: sys.exit("No retained completed NBA rows for that season. Check /status.") # Hold out the latest 20% of dates so the constant is learned only from earlier games. cutoff = rows[int(len(rows) * 0.8)][2] train = [r[3] for r in rows if r[2] < cutoff] constant = sum(train) / len(train) if train else 0.0 history, baseline_errors, constant_errors = {}, [], [] for athlete_id, _, starts_at, value in rows: prior = history.setdefault(athlete_id, []) if starts_at >= cutoff and len(prior) >= args.min_games: baseline_errors.append(abs(sum(prior) / len(prior) - value)) constant_errors.append(abs(constant - value)) prior.append(value) print(f"Rows {len(rows)}, athletes {len(history)}, held-out cutoff {cutoff}, requests {client.requests}") if not baseline_errors: print(f"No held-out athlete has {args.min_games} prior games in the retained sample. Lower --min-games to exercise the code, or wait for more data.") return print(f"Held-out predictions: {len(baseline_errors)}") print(f"MAE prior-games mean: {sum(baseline_errors) / len(baseline_errors):.2f}") print(f"MAE constant: {sum(constant_errors) / len(constant_errors):.2f}") print("A small retained sample. This does not establish predictive skill, an edge or profit.") if __name__ == "__main__": try: main() except ApiError as error: sys.exit(str(error))