| from __future__ import annotations | |
| from pathlib import Path | |
| from typing import Any | |
| import pandas as pd | |
| import requests | |
| FEATURES = ["form", "next_fdr"] | |
| BASE_URL = "https://fantasy.premierleague.com/api" | |
| PLAYER_COLUMNS = ["id", "web_name", "team", "element_type"] | |
| def get_json(url: str, *, timeout: int = 30) -> Any: | |
| """Fetch JSON and raise a useful error for non-successful responses.""" | |
| response = requests.get(url, timeout=timeout) | |
| response.raise_for_status() | |
| return response.json() | |
| def fetch_players() -> pd.DataFrame: | |
| data = get_json(f"{BASE_URL}/bootstrap-static/") | |
| players = pd.DataFrame(data["elements"])[PLAYER_COLUMNS].copy() | |
| return players[players["element_type"].isin([1, 2, 3, 4])].reset_index(drop=True) | |
| def fetch_fixtures() -> pd.DataFrame: | |
| fixtures = pd.DataFrame(get_json(f"{BASE_URL}/fixtures/")).copy() | |
| numeric_columns = [ | |
| "id", "event", "team_h", "team_a", "team_h_difficulty", | |
| "team_a_difficulty", | |
| ] | |
| for column in numeric_columns: | |
| if column in fixtures: | |
| fixtures[column] = pd.to_numeric(fixtures[column], errors="coerce") | |
| return fixtures | |
| def fetch_player_history(player_id: int) -> list[dict[str, Any]]: | |
| data = get_json(f"{BASE_URL}/element-summary/{player_id}/") | |
| return data.get("history", []) | |
| def as_float(value: Any) -> float | None: | |
| try: | |
| result = float(value) | |
| except (TypeError, ValueError): | |
| return None | |
| return result if pd.notna(result) else None | |
| def next_fixture_fdr(team_id: int, fixtures: pd.DataFrame) -> float | None: | |
| """Return the FDR of the first unfinished fixture for a team.""" | |
| upcoming = fixtures[ | |
| (((fixtures["team_h"] == team_id) | (fixtures["team_a"] == team_id)) | |
| & (~fixtures["finished"].astype(bool))) | |
| ].dropna(subset=["event"]) | |
| if upcoming.empty: | |
| return None | |
| fixture = upcoming.sort_values("event").iloc[0] | |
| if fixture["team_h"] == team_id: | |
| return fixture.get("team_h_difficulty") | |
| return fixture.get("team_a_difficulty") | |
| def fixture_fdr(fixture_id: int, was_home: bool, fixtures: pd.DataFrame) -> float | None: | |
| """Return the historical fixture difficulty for a player's fixture.""" | |
| matches = fixtures[fixtures["id"] == fixture_id] | |
| if matches.empty: | |
| return None | |
| fixture = matches.iloc[0] | |
| return fixture.get("team_h_difficulty" if was_home else "team_a_difficulty") | |
| def previous_form(points: list[float], window: int = 5) -> float | None: | |
| """Calculate form from points before the target gameweek.""" | |
| if not points: | |
| return None | |
| return float(pd.Series(points[-window:], dtype="float64").mean()) | |
| def build_training_frame( | |
| players: pd.DataFrame, | |
| histories: dict[int, list[dict[str, Any]]], | |
| fixtures: pd.DataFrame, | |
| ) -> pd.DataFrame: | |
| """Build rows where every feature was available before the target game.""" | |
| rows: list[dict[str, Any]] = [] | |
| names = players.set_index("id")["web_name"].to_dict() | |
| for player_id, history in histories.items(): | |
| prior_points: list[float] = [] | |
| for game in sorted(history, key=lambda item: (item.get("round") or 0, item.get("fixture") or 0)): | |
| form = previous_form(prior_points) | |
| fdr = fixture_fdr(int(game["fixture"]), bool(game["was_home"]), fixtures) | |
| points = as_float(game.get("total_points")) | |
| gameweek = as_float(game.get("round")) | |
| if form is not None and fdr is not None and points is not None and gameweek is not None: | |
| rows.append({ | |
| "player_id": player_id, | |
| "web_name": names.get(player_id, str(player_id)), | |
| "gameweek": int(gameweek), | |
| "form": form, | |
| "next_fdr": float(fdr), | |
| "points": float(points), | |
| }) | |
| if points is not None: | |
| prior_points.append(points) | |
| columns = ["player_id", "web_name", "gameweek", "form", "next_fdr", "points"] | |
| return pd.DataFrame(rows, columns=columns).sort_values("gameweek").reset_index(drop=True) | |
| def build_current_frame( | |
| players: pd.DataFrame, | |
| histories: dict[int, list[dict[str, Any]]], | |
| fixtures: pd.DataFrame, | |
| ) -> pd.DataFrame: | |
| """Build current-gameweek features using each player's recent history.""" | |
| rows: list[dict[str, Any]] = [] | |
| for player in players.itertuples(index=False): | |
| player_id = int(str(player.id)) | |
| team_id = int(str(player.team)) | |
| history = sorted(histories.get(player_id, []), key=lambda item: item.get("fixture") or 0) | |
| values = pd.to_numeric( | |
| pd.Series([game.get("total_points") for game in history]), errors="coerce" | |
| ).dropna().tolist() | |
| form = previous_form([float(value) for value in values]) | |
| fdr = next_fixture_fdr(team_id, fixtures) | |
| if form is not None and fdr is not None: | |
| rows.append({ | |
| "id": player_id, | |
| "web_name": player.web_name, | |
| "form": form, | |
| "next_fdr": float(fdr), | |
| }) | |
| return pd.DataFrame(rows, columns=["id", "web_name", "form", "next_fdr"]) | |
| def cache_histories(players: pd.DataFrame, cache_dir: Path) -> dict[int, list[dict[str, Any]]]: | |
| """Fetch player histories, caching each response for reproducible retraining.""" | |
| cache_dir.mkdir(parents=True, exist_ok=True) | |
| histories: dict[int, list[dict[str, Any]]] = {} | |
| player_ids = players["id"].astype(int).tolist() | |
| total = len(player_ids) | |
| for index, player_id in enumerate(player_ids, start=1): | |
| cache_file = cache_dir / f"{player_id}.json" | |
| if cache_file.exists(): | |
| records = pd.read_json(cache_file).to_dict("records") | |
| histories[player_id] = [{str(key): value for key, value in record.items()} for record in records] | |
| continue | |
| history = fetch_player_history(player_id) | |
| cache_file.write_text(pd.DataFrame(history).to_json(orient="records"), encoding="utf-8") | |
| histories[player_id] = history | |
| if index == 1 or index % 25 == 0 or index == total: | |
| print(f"Collected history for {index}/{total} players...") | |
| return histories |