Build an NRL product with Odds API

"""Discover NRL games, load every available odds row, and follow each game.

Install: python -m pip install httpx
Run:     ODDS_API_KEY=your_key python nrl_full_feed.py

The program writes JSON Lines to stdout. Set NRL_PAST_HOURS and
NRL_FUTURE_HOURS to change the rolling discovery window.
"""

from __future__ import annotations

import asyncio
import json
import os
import random
import sys
import time
from typing import Any, AsyncIterator

import httpx


BASE_URL = os.getenv("ODDS_API_BASE_URL", "https://api.odds-api.net/v1").rstrip("/") + "/"
API_KEY = os.getenv("ODDS_API_KEY", "").strip()
PAST_HOURS = int(os.getenv("NRL_PAST_HOURS", "8"))
FUTURE_HOURS = int(os.getenv("NRL_FUTURE_HOURS", "48"))
DISCOVERY_SECONDS = int(os.getenv("NRL_DISCOVERY_SECONDS", "300"))
ODDS_SHAPE = {
    "price_fields": "all",
    "include_unavailable": "true",
    "include_source": "true",
}


def emit(kind: str, **payload: Any) -> None:
    print(json.dumps({"kind": kind, **payload}, separators=(",", ":")), flush=True)


async def get_json(client: httpx.AsyncClient, path: str, params: dict | None = None) -> dict:
    response = await client.get(path, params=params)
    response.raise_for_status()
    return response.json()


async def nrl_events(client: httpx.AsyncClient) -> list[dict]:
    now = int(time.time())
    params = {
        "sport": "rugby league",
        "league": "NRL",
        "start_from": now - PAST_HOURS * 3600,
        "start_to": now + FUTURE_HOURS * 3600,
        "limit": 1000,
    }
    events: list[dict] = []
    while True:
        page = await get_json(client, "events", params)
        events.extend(page.get("items") or [])
        cursor = page.get("next_cursor")
        if not cursor:
            return events
        params["cursor"] = cursor


async def complete_snapshot(client: httpx.AsyncClient, event_id: str) -> tuple[dict, dict[str, dict]]:
    params = {"limit": 10000, **ODDS_SHAPE}
    rows: dict[str, dict] = {}
    first_page: dict | None = None
    while True:
        page = await get_json(client, f"events/{event_id}/odds/snapshot", params)
        if first_page is None:
            first_page = page
        for odd in page.get("items") or []:
            if odd.get("id"):
                rows[str(odd["id"])] = odd
        cursor = page.get("next_cursor")
        if not cursor:
            break
        params["cursor"] = cursor

    # Start the stream at the first page's resume token. Changes that arrive
    # while later snapshot pages load can then be replayed by the stream.
    return first_page or {}, rows


async def sse_messages(response: httpx.Response) -> AsyncIterator[tuple[str, dict]]:
    event = "message"
    data: list[str] = []
    async for line in response.aiter_lines():
        if line == "":
            if data:
                yield event, json.loads("\n".join(data))
            event, data = "message", []
        elif line.startswith("event:"):
            event = line[6:].strip()
        elif line.startswith("data:"):
            data.append(line[5:].lstrip())
    if data:
        yield event, json.loads("\n".join(data))


def apply_delta(rows: dict[str, dict], payload: dict) -> None:
    for change in payload.get("changes") or []:
        odd = change.get("odd") or {}
        row_id = str(odd.get("id") or "")
        if not row_id:
            continue
        if change.get("op") in {"remove", "delete"}:
            rows.pop(row_id, None)
        elif change.get("op") == "upsert":
            # Keep is_available=false rows so the product can show suspensions.
            rows[row_id] = odd


async def follow_event(client: httpx.AsyncClient, event_id: str) -> None:
    resume = ""
    rows: dict[str, dict] = {}
    needs_snapshot = True
    retry_seconds = 1.0

    while True:
        try:
            if needs_snapshot:
                snapshot, rows = await complete_snapshot(client, event_id)
                resume = str(snapshot.get("resume") or "$")
                emit(
                    "odds_snapshot",
                    event_id=event_id,
                    as_of_ts_ms=snapshot.get("as_of_ts_ms"),
                    ttl_seconds=snapshot.get("ttl_seconds"),
                    resume=resume,
                    items=list(rows.values()),
                )
                needs_snapshot = False

            params = {"since": resume, "catchup": "true", "max_batch": 2000, **ODDS_SHAPE}
            async with client.stream(
                "GET", f"events/{event_id}/odds/stream", params=params,
                headers={"Accept": "text/event-stream"},
            ) as response:
                response.raise_for_status()
                retry_seconds = 1.0
                async for event, payload in sse_messages(response):
                    if event == "delta":
                        apply_delta(rows, payload)
                        resume = str(payload.get("resume") or resume)
                        emit("odds_delta", event_id=event_id, resume=resume, changes=payload.get("changes") or [])
                    elif event == "resync":
                        emit("odds_resync", event_id=event_id, reason=payload.get("reason"))
                        needs_snapshot = True
                        break
                    # A heartbeat contains no price change.

        except asyncio.CancelledError:
            raise
        except (httpx.HTTPError, ValueError, json.JSONDecodeError) as exc:
            emit("stream_error", event_id=event_id, error=str(exc))

        if needs_snapshot:
            continue
        await asyncio.sleep(retry_seconds + random.uniform(0, 0.5))
        retry_seconds = min(retry_seconds * 2, 30.0)


async def game_record(client: httpx.AsyncClient, event_id: str) -> None:
    try:
        detail, result = await asyncio.gather(
            get_json(client, f"events/{event_id}"),
            get_json(client, f"events/{event_id}/results"),
        )
        emit("event", event_id=event_id, data=detail)
        emit("result", event_id=event_id, data=result)
    except httpx.HTTPError as exc:
        emit("record_error", event_id=event_id, error=str(exc))


async def bounded_game_record(client: httpx.AsyncClient, semaphore: asyncio.Semaphore, event_id: str) -> None:
    async with semaphore:
        await game_record(client, event_id)


async def main() -> None:
    if not API_KEY:
        raise SystemExit("Set ODDS_API_KEY before running this example.")
    timeout = httpx.Timeout(connect=10.0, read=None, write=10.0, pool=10.0)
    tasks: dict[str, asyncio.Task] = {}
    record_limit = asyncio.Semaphore(6)
    async with httpx.AsyncClient(
        base_url=BASE_URL,
        headers={"X-API-Key": API_KEY, "Accept": "application/json"},
        timeout=timeout,
    ) as client:
        try:
            while True:
                try:
                    events = await nrl_events(client)
                    ids = {str(item["event_id"]) for item in events if item.get("event_id")}
                    emit("event_list", count=len(events), items=events)

                    for event_id in ids:
                        if event_id not in tasks or tasks[event_id].done():
                            tasks[event_id] = asyncio.create_task(follow_event(client, event_id))
                    for event_id in set(tasks) - ids:
                        tasks.pop(event_id).cancel()

                    await asyncio.gather(
                        *(bounded_game_record(client, record_limit, event_id) for event_id in ids)
                    )
                except httpx.HTTPError as exc:
                    emit("discovery_error", error=str(exc))
                await asyncio.sleep(DISCOVERY_SECONDS)
        finally:
            for task in tasks.values():
                task.cancel()
            await asyncio.gather(*tasks.values(), return_exceptions=True)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("Stopped.", file=sys.stderr)