odds_workspace.py

#!/usr/bin/env python3
"""Export one ParlayAPI odds response to CSV. Python 3.9+, standard library only.

After installing parlay-api, the same starter runs as:
    python -m parlay_api
    python -m parlay_api --full --markets h2h,spreads,totals > odds.csv

No key or package installation is needed for the default demo request:
    python3 odds_workspace.py > odds.csv
    python3 odds_workspace.py --sport americanfootball_nfl > odds.csv

For an explicit full request, set PARLAY_API_KEY in your terminal first:
    python3 odds_workspace.py --full --markets h2h,spreads,totals > odds.csv
    python3 odds_workspace.py --full --no-include-live > pregame.csv

Demo mode ignores PARLAY_API_KEY and is always first-five US moneyline events.
Full mode includes commenced games by default, with include_live explicit in the
request. There are no automatic retries, polling, redirects or key persistence.
CSV contains source prices and timestamps, with ambiguous groups excluded.
Configure a matching request visually: https://parlay-api.com/playground#project

MIT License. Copyright (c) 2026 Jacob. See the repository LICENSE.
"""

import argparse
import csv
import io
import json
import math
import os
import re
import sys
from collections import Counter
from datetime import datetime
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, build_opener, HTTPRedirectHandler

USER_AGENT = "ParlayAPI-odds-workspace/1.0"
SPORTS = {
    "baseball_mlb",
    "basketball_nba",
    "americanfootball_nfl",
    "icehockey_nhl",
    "soccer_epl",
    "mma_mixed_martial_arts",
}
MARKETS = {"h2h", "spreads", "totals"}
CSV_FIELDS = [
    "event_id",
    "sport_key",
    "home_team",
    "away_team",
    "commence_time",
    "bookmaker_key",
    "bookmaker_title",
    "bookmaker_last_update",
    "market_key",
    "market_last_update",
    "outcome_name",
    "point",
    "price",
]


def is_number(value):
    return type(value) in (int, float) and math.isfinite(value)


def is_text(value):
    return (
        isinstance(value, str)
        and 0 < len(value) <= 200
        and value.strip() == value
        and not re.search(r"[\x00-\x1f\x7f]", value)
    )


def is_key(value):
    return (
        isinstance(value, str)
        and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:-]{0,159}", value) is not None
    )


def inspect_sample(payload):
    rows, issues = [], []
    source = (
        payload
        if isinstance(payload, list)
        else payload.get("events")
        if isinstance(payload, dict) and payload.get("demo") is True
        else None
    )
    if not isinstance(source, list) or len(source) > 5000:
        raise ValueError("Expected an event array or demo events array, maximum 5,000 events")

    def issue(code, event_id=None, book_key=None, market_key=None):
        issues.append(
            {"code": code, "eventId": event_id, "bookmakerKey": book_key, "marketKey": market_key}
        )

    def counts(items, field):
        return Counter(
            item[field] for item in items if isinstance(item, dict) and is_key(item.get(field))
        )

    def stamp(value, event_id, book_key=None, market_key=None):
        if value is None:
            return None
        try:
            if not is_text(value) or not re.fullmatch(
                r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})", value
            ):
                raise ValueError()
            datetime.fromisoformat(value.replace("Z", "+00:00"))
            return value
        except ValueError:
            issue("invalid_timestamp", event_id, book_key, market_key)
            return None

    ids, canonical_ids = counts(source, "id"), counts(source, "canonical_event_id")
    seen = set()
    for event in source:
        if (
            not isinstance(event, dict)
            or not is_key(event.get("id"))
            or not isinstance(event.get("sport_key"), str)
            or event["sport_key"] not in SPORTS
            or not is_text(event.get("home_team"))
            or not is_text(event.get("away_team"))
            or event["home_team"] == event["away_team"]
            or not isinstance(event.get("bookmakers"), list)
            or len(event["bookmakers"]) > 1000
        ):
            issue(
                "invalid_event",
                event.get("id") if isinstance(event, dict) and is_key(event.get("id")) else None,
            )
            continue
        event_id = event["id"]
        if ids[event_id] > 1 or (
            is_key(event.get("canonical_event_id"))
            and canonical_ids[event["canonical_event_id"]] > 1
        ):
            if event_id not in seen:
                issue("duplicate_event", event_id)
            seen.add(event_id)
            continue
        commence = stamp(event.get("commence_time"), event_id)
        book_counts, visited_books = counts(event["bookmakers"], "key"), set()
        for book in event["bookmakers"]:
            if (
                not isinstance(book, dict)
                or not is_key(book.get("key"))
                or not isinstance(book.get("markets"), list)
                or len(book["markets"]) > 100
            ):
                issue("invalid_bookmaker", event_id)
                continue
            book_key = book["key"]
            if book_counts[book_key] > 1:
                if book_key not in visited_books:
                    issue("duplicate_bookmaker", event_id, book_key)
                visited_books.add(book_key)
                continue
            book_stamp = stamp(book.get("last_update"), event_id, book_key)
            market_counts, visited_markets = counts(book["markets"], "key"), set()
            for market in book["markets"]:
                if (
                    not isinstance(market, dict)
                    or not isinstance(market.get("key"), str)
                    or market["key"] not in MARKETS
                ):
                    issue("unsupported_market", event_id, book_key)
                    continue
                market_key = market["key"]
                if market_counts[market_key] > 1:
                    if market_key not in visited_markets:
                        issue("duplicate_market", event_id, book_key, market_key)
                    visited_markets.add(market_key)
                    continue
                outcomes = market.get("outcomes")
                valid = isinstance(outcomes, list) and 2 <= len(outcomes) <= (
                    3 if market_key == "h2h" else 2
                )
                names = set()
                if valid:
                    for outcome in outcomes:
                        if (
                            not isinstance(outcome, dict)
                            or not is_text(outcome.get("name"))
                            or outcome["name"] in names
                            or not is_number(outcome.get("price"))
                            or abs(outcome["price"]) < 100
                        ):
                            valid = False
                            break
                        names.add(outcome["name"])
                        if market_key == "h2h":
                            valid = (
                                valid
                                and outcome.get("point") is None
                                and outcome["name"]
                                in (event["home_team"], event["away_team"], "Draw")
                            )
                        elif market_key == "spreads":
                            valid = (
                                valid
                                and is_number(outcome.get("point"))
                                and outcome["name"] in (event["home_team"], event["away_team"])
                            )
                        else:
                            valid = (
                                valid
                                and is_number(outcome.get("point"))
                                and outcome["name"] in ("Over", "Under")
                            )
                    if market_key in ("h2h", "spreads"):
                        valid = (
                            valid and event["home_team"] in names and event["away_team"] in names
                        )
                    if valid and market_key == "spreads":
                        valid = outcomes[0]["point"] == -outcomes[1]["point"]
                    if valid and market_key == "totals":
                        valid = (
                            "Over" in names
                            and "Under" in names
                            and outcomes[0]["point"] == outcomes[1]["point"]
                        )
                if not valid:
                    issue("invalid_outcomes", event_id, book_key, market_key)
                    continue
                market_stamp = stamp(market.get("last_update"), event_id, book_key, market_key)
                for outcome in outcomes:
                    rows.append(
                        {
                            "event_id": event_id,
                            "sport_key": event["sport_key"],
                            "home_team": event["home_team"],
                            "away_team": event["away_team"],
                            "commence_time": commence,
                            "bookmaker_key": book_key,
                            "bookmaker_title": book.get("title")
                            if is_text(book.get("title"))
                            else None,
                            "bookmaker_last_update": book_stamp,
                            "market_key": market_key,
                            "market_last_update": market_stamp,
                            "outcome_name": outcome["name"],
                            "point": outcome.get("point"),
                            "price": outcome["price"],
                        }
                    )
    return rows, issues


def to_csv(rows):
    def cell(value):
        if value is None:
            return ""
        if is_number(value):
            return value
        value = str(value)
        if re.match(r"^[\s\ufeff]*[=+\-@]", value) or re.match(r"^[\t\r\n]", value):
            value = "'" + value
        return value

    output = io.StringIO(newline="")
    writer = csv.writer(output, lineterminator="\r\n")
    writer.writerow(CSV_FIELDS)
    for row in rows:
        writer.writerow([cell(row.get(field)) for field in CSV_FIELDS])
    return output.getvalue()


class NoRedirect(HTTPRedirectHandler):
    # Never forward the API key to a redirect destination.
    def redirect_request(self, request, fp, code, msg, headers, newurl):
        return None


def parse_markets(value):
    requested = value.split(",")
    if not requested or any(market not in MARKETS for market in requested):
        raise argparse.ArgumentTypeError("use h2h, spreads or totals, separated by commas")
    return [market for market in ("h2h", "spreads", "totals") if market in requested]


def parse_args(argv=None, prog=None):
    parser = argparse.ArgumentParser(
        prog=prog, description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "--sport",
        choices=sorted(SPORTS),
        default="baseball_mlb",
        help="one sport; default: baseball_mlb",
    )
    parser.add_argument(
        "--full",
        action="store_true",
        help="make a full API request using PARLAY_API_KEY; consumes account credits",
    )
    parser.add_argument(
        "--markets",
        type=parse_markets,
        default=["h2h"],
        help="full mode: h2h,spreads,totals; default: h2h",
    )
    parser.add_argument(
        "--include-live",
        action=argparse.BooleanOptionalAction,
        default=None,
        help="full mode: include commenced games; enabled by default; --no-include-live requests pregame",
    )
    args = parser.parse_args(argv)
    if not args.full and (args.markets != ["h2h"] or args.include_live is not None):
        parser.error(
            "market and live filters require --full. The demo always returns US moneyline; it accepts no live filter."
        )
    return args


def request_url(args):
    if not args.full:
        return "https://parlay-api.com/v1/try/" + args.sport + "/odds"
    params = {
        "regions": "us",
        "markets": ",".join(args.markets),
        "oddsFormat": "american",
        "dateFormat": "iso",
        "include_live": "false" if args.include_live is False else "true",
    }
    return "https://parlay-api.com/v1/sports/" + args.sport + "/odds?" + urlencode(params)


class ResponseContractError(ValueError):
    """A response does not match the explicit request mode or filters."""


def validate_response(payload, args):
    if args.full:
        if not isinstance(payload, list):
            raise ResponseContractError("full mode requires an event array")
        events = payload
    else:
        if (
            not isinstance(payload, dict)
            or payload.get("demo") is not True
            or not isinstance(payload.get("events"), list)
            or len(payload["events"]) > 5
        ):
            raise ResponseContractError(
                "demo mode requires a demo wrapper with at most five events"
            )
        events = payload["events"]
    if any(not isinstance(event, dict) or event.get("sport_key") != args.sport for event in events):
        raise ResponseContractError("returned events do not match the requested sport")
    rows, issues = inspect_sample(payload)
    if any(row["market_key"] not in args.markets for row in rows):
        raise ResponseContractError("returned markets do not match the requested markets")
    return rows, issues


def main(argv=None, prog=None):
    args = parse_args(argv, prog=prog)
    headers = {"Accept": "application/json", "User-Agent": USER_AGENT}
    if args.full:
        api_key = os.environ.get("PARLAY_API_KEY", "").strip()
        if not api_key or "\n" in api_key or "\r" in api_key:
            print(
                "Set PARLAY_API_KEY in your terminal for --full. Or omit --full to try the no-key demo.",
                file=sys.stderr,
            )
            return 2
        headers["X-API-Key"] = api_key
        print(
            "Full API: one US-region request with explicit markets and include_live. Account credits apply.",
            file=sys.stderr,
        )
    else:
        print(
            "Demo: first five US moneyline events; shared limit 60 requests/hour per IP. PARLAY_API_KEY is ignored.",
            file=sys.stderr,
        )
    request = Request(request_url(args), headers=headers)
    try:
        with build_opener(NoRedirect()).open(request, timeout=30) as response:
            if response.status != 200:
                raise ValueError("Expected HTTP 200")
            body = response.read(10_000_001)
            if len(body) > 10_000_000:
                raise ValueError("Response exceeds the 10 MB safety limit")
            payload = json.loads(body)
        rows, issues = validate_response(payload, args)
    except HTTPError as error:
        retry = error.headers.get("Retry-After") if error.headers else None
        suffix = " Retry-After: " + str(retry)[:80] if error.code == 429 and retry else ""
        if args.full:
            hint = ". Check your key, allowance and request settings."
        elif error.code == 403:
            hint = ". The anonymous demo request was denied; this endpoint does not require an API key."
        else:
            hint = (
                ". The no-key demo request failed. Check service availability and request settings."
            )
        print(
            "Request failed: HTTP " + str(error.code) + hint + suffix,
            file=sys.stderr,
        )
        error.close()
        return 1
    except ResponseContractError as error:
        print("Response validation failed: " + str(error) + ". No CSV exported.", file=sys.stderr)
        return 1
    except (URLError, TimeoutError, OSError, ValueError, TypeError) as error:
        print(
            "Request or response validation failed ("
            + type(error).__name__
            + "). Check connectivity and the API response. No automatic retry.",
            file=sys.stderr,
        )
        return 1
    sys.stdout.write(to_csv(rows))
    print(
        str(len(rows))
        + " source rows exported; "
        + str(len(issues))
        + " validation issues. Timestamps are source fields, not freshness guarantees.",
        file=sys.stderr,
    )
    for issue in issues[:25]:
        print(json.dumps(issue, ensure_ascii=True), file=sys.stderr)
    if not rows:
        print(
            "No usable rows returned. This response does not establish coverage or explain why data is absent.",
            file=sys.stderr,
        )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())