File size: 4,523 Bytes
9ef68b1
 
27154c4
 
9ef68b1
 
555ed46
 
 
 
 
 
27154c4
 
 
 
 
 
 
 
 
 
 
 
9ef68b1
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
27154c4
 
 
 
 
 
 
 
 
 
 
 
555ed46
 
 
 
 
 
 
 
 
27154c4
 
555ed46
27154c4
 
 
 
 
 
 
 
 
 
 
 
 
 
9ef68b1
 
 
 
27154c4
 
 
 
 
9ef68b1
 
27154c4
 
 
 
 
 
 
 
 
13f323e
555ed46
 
27154c4
 
 
 
 
 
555ed46
 
 
27154c4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
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)