from typing import Dict, List import requests import pandas as pd import json import os import re from django.conf import settings import logging logger = logging.getLogger("fpl_data") # Get the logger for the current module # Pandas options to avoid downcasting warnings pd.set_option("future.no_silent_downcasting", True) POSITIONS = {1: "GK", 2: "DEF", 3: "MID", 4: "FWD"} FPL_API_URL = 'https://fantasy.premierleague.com/api/bootstrap-static/' FPL_FIXTURE_API_URL = 'https://fantasy.premierleague.com/api/fixtures/' TEAMS_STORAGE_PATH = settings.FPL_TEAMS_FILE PLAYERS_STORAGE_PATH = settings.FPL_PLAYERS_FILE GAMEWEEKS_STORAGE_PATH = settings.FPL_GAMEWEEKS_FILE FIXTURES_STORAGE_PATH = settings.FPL_FIXTIRES_FILE INFO_FILE_PATH = settings.INFO_FILE UPCOMING_TEAM_FIXTIRES_DIFFICULT = settings.FPL_UPCOMING_TEAM_FIXTIRES_DIFFICULTY_FILE MODEL_PREDICTION_FOLDER = settings.MODEL_PREDICTION_FOLDER DREAM_TEAM_FOLDER = settings.DREAM_TEAM_FOLDER FASTAPI_SERVICE_URL = "http://127.0.0.1:8002" info = None def load_info(): try: with open(INFO_FILE_PATH, 'r') as f: return json.load(f) except FileNotFoundError: logger.warning("Info file not found. Returning empty dictionary.") return {} def save_info(info): try: with open(INFO_FILE_PATH, 'w') as f: json.dump(info, f, indent=4) except IOError as e: logger.error(f"Error saving info: {e}") print(f"Error saving info: {e}") def fetch_fpl_data(): """ Fetches data from the FPL API. Handles errors and logs them. """ global info info = load_info() try: logger.info("Fetching data from FPL API...") # Exclude columns that are not needed for the analysis exlude_player_cols = [ "news_added", "squad_number", "region", "now_cost_rank", "now_cost_rank_type", "form_rank", "form_rank_type", "points_per_game_rank", "points_per_game_rank_type", "selected_rank", "selected_rank_type", "influence_rank", "influence_rank_type", "creativity_rank", "creativity_rank_type", "threat_rank", "threat_rank_type", "ict_index_rank", "ict_index_rank_type", "corners_and_indirect_freekicks_text", "direct_freekicks_text", "penalties_text", "chance_of_playing_next_round", ] exlude_team_cols = [ "draw", "form", "loss", "played", "points", "position", "team_division", "unavailable", "win", "pulse_id", ] exlude_events_cols = [ "deadline_time_game_offset", "release_time", "cup_leagues_created", "h2h_ko_matches_created", "released", "can_enter", "can_manage", "ranked_count", "overrides", ] # Fetching data from the FPL API info["updating_data"] = True response = requests.get(FPL_API_URL) response.raise_for_status() # Raise an exception for bad status codes data = response.json() # Extracting teams and players data # get the teams, players, gameweeks, and fixtures data from the API response and Current gameweek teams = pd.json_normalize(data["teams"]) players = pd.json_normalize(data["elements"]) gameweeks = pd.DataFrame(data["events"]) gameweeks.fillna(-1.0, inplace=True) fixtures = fetch_fixtures_data() curr_gw = int( (gameweeks["finished"] == True).sum() + 1) logger.info(f"Current GW : {curr_gw}") info["gw"] = curr_gw # Model Data Update logger.info("Updating Model Data.") _ = requests.get(f"{FASTAPI_SERVICE_URL}/update") logger.info("Model Data Updated Successfully ") logger.info(f"Getting Model Predictions for Gw : {curr_gw}.") # Model retrain if ((curr_gw - 1) % 3 == 0 and (curr_gw - 1) != info["retrain_gw"] ): logger.info("Starting Model Retrain") _ = requests.get(f"{FASTAPI_SERVICE_URL}/retrain/{curr_gw - 1}") info["retrain_gw"] = curr_gw - 1 logger.info(f"Model Retrain Complete.") save_info(info) if curr_gw <39: response = requests.post(f"{FASTAPI_SERVICE_URL}/predict",data=json.dumps({"gw":curr_gw})) data = response.json() try: result = pd.json_normalize(data) result.to_csv(f"{MODEL_PREDICTION_FOLDER}gw{curr_gw}.csv",index=False) logger.info("Prediction Fetched Successfully.") except Exception as e: logger.error("Error decoding JSON response from FastAPI service: %s", e) # DATA PROCESSING players["now_cost"] = players["now_cost"] / 10 players['element_type']= players["element_type"].map( lambda elem: POSITIONS[elem] ) players["chance_of_playing_this_round"] = players.apply( clean_playing_chance_col, axis=1 ) players = players.apply(clean_setpieces_ranking,axis=1) # TODO: Current Game Week Will be Used in the Next Phase # gameweeks_rem = pd.to_datetime(gameweeks['deadline_time'], format="%Y-%m-%dT%H:%M:%SZ") # # Get the current datetime # now = datetime.now() # # Filter the DataFrame to keep only rows where 'deadline_time' is less than now # gameweeks_rem = gameweeks[gameweeks_rem > now] # current_game_week = int( gameweeks_rem.iloc[0]['id']) # print(current_game_week) # Each team upcoming matches difficulty upcoimg_5_gw_difficulty = calculate_teams_upcoming_matches_avg_difficulty(teams,fixtures,curr_gw) leaderboard = players.loc[:, ~players.columns.isin(exlude_player_cols)] # Adding expected Points For Testing Purpose xp_df = pd.read_csv(f"{MODEL_PREDICTION_FOLDER}gw{curr_gw}.csv").rename(columns={"xP":"XP"}) leaderboard = pd.merge(left=leaderboard,right=xp_df,on="id",how="left") leaderboard["XP"]=leaderboard["XP"].fillna(0) # Converting the DataFrame to a list of dictionaries leaderboard = leaderboard.sort_values(["id"], ascending=True).to_dict( orient="records" ) teams = teams.loc[:, ~teams.columns.isin(exlude_team_cols)].to_dict( orient="records" ) gameweeks = gameweeks.loc[ :, ~gameweeks.columns.isin(exlude_events_cols) ].to_dict(orient="records") fixtures = fixtures.to_dict( orient="records" ) upcoimg_5_gw_difficulty = upcoimg_5_gw_difficulty.to_dict( orient="records" ) logger.info("Fetched players ,gameweeks, fixtures ,and teams from FPL API Successfully.") return leaderboard, teams, gameweeks, fixtures, upcoimg_5_gw_difficulty except json.JSONDecodeError as e: print(f"Error decoding JSON: {e}") logger.error(f"Error decoding JSON: {e}") return None except requests.exceptions.HTTPError as e: print(f"HTTP error occurred: {e}") logger.error(f"HTTP error occurred: {e}") return None except requests.exceptions.RequestException as e: print(f"Error fetching FPL data: {e}") logger.error(f"Error decoding JSON: {e}") return None except Exception as e: print(f"An unexpected error occurred: {e}") logger.error(f"An unexpected error occurred: {e}") return None def save_fpl_data(data): """ Saves the FPL data to a JSON file. Handles errors and logs them. """ global info if data is None: logger.warning("No data to save.") return try: # Ensure the directory exists os.makedirs(os.path.dirname(TEAMS_STORAGE_PATH), exist_ok=True) os.makedirs(os.path.dirname(PLAYERS_STORAGE_PATH), exist_ok=True) os.makedirs(os.path.dirname(GAMEWEEKS_STORAGE_PATH), exist_ok=True) os.makedirs(os.path.dirname(FIXTURES_STORAGE_PATH), exist_ok=True) os.makedirs(os.path.dirname(UPCOMING_TEAM_FIXTIRES_DIFFICULT), exist_ok=True) # Saving the teams data to a JSON file with open(UPCOMING_TEAM_FIXTIRES_DIFFICULT, "w") as f: json.dump(data[4], f, indent=4) with open(FIXTURES_STORAGE_PATH, "w") as f: json.dump(data[3], f, indent=4) with open(GAMEWEEKS_STORAGE_PATH, "w") as f: json.dump(data[2], f, indent=4) with open(TEAMS_STORAGE_PATH, "w") as f: json.dump(data[1], f, indent=4) # Saving the players to a JSON file with open(PLAYERS_STORAGE_PATH, "w") as f: json.dump(data[0], f, indent=4) info["updating_data"] = False save_info(info) if info['gw'] < 39: save_dream_teams(info['gw']) logger.info("FPL data saved successfully.") except (IOError, OSError) as e: logger.error(f"Error saving FPL data {e}") except TypeError as e: logger.error(f"Type error while saving FPL data: {e}") except Exception as e: logger.error(f"An unexpected error occurred: {e}") def load_teams_data(): """ Loads the teams data from the JSON file. Handles errors, logs them, and returns None on failure. """ try: with open(TEAMS_STORAGE_PATH, "r") as f: return json.load(f) except FileNotFoundError: logger.warning( f"FPL teams file not found at {TEAMS_STORAGE_PATH}. Returning None." ) return None except json.JSONDecodeError: logger.error(f"Error decoding JSON from {TEAMS_STORAGE_PATH}. Returning None.") return None except (IOError, OSError) as e: logger.error(f"Error reading FPL teams data from {TEAMS_STORAGE_PATH}: {e}") return None except Exception as e: logger.error(f"An unexpected error occurred: {e}") return None def load_gameweeks_data(): """ Loads the gameweeks data from the JSON file. Handles errors, logs them, and returns None on failure. """ try: with open(GAMEWEEKS_STORAGE_PATH, "r") as f: return json.load(f) except FileNotFoundError: logger.warning( f"FPL teams file not found at {GAMEWEEKS_STORAGE_PATH}. Returning None." ) return None except json.JSONDecodeError: logger.error( f"Error decoding JSON from {GAMEWEEKS_STORAGE_PATH}. Returning None." ) return None except (IOError, OSError) as e: logger.error(f"Error reading FPL teams data from {GAMEWEEKS_STORAGE_PATH}: {e}") return None except Exception as e: logger.error(f"An unexpected error occurred: {e}") return None def load_fixtures_data(): """ Loads the gameweeks data from the JSON file. Handles errors, logs them, and returns None on failure. """ try: with open(FIXTURES_STORAGE_PATH, "r") as f: return json.load(f) except FileNotFoundError: logger.warning( f"FPL teams file not found at {FIXTURES_STORAGE_PATH}. Returning None." ) return None except json.JSONDecodeError: logger.error( f"Error decoding JSON from {FIXTURES_STORAGE_PATH}. Returning None." ) return None except (IOError, OSError) as e: logger.error(f"Error reading FPL teams data from {FIXTURES_STORAGE_PATH}: {e}") return None except Exception as e: logger.error(f"An unexpected error occurred: {e}") return None def load_players_data(): """ Loads the players data from the JSON file. Handles errors, logs them, and returns None on failure. """ try: with open(PLAYERS_STORAGE_PATH, "r") as f: return json.load(f) except FileNotFoundError: logger.warning( f"FPL players file not found at {PLAYERS_STORAGE_PATH}. Returning None." ) return None except json.JSONDecodeError: logger.error( f"Error decoding JSON from {PLAYERS_STORAGE_PATH}. Returning None." ) return None except (IOError, OSError) as e: logger.error(f"Error reading FPL players data from {PLAYERS_STORAGE_PATH}: {e}") return None except Exception as e: logger.error(f"An unexpected error occurred: {e}") return None def load_team_upcoming_fix_difficulty_data(): """ Loads the players data from the JSON file. Handles errors, logs them, and returns None on failure. """ try: with open(UPCOMING_TEAM_FIXTIRES_DIFFICULT, "r") as f: return json.load(f) except FileNotFoundError: logger.warning( f"FPL players file not found at {UPCOMING_TEAM_FIXTIRES_DIFFICULT}. Returning None." ) return None except json.JSONDecodeError: logger.error( f"Error decoding JSON from {UPCOMING_TEAM_FIXTIRES_DIFFICULT}. Returning None." ) return None except (IOError, OSError) as e: logger.error(f"Error reading FPL players data from {UPCOMING_TEAM_FIXTIRES_DIFFICULT}: {e}") return None except Exception as e: logger.error(f"An unexpected error occurred: {e}") return None def fetch_fixtures_data(): response = requests.get(FPL_FIXTURE_API_URL) data = response.json() fixtures = pd.json_normalize(data) fixtures.fillna(-1,inplace=True) return fixtures[["event","team_h","team_h_difficulty",'team_h_score',"team_a","team_a_difficulty","team_a_score","finished"]] def calculate_teams_upcoming_matches_avg_difficulty(teams:pd.DataFrame, fixtures:pd.DataFrame,current_gw : int) -> pd.DataFrame: team_ids = teams["id"].to_list() teams_upcoming_matches_avg_difficulty = pd.DataFrame() teams_upcoming_matches_avg_difficulty["id"] = team_ids gw_diff_per_team = add_gameweek_difficclty(current_gw,team_ids,fixtures) # Adding Upcoming Gameweeks Difficulties for k ,v in gw_diff_per_team.items(): col_name = f"GW_{k}" teams_upcoming_matches_avg_difficulty[col_name] = v # Melting The Gw Columns into one by Calculating the average # The following will ignore column values of zero ( Blank Game weeks) id_vars = ['id'] value_vars = [col for col in teams_upcoming_matches_avg_difficulty.columns if col.startswith('GW_')] melted_df = teams_upcoming_matches_avg_difficulty.melt( id_vars=id_vars, value_vars=value_vars, var_name='GameWeek', value_name='Difficulty' ) # Remove rows where Difficulty is 0 melted_df = melted_df[melted_df['Difficulty'] != 0] # Calculate the average difficulty for each team average_difficulty = melted_df.groupby('id')['Difficulty'].mean() # Print the result return pd.merge(teams_upcoming_matches_avg_difficulty, average_difficulty, on='id', how='left') def add_gameweek_difficclty (curr_gw:int,teams:List[int],fixtures:pd.DataFrame)->Dict[int, List[int]]: difficulties = {} final = curr_gw + 4 if curr_gw + 4 <= 38 else 38 for index,week in enumerate(range(curr_gw,final + 1)): diff_by_team = [] for team in teams : current_gw_fixtures = fixtures[fixtures["event"] == week] team_row = current_gw_fixtures[current_gw_fixtures['team_h'] == team] if not team_row.empty: diff_by_team.append(int (team_row.iloc[0]['team_h_difficulty'])) continue team_row = current_gw_fixtures[current_gw_fixtures['team_a'] == team] if not team_row.empty: diff_by_team.append(int (team_row.iloc[0]['team_a_difficulty'])) continue diff_by_team.append(0) difficulties[index + 1] = diff_by_team return difficulties def clean_playing_chance_col(record): if pd.isna(record["chance_of_playing_this_round"]): if record["status"] == "a": return 100 elif record["status"] == "d": percentage_match = re.search(r"(\d+)%", record["news"]) if percentage_match.group(1): number = percentage_match.group(1) return float(number) else: return 0 else: return record["chance_of_playing_this_round"] def clean_setpieces_ranking(record): if pd.isna(record["penalties_order"]): record["penalties_order"] = 0 if pd.isna(record["direct_freekicks_order"]): record["direct_freekicks_order"] = 0 if pd.isna(record["corners_and_indirect_freekicks_order"]): record["corners_and_indirect_freekicks_order"] = 0 return record def save_dream_teams(curr_gw: int): from squad.services import load_ai_dream_team ai_dream_team = load_ai_dream_team(curr_gw) save_ai_dream_team(ai_dream_team, curr_gw) if curr_gw > 1: save_actual_dream_team(curr_gw - 1) def save_actual_dream_team( gw: int, filename="actual_dream_team.json"): """ Save actual dream team only once per gameweek. If the file has fewer entries than the current gameweek, add the new dream team. Otherwise, do nothing. """ from squad.services import load_dream_team path = f"{DREAM_TEAM_FOLDER}{filename}" try: # Load existing data if file exists if os.path.exists(path): with open(path, "r") as f: data = json.load(f) else: data = {} # Only add if current gw is greater than number of entries if len(data) < gw: data[str(gw)] = load_dream_team(gw) with open(path, "w") as f: json.dump(data, f, indent=4) logger.info(f"Saved actual dream team for GW {gw}") else: logger.info(f"Actual dream team for GW {gw} already saved. No action taken.") except Exception as e: logger.error(f"Error saving actual dream team: {e}") # Usage example: # save_actual_dream_team(actual_dream_team, curr_gw) def save_ai_dream_team(ai_dream_team: dict, gw: int, filename="ai_dream_team.json"): """ Save ai_dream_team for a given gameweek. If gw exists, replace its content; if not, add new gw content. """ path = f"{DREAM_TEAM_FOLDER}{filename}" try: # Load existing data if file exists if os.path.exists(path): with open(path, "r") as f: data = json.load(f) else: data = {} # Update or add the gameweek content data[str(gw)] = ai_dream_team # Save back to file with open(path, "w") as f: json.dump(data, f, indent=4) logger.info(f"Saved AI dream team for GW {gw} ") except Exception as e: logger.error(f"Error saving AI dream team: {e}") if __name__ == "__main__": """ This block is only executed when this script is run directly (e.g., for testing). It's good practice to include such a block for testing purposes. """ data = fetch_fpl_data() if data: save_fpl_data(data) else: print("Failed to fetch data.") # Commented Out Code # players['selected_by_percent'] = players['selected_by_percent'] + " %" # players['team'] = players['team'].map(lambda team: teams[team-1]['name']) # leaderboard = players[['web_name','now_cost','form','selected_by_percent','element_type',"team","transfers_in_event","transfers_out_event",'total_points']].sort_values(by='total_points', ascending=False) # leaderboard = leaderboard.rename(columns={'web_name': 'name', 'now_cost': 'price', 'form': 'form','selected_by_percent': 'selection','element_type':'position','team':'team','transfers_in_event': 'Transfers In','transfers_out_event': 'Transfers Out','total_points': 'Total Points'})