#!/usr/bin/env python3 """ fetch_guide_stats.py -- War-room intel for guide.html. Run alongside 1_fetch_players (draft morning). Pulls, merges, and writes guide_stats.json (served ONLY via the relay's gated /guide_stats endpoint): * FantasyPros (your API key in .fantasypros.env — public tier = 10 rows/request, so we take the top 10 per position): expert consensus rank (ECR), TIER, position rank, rank spread; season PPR projections for those same players; and the latest league-wide news ticker. * ESPN "kona" fantasy API (free): season PPR projections for ~600 players (full-coverage backfill). * ESPN athlete stats (free, threaded): the last 4 seasons per ADP-ranked player, rescored to THIS league's rules (1 pt/25 pass yds, 4 pass TD, -2 INT, 1 pt/10 rush+rec yds, 6 TD, 1 pt/rec, -2 fumble lost) so history is apples-to-apples with the draft. Stdlib only. The API key never leaves this machine; guide_stats.json is blocked from static serving. """ import concurrent.futures, json, os, re, sys, time, urllib.error, urllib.request ROOT = os.path.dirname(os.path.abspath(__file__)) SEASON = 2026 HIST_YEARS = [2022, 2023, 2024, 2025] FP_BASE = "https://api.fantasypros.com/public/v2/json/nfl" def load_key(): path = os.path.join(ROOT, ".fantasypros.env") try: key = None for line in open(path, encoding="utf-8-sig"): line = line.strip() if not line or line.startswith("#"): continue if "=" in line: k, v = line.split("=", 1) if "key" in k.lower() or "api" in k.lower(): key = v.strip().strip('"').strip("'") elif key is None and len(line) > 10: key = line return key except FileNotFoundError: return None # ESPN keeps changing which User-Agents it accepts (it has 403'd both "DraftBoard/1.0" and # browser-style "Mozilla/5.0", while urllib's default works). Rotate through profiles on 403 # and lock onto whichever succeeds; keep the long backoff for FantasyPros' 429 rate limit. UA_PROFILES = [ {}, # urllib default (works 2026-08) {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"}, {"User-Agent": "Mozilla/5.0"}, ] _profile = 0 def get(url, headers=None, timeout=25, tries=4): """GET that survives FantasyPros 429 bursts AND ESPN's shifting User-Agent blocking.""" global _profile last = None for attempt in range(tries): order = sorted(range(len(UA_PROFILES)), key=lambda i: 0 if i == _profile else 1) for i in order: h = dict(UA_PROFILES[i]) h.update(headers or {}) try: req = urllib.request.Request(url, headers=h) with urllib.request.urlopen(req, timeout=timeout) as r: if i != _profile: print(f" (switched to HTTP header profile #{i} after a block)") _profile = i return json.load(r) except urllib.error.HTTPError as e: last = e if e.code == 403: continue # blocked UA -> try the next profile break # 429/other -> fall through to the backoff below except Exception as e: # IncompleteRead/timeout on big payloads -> worth a retry last = e break retryable = isinstance(last, urllib.error.HTTPError) and last.code == 429 retryable = retryable or not isinstance(last, urllib.error.HTTPError) if last is not None and retryable and attempt < tries - 1: time.sleep(2.5 * (attempt + 1)) continue raise last def norm_name(s): s = (s or "").lower().replace(".", "").replace("'", "").replace("-", " ") s = re.sub(r"\b(jr|sr|ii|iii|iv|v)\b", "", s) s = re.sub(r"[^a-z ]", "", s) return " ".join(s.split()) def fnum(x): try: return float(str(x).replace(",", "")) except Exception: return 0.0 # ---------------- FantasyPros ---------------- def fp_rankings(key): out = {} for pos in ("QB", "RB", "WR", "TE", "K", "DST", "ALL"): try: d = get(f"{FP_BASE}/{SEASON}/consensus-rankings?type=draft&scoring=PPR&week=0&position={pos}", {"x-api-key": key}) except Exception as e: print(f" !! FP rankings {pos}: {e}") continue for p in d.get("players", []): k = norm_name(p.get("player_name")) + "|" + (p.get("player_position_id") or "") if (p.get("player_position_id") or "") == "DST": k = "DST|" + (p.get("player_team_id") or "") out.setdefault(k, {}).update({ "ecr": p.get("rank_ecr"), "tier": p.get("tier"), "posRank": p.get("pos_rank"), "rankMin": p.get("rank_min"), "rankMax": p.get("rank_max"), "ecrDelta": p.get("player_ecr_delta"), }) time.sleep(1.3) print(f" FantasyPros rankings: {len(out)} players (top-10 per position — public tier cap)") return out def fp_projections(key): out = {} for pos in ("QB", "RB", "WR", "TE", "K", "DST"): try: d = get(f"{FP_BASE}/{SEASON}/projections?position={pos}&scoring=PPR&week=draft", {"x-api-key": key}) except Exception as e: print(f" !! FP projections {pos}: {e}") continue for p in d.get("players", []): st = p.get("stats") or {} k = norm_name(p.get("name")) + "|" + pos if pos == "DST": k = "DST|" + (p.get("team_id") or "") out[k] = {"fpProj": round(fnum(st.get("points_ppr") or st.get("points")), 1)} time.sleep(1.3) print(f" FantasyPros projections: {len(out)} players") return out def fp_news(key): try: d = get(f"{FP_BASE}/news?limit=10", {"x-api-key": key}) except Exception as e: print(f" !! FP news: {e}") return [] items = [] for it in d.get("items", []): items.append({"when": it.get("created_formated") or "", "desc": (it.get("desc") or "")[:220], "impact": (it.get("impact") or "")[:180]}) print(f" FantasyPros news ticker: {len(items)} items") return items # ---------------- ESPN projections (full coverage) ---------------- def espn_kona(): url = (f"https://lm-api-reads.fantasy.espn.com/apis/v3/games/ffl/seasons/{SEASON}" f"/segments/0/leaguedefaults/3?scoringPeriodId=0&view=kona_player_info") # ESPN truncates very large responses (IncompleteRead ~14MB), so ask for less and step down. d = None for limit in (450, 300, 200): filt = {"players": {"limit": limit, "sortDraftRanks": {"sortPriority": 100, "sortAsc": True, "value": "PPR"}}} try: d = get(url, {"X-Fantasy-Filter": json.dumps(filt)}) break except Exception as e: print(f" !! ESPN kona projections at limit {limit}: {e}") if d is None: return {} POS = {1: "QB", 2: "RB", 3: "WR", 4: "TE", 5: "K", 16: "DST"} out = {} for row in d.get("players", []): p = row.get("player") or {} pos = POS.get(p.get("defaultPositionId")) if not pos: continue proj = actual_last = None season_rows = [s for s in (p.get("stats") or []) if s.get("statSplitTypeId") == 0] pg_rows = [s for s in (p.get("stats") or []) if s.get("statSplitTypeId") == 1] for s in season_rows: if s.get("seasonId") == SEASON and s.get("statSourceId") == 1: proj = s.get("appliedTotal") if s.get("seasonId") == SEASON - 1 and s.get("statSourceId") == 0: actual_last = s.get("appliedTotal") if proj is None: # only per-game splits present -> scale to a 17-game season pg = [s.get("appliedAverage") or s.get("appliedTotal") for s in pg_rows if s.get("seasonId") == SEASON and s.get("statSourceId") == 1] if pg: proj = (sum(pg) / len(pg)) * 17 k = norm_name(p.get("fullName")) + "|" + pos if pos == "DST": k = "DST|" + norm_name(p.get("fullName")) # matched later by name contains out[k] = {"espnProj": round(proj or 0, 1)} if actual_last: out[k]["espnLastYr"] = round(actual_last, 1) print(f" ESPN projections: {len(out)} players") return out # ---------------- ESPN 4-year history, league-scored ---------------- def league_points(cats): """cats: category name -> {label: value}. Returns this league's fantasy points.""" pa, ru, re_, fu = cats.get("passing", {}), cats.get("rushing", {}), cats.get("receiving", {}), {} pts = 0.0 pts += fnum(pa.get("YDS")) / 25.0 + fnum(pa.get("TD")) * 4 - fnum(pa.get("INT")) * 2 pts += fnum(ru.get("YDS")) / 10.0 + fnum(ru.get("TD")) * 6 pts += fnum(re_.get("REC")) * 1 + fnum(re_.get("YDS")) / 10.0 + fnum(re_.get("TD")) * 6 lost = max(fnum(pa.get("LST")), fnum(ru.get("LST")), fnum(re_.get("LST"))) # LST repeats per category — count once pts -= lost * 2 return round(pts, 1) def line_for(pos, cats): pa, ru, re_ = cats.get("passing", {}), cats.get("rushing", {}), cats.get("receiving", {}) if pos == "QB": return f"{int(fnum(pa.get('YDS')))} pa yds · {int(fnum(pa.get('TD')))} TD · {int(fnum(pa.get('INT')))} INT · {int(fnum(ru.get('YDS')))} ru yds" if pos == "RB": return f"{int(fnum(ru.get('CAR')))} car · {int(fnum(ru.get('YDS')))} yds · {int(fnum(ru.get('TD')))} TD · {int(fnum(re_.get('REC')))} rec" return f"{int(fnum(re_.get('REC')))} rec · {int(fnum(re_.get('YDS')))} yds · {int(fnum(re_.get('TD')))} TD · {int(fnum(ru.get('YDS')))} ru yds" def player_history(pid, pos): try: d = get(f"https://site.web.api.espn.com/apis/common/v3/sports/football/nfl/athletes/{pid}/stats", timeout=20) except Exception: return None by_year = {} for cat in d.get("categories", []): name, labels = cat.get("name"), cat.get("labels") or [] if name not in ("passing", "rushing", "receiving"): continue for row in cat.get("statistics", []): yr = (row.get("season") or {}).get("year") if yr not in HIST_YEARS: continue vals = row.get("stats") or [] m = {labels[i]: vals[i] for i in range(min(len(labels), len(vals)))} by_year.setdefault(yr, {"gp": fnum(m.get("GP"))})[name] = m seasons = [] for yr in sorted(by_year): cats = by_year[yr] seasons.append({"yr": yr, "gp": int(cats.get("gp") or 0), "pts": league_points(cats), "line": line_for(pos, cats)}) return seasons or None def add_history(players): targets = [p for p in players if p.get("adp", 999.9) < 900 and str(p["id"]).isdigit() and p["pos"] != "DST"] print(f" fetching {len(targets)} player histories from ESPN (last {len(HIST_YEARS)} seasons)...") out, got = {}, 0 with concurrent.futures.ThreadPoolExecutor(max_workers=12) as ex: futs = {ex.submit(player_history, p["id"], p["pos"]): p for p in targets} for fut in concurrent.futures.as_completed(futs): p = futs[fut] res = fut.result() if res: out[norm_name(p["name"]) + "|" + p["pos"]] = res got += 1 print(f" histories: {got}") return out # ---------------- Fantasy-playoff schedule strength (weeks 15-17) ---------------- PLAYOFF_WEEKS = (15, 16, 17) def playoff_sos(): """Per NFL team: playoff-week opponents + strength (avg opponent 2025 win%). Crude but honest.""" try: st = get(f"https://site.api.espn.com/apis/v2/sports/football/nfl/standings?season={SEASON-1}") except Exception as e: print(f" !! standings: {e}") return {} winpct = {} for child in st.get("children", []): for e in (child.get("standings") or {}).get("entries", []): tid = str((e.get("team") or {}).get("id")) for s in e.get("stats", []): if (s.get("name") or "").lower() == "winpercent": winpct[tid] = float(s.get("value") or 0.5) try: tdata = get("https://site.api.espn.com/apis/site/v2/sports/football/nfl/teams") teams = [t["team"] for t in tdata["sports"][0]["leagues"][0]["teams"]] except Exception as e: print(f" !! teams for SOS: {e}") return {} out = {} for t in teams: tid, abbr = str(t["id"]), t.get("abbreviation", "") try: sc = get(f"https://site.api.espn.com/apis/site/v2/sports/football/nfl/teams/{tid}/schedule?season={SEASON}") except Exception: continue games, vals = [], [] for ev in sc.get("events", []): wk = (ev.get("week") or {}).get("number") stype = ((ev.get("seasonType") or {}).get("type")) if stype != 2 or wk not in PLAYOFF_WEEKS: continue comps = (ev.get("competitions") or [{}])[0].get("competitors", []) me = next((c for c in comps if str((c.get("team") or {}).get("id")) == tid), None) opp = next((c for c in comps if str((c.get("team") or {}).get("id")) != tid), None) if not opp: continue ot = opp.get("team") or {} home = (me or {}).get("homeAway") == "home" games.append((wk, ("vs " if home else "@ ") + (ot.get("abbreviation") or "?"))) vals.append(winpct.get(str(ot.get("id")), 0.5)) if games: games.sort() out[abbr] = {"opps": " · ".join(f"W{w} {o}" for w, o in games), "sos": round(sum(vals) / len(vals), 3)} time.sleep(0.1) print(f" playoff SOS (weeks {PLAYOFF_WEEKS[0]}-{PLAYOFF_WEEKS[-1]}): {len(out)} teams") return out def adp_key(p): return ("DST|" + p["team"]) if p["pos"] == "DST" else (norm_name(p["name"]) + "|" + p["pos"]) def main(): key = load_key() if key: print("FantasyPros key: loaded from .fantasypros.env") else: print("FantasyPros key: NOT FOUND (.fantasypros.env) — FP fields will be empty") try: players = json.load(open(os.path.join(ROOT, "players.json"), encoding="utf-8"))["players"] except Exception as e: print("ERROR: players.json missing — run fetch_players.py first.", e) sys.exit(1) # previous run's ADP snapshot -> momentum deltas on the next run prev_snap, prev_date, prev_all = None, "", {} try: prev = json.load(open(os.path.join(ROOT, "guide_stats.json"), encoding="utf-8")) prev_all = prev prev_snap = prev.get("adpSnap") prev_date = (prev.get("generated") or "")[:10] # carry the OLDEST snapshot forward so momentum spans more than one run-pair if prev.get("adpPrev") and prev["adpPrev"].get("adps"): prev_snap, prev_date = prev["adpPrev"]["adps"], prev["adpPrev"]["date"] except Exception: pass fpR = fp_rankings(key) if key else {} fpP = fp_projections(key) if key else {} ticker = fp_news(key) if key else [] kona = espn_kona() sos = playoff_sos() hist = add_history(players) merged = {} def slot(k): return merged.setdefault(k, {}) for k, v in fpR.items(): slot(k).update(v) for k, v in fpP.items(): slot(k).update(v) for k, v in kona.items(): if k.startswith("DST|"): continue slot(k).update(v) for k, v in hist.items(): slot(k)["seasons"] = v # DST: match kona DST rows to players.json DST teams by name inclusion dst_kona = {k[4:]: v for k, v in kona.items() if k.startswith("DST|")} for p in players: if p["pos"] != "DST": continue nm = norm_name(p["teamName"]) for kn, v in dst_kona.items(): if kn and (kn in nm or nm in kn): slot("DST|" + p["team"]).update(v) break # ---- NEVER lose data: anything a source failed to return falls back to the last run ---- prev_players = (prev_all.get("players") or {}) carried = 0 for k, old_v in prev_players.items(): cur = merged.setdefault(k, {}) for f in ("ecr", "tier", "posRank", "rankMin", "rankMax", "ecrDelta", "fpProj", "espnProj", "espnLastYr", "seasons"): if not cur.get(f) and old_v.get(f): cur[f] = old_v[f] carried += 1 if not sos: sos = prev_all.get("teamPlayoff") or {} if sos: print(" !! playoff SOS unavailable -> reusing the last run's") if not ticker: ticker = prev_all.get("newsTicker") or [] if ticker: print(" !! news ticker unavailable -> reusing the last run's") if carried: print(f" carried forward {carried} stat fields from the last run (sources that didn't answer)") snap = {adp_key(p): p["adp"] for p in players if p.get("adp", 999.9) < 900} out = { "generated": time.strftime("%Y-%m-%d %H:%M:%S"), "season": SEASON, "scoringNote": "History rescored to THIS league: 1/25 pass yd, 4 pass TD, -2 INT, 1/10 rush+rec yd, 6 TD, 1/rec, -2 fum lost. Projections: FantasyPros PPR (top-10/pos) + ESPN PPR (full).", "newsTicker": ticker, "teamPlayoff": sos, "adpSnap": snap, "adpPrev": ({"date": prev_date, "adps": prev_snap} if prev_snap else None), "players": merged, } path = os.path.join(ROOT, "guide_stats.json") with open(path, "w", encoding="utf-8") as f: json.dump(out, f) n_t = sum(1 for v in merged.values() if v.get("tier")) n_pj = sum(1 for v in merged.values() if v.get("espnProj") or v.get("fpProj")) n_h = sum(1 for v in merged.values() if v.get("seasons")) print(f"\nWrote guide_stats.json — {len(merged)} players ({n_t} tiered, {n_pj} with projections, {n_h} with history)") if __name__ == "__main__": try: main() except Exception as e: print("ERROR:", e) sys.exit(1)