global_func/contest_pricing_api.py

from datetime import datetime, timezone
from typing import Optional, Tuple, Union

import pandas as pd
import pytz

try:
    from curl_cffi import requests as http_requests
    USE_IMPERSONATE = True
except Exception:
    import requests as http_requests
    USE_IMPERSONATE = False


SPORT_API_MAP = {
    "NASCAR": "NAS",
    "GOLF": "GOLF",
    "MLB": "MLB",
    "NBA": "NBA",
    "NHL": "NHL",
    "NFL": "NFL",
    "MMA": "MMA",
}

# MLB / NBA / NHL: only list contests whose slate date is "today" in US/Eastern (excludes next-day slates).
_TODAY_SLATE_SPORTS = frozenset({"MLB", "NBA", "NHL"})
_EASTERN = pytz.timezone("US/Eastern")


def _parse_sd_to_eastern_date_strings(start_raw) -> Tuple[Optional[str], Optional[str]]:
    """Return (YYYYMMDD, display YYYY-MM-DD) in US/Eastern from DK `sd` field, or (None, None)."""
    if start_raw is None:
        return None, None
    try:
        ts_ms = int(str(start_raw)[6:-2])
        dt_utc = datetime.fromtimestamp(ts_ms / 1000.0, tz=timezone.utc)
        dt_eastern = dt_utc.astimezone(_EASTERN)
        return dt_eastern.strftime("%Y%m%d"), dt_eastern.strftime("%Y-%m-%d")
    except Exception:
        return None, None


def _today_yyyymmdd_eastern() -> str:
    return datetime.now(_EASTERN).strftime("%Y%m%d")


def _contest_type_matches(game_type: str, type_var: str) -> bool:
    if type_var == "Showdown":
        return game_type in {"Showdown", "Showdown Captain Mode"}
    return game_type in {"Classic", "Single Match", "Short Slate"}


def _clean_contest_name(name: str) -> str:
    name = name or ""
    return name.strip()


def _http_get_json(url: str, timeout: int = 20) -> dict:
    if USE_IMPERSONATE:
        response = http_requests.get(url, impersonate="chrome", timeout=timeout)
    else:
        response = http_requests.get(url, timeout=timeout)
    response.raise_for_status()
    return response.json()


def fetch_contests_for_selection(sport_var: str, type_var: str) -> list[dict]:
    sport_code = SPORT_API_MAP.get(sport_var, sport_var)
    data = _http_get_json(f"https://www.draftkings.com/lobby/getcontests?sport={sport_code}")
    contests = data.get("Contests", [])

    filtered = []
    for contest in contests:
        name = _clean_contest_name(contest.get("n"))
        game_type = contest.get("gameType", "")
        if not name:
            continue
        if "-Player" in name or "50-50" in name or "Winner Take All" in name:
            continue
        if not _contest_type_matches(game_type, type_var):
            continue

        start_raw = contest.get("sd")
        contest_date, contest_date_display = _parse_sd_to_eastern_date_strings(start_raw)
        if sport_var in _TODAY_SLATE_SPORTS:
            if not contest_date or contest_date != _today_yyyymmdd_eastern():
                continue

        filtered.append(
            {
                "contest_name": name,
                "contest_id": contest.get("id"),
                "contest_date": contest_date or "",
                "contest_date_display": contest_date_display or "",
                "game_type": game_type,
                "prize_pool": contest.get("po", 0),
            }
        )

    filtered.sort(key=lambda x: x.get("prize_pool", 0), reverse=True)
    return filtered


def fetch_pricing_for_contest(contest_id: Union[int, str]) -> pd.DataFrame:
    contest_data = _http_get_json(
        f"https://api.draftkings.com/contests/v1/contests/{contest_id}?format=json"
    )
    contest_detail = contest_data.get("contestDetail", {})
    draft_group_id = contest_detail.get("draftGroupId")
    if draft_group_id is None:
        raise ValueError("No draftGroupId returned for selected contest.")

    draftables_data = _http_get_json(
        f"https://api.draftkings.com/draftgroups/v1/draftgroups/{draft_group_id}/draftables"
    ).get("draftables", [])

    rows = []
    for draftable in draftables_data:
        rows.append(
            {
                "Name": draftable.get("displayName"),
                "ID": draftable.get("draftableId"),
                "Roster Position": draftable.get("position"),
                "Salary": draftable.get("salary"),
                "Team": draftable.get("teamAbbreviation"),
            }
        )

    pricing_df = pd.DataFrame(rows)
    if pricing_df.empty:
        return pricing_df

    pricing_df = pricing_df.dropna(subset=["Name", "ID", "Roster Position", "Salary"])
    pricing_df = pricing_df.drop_duplicates(subset=["Name", "ID"])
    return pricing_df.reset_index(drop=True)