import hashlib import itertools as _itertools import json import os import threading import time from typing import Any, Dict, List, Optional import numpy as np import pandas as pd import requests from fastapi import FastAPI, HTTPException, Depends from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from sqlalchemy.orm import Session from sqlalchemy.orm.attributes import flag_modified import engine from baseline_join import join_baselines from auth import router as auth_router from fpl_api import get_fpl_team_data from solver import run_milp_model from solver_engine import prep_solver_data from database import get_db, User, SessionLocal, GlobalConfig # --- PYDANTIC MODELS FOR REACT PAYLOAD --- class PlayerData(BaseModel): id: int name: str pos: str team: str now_cost: float sell_price: Optional[float] = None evs: Dict[str, float] # JSON keys are strings; horizon GW keyed class SolveRequest(BaseModel): team_id: int horizon_gws: List[int] current_squad_ids: List[Any] market_players: List[PlayerData] in_the_bank: float free_transfers: int settings: dict = {} comprehensive_settings: dict = {} class ChipSolveRequest(BaseModel): team_id: int horizon_gws: List[int] current_squad_ids: List[Any] market_players: List[PlayerData] in_the_bank: float free_transfers: int settings: dict = {} comprehensive_settings: dict = {} # { "wc": [gw, ...], "fh": [gw, ...], "bb": [gw, ...], "tc": [gw, ...] } chip_gw_options: Dict[str, List[int]] = {} class SettingsPayload(BaseModel): team_id: int quick_settings: Dict[str, Any] advanced_settings: Dict[str, Any] app = FastAPI(title="Luigi's Mansion FPL API") @app.get("/") def read_root(): return { "status": "success", "message": "Luigi's Mansion FPL Solver API is live and running!", } app.include_router(auth_router) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) ADMIN_PASSWORD = "Monkeyrocks11$$" TEAMS_DICT = { "Arsenal": 1, "Aston Villa": 2, "AFC Bournemouth": 3, "Brentford": 4, "Brighton and Hove Albion": 5, "Chelsea": 6, "Coventry City": 7, "Crystal Palace": 8, "Everton": 9, "Fulham": 10, "Hull City": 11, "Ipswich Town": 12, "Leeds United": 13, "Liverpool": 14, "Manchester City": 15, "Manchester United": 16, "Newcastle United": 17, "Nottingham Forest": 18, "Tottenham Hotspur": 19, "Sunderland": 20, } TEAMS_DICT_REVERSE = {v: k for k, v in TEAMS_DICT.items()} POS_MAP = {1: "G", 2: "D", 3: "M", 4: "F"} POINTS_CONFIG = { "goal": {1: 10, 2: 6, 3: 5, 4: 4}, "assist": 3, "clean_sheet": {1: 4, 2: 4, 3: 1, 4: 0}, "saves_per_3": 1, "penalty_points_per_position": {2: 0.9, 3: 0.7, 4: 0.5}, } class AppData: finalized_df = None match_df = None output_df = None # All JSON States player_penalty_shares = {} admin_xmins_overrides = {} admin_mins_probs = {} # Each gameweek's squad contributions from the last full engine run, so a # single-player recalculation closes its shares against the full squad. share_context = None admin_baseline_overrides = {} player_status_overrides = {} availability_multipliers = {} # {FPL code (str): share} - hand-pinned penalty takers. penalty_overrides = {} admin_fixture_overrides = {} decay_rates = { "default": 0.99, "suspended": 0.99, "injured_decay": 0.99, "rotational_risk": 0.95, } ramp_up_rates = { "default": 3, "injured": 9, "suspended": 3, "starter": 0, "rotational_risk": 2, } MINS_THRESHOLD = 30 RAMP_UP_PERIOD = 3 app_data = AppData() # A small share of a team's penalties is left to everyone in proportion to # open-play xG, so a side with no recorded taker still hands its penalties to # its likeliest shooters rather than losing them. PENALTY_NPXG_FLOOR = 0.05 def _penalty_strength(final_df, overrides): """Per-player weight for sharing out a team's penalties. The taker score from build_penalty_takers.py (penalties taken over penalties his team won while he was on the pitch, recency-weighted), joined on FPL `code` - the Opta id, stable across seasons, so it cannot drift onto another player the way id-keyed shares did - plus a small open-play xG floor. `overrides` ({code: share}) pins a known taker. Everyone else on his team then shares whatever is left, in proportion to their data-driven weight. engine.closed_shares turns these weights into shares of each match's penalties among the players actually expected on the pitch. """ scores = {} if os.path.exists("penalty_takers.csv"): takers = pd.read_csv("penalty_takers.csv") scores = dict(zip(takers["code"].astype(int), takers["taker_score"].astype(float))) code = final_df["code"].astype(int) npxg = final_df.get("baseline_npxG_p90", final_df.get("baseline_xG_p90", 0.0)) strength = code.map(scores).fillna(0.0) + PENALTY_NPXG_FLOOR * pd.Series(npxg, index=final_df.index).astype(float) pinned = code.map(lambda c: overrides.get(str(c))).astype(float) for team, idx in final_df.groupby("team").groups.items(): team_pins = pinned.loc[idx].dropna() if team_pins.empty: continue rest = [i for i in idx if i not in team_pins.index] remaining = max(1.0 - float(team_pins.sum()), 0.0) rest_total = float(strength.loc[rest].sum()) if rest_total > 0: strength.loc[rest] = strength.loc[rest] * remaining / rest_total strength.loc[team_pins.index] = team_pins return strength def load_db_int_keys(db_key, default): db = SessionLocal() try: config = db.query(GlobalConfig).filter(GlobalConfig.key == db_key).first() if config and config.value: return {int(k): v for k, v in config.value.items()} finally: db.close() return default def load_db_string_keys(db_key, default): db = SessionLocal() try: config = db.query(GlobalConfig).filter(GlobalConfig.key == db_key).first() if config and config.value: return config.value finally: db.close() return default def save_config_to_db(db_key, value): db = SessionLocal() try: config = db.query(GlobalConfig).filter(GlobalConfig.key == db_key).first() if config: config.value = value flag_modified(config, "value") else: config = GlobalConfig(key=db_key, value=value) db.add(config) db.commit() finally: db.close() def load_fpl_data(): print("Fetching FPL API data...") r = requests.get( "https://fantasy.premierleague.com/api/bootstrap-static/", timeout=10 ).json() players = pd.DataFrame(r["elements"]) players["name"] = players["first_name"] + " " + players["second_name"] players = players[ [ "id", "code", "name", "web_name", "element_type", "now_cost", "team", "chance_of_playing_this_round", "news", "photo", ] ] players["now_cost"] = players["now_cost"] / 10 print("Loading baseline stats...") final_df = join_baselines(players) final_df.fillna(0, inplace=True) final_df["Avg_BPS"] = 0.0 final_df.loc[final_df["element_type"] == 1, "Avg_BPS"] = final_df[ "baseline_gk_bps_p90" ].astype(float) final_df.loc[final_df["element_type"] == 2, "Avg_BPS"] = ( final_df["baseline_Neutral_BPS_p90"] + final_df["baseline_Def_BPS_p90"] ) final_df.loc[final_df["element_type"] == 3, "Avg_BPS"] = ( final_df["baseline_Neutral_BPS_p90"] + final_df["baseline_Mid_BPS_p90"] ) final_df.loc[final_df["element_type"] == 4, "Avg_BPS"] = ( final_df["baseline_Neutral_BPS_p90"] + final_df["baseline_Fwd_BPS_p90"] ) print("Loading team and match projections...") team_baselines = pd.read_excel("team_totals.xlsx", sheet_name="Sheet2") team_baselines["Teams"] = team_baselines["Teams"].replace(TEAMS_DICT) for stat in ["xG", "xA", "CBIT", "CBITR", "YC", "RC"]: final_df[f"Team_{stat}"] = final_df["team"].map( team_baselines.set_index("Teams")[stat].to_dict() ) final_df["Team"] = final_df["team"].map(TEAMS_DICT_REVERSE) final_df["xG_share"] = final_df["baseline_xG_p90"] / final_df["Team_xG"].replace( 0, np.nan ) final_df["xA_share"] = final_df["baseline_xA_p90"] / final_df["Team_xA"].replace( 0, np.nan ) final_df["xCBIT_share"] = final_df["baseline_CBIT_p90"] / final_df[ "Team_CBIT" ].replace(0, np.nan) final_df["xCBITR_share"] = final_df["baseline_CBITR_p90"] / final_df[ "Team_CBITR" ].replace(0, np.nan) final_df["YC_share"] = final_df["baseline_yc_p90"] / final_df["Team_YC"].replace( 0, np.nan ) final_df["RC_share"] = final_df["baseline_rc_p90"] / final_df["Team_RC"].replace( 0, np.nan ) final_df.fillna(0, inplace=True) final_df["pen_strength"] = _penalty_strength( final_df, load_db_string_keys("penalty_overrides", {}) ) match_df = pd.read_csv("ewmapois_model.csv").rename(columns=lambda x: x.strip()) match_df["home_team_num"] = match_df["home_team"].map(TEAMS_DICT) match_df["away_team_num"] = match_df["away_team"].map(TEAMS_DICT) app_data.finalized_df = final_df app_data.match_df = match_df # --- LOAD ALL DB OVERRIDES --- # Penalty takers come from the match data now (build_penalty_takers.py), # keyed on FPL `code`. The old "penalty_shares" config was keyed on FPL # element id, which FPL renumbers every season, so it is no longer read. app_data.player_penalty_shares = {} app_data.penalty_overrides = load_db_string_keys("penalty_overrides", {}) if os.path.exists("penalty_model.json"): with open("penalty_model.json", "r", encoding="utf-8") as f: POINTS_CONFIG["penalty_model"] = json.load(f) # Positional bonus base rates from FPL's own history (build_bonus_rates.py). if os.path.exists("bonus_model.json"): with open("bonus_model.json", "r", encoding="utf-8") as f: POINTS_CONFIG["bonus_model"] = json.load(f) raw_xmins = load_db_int_keys("admin_xmins", {}) processed_overrides = {} for pid, gws in raw_xmins.items(): processed_gws = {} for gw_key, val in gws.items(): if str(gw_key).isdigit(): processed_gws[int(gw_key)] = val else: processed_gws[str(gw_key)] = val processed_overrides[int(pid)] = processed_gws app_data.admin_xmins_overrides = processed_overrides # Solio's minutes curve: {player: {gw: [p_start, p_60]}}. engine.py scores # appearance, clean sheets and minutes BPS against these instead of # thresholding the mean at 60. Absent players fall back to the threshold. app_data.admin_mins_probs = { int(pid): {int(gw): val for gw, val in gws.items() if str(gw).isdigit()} for pid, gws in load_db_int_keys("admin_mins_probs", {}).items() } app_data.admin_baseline_overrides = load_db_int_keys("admin_baselines", {}) app_data.player_status_overrides = load_db_int_keys("player_status", {}) app_data.availability_multipliers = load_db_int_keys("availability", {}) app_data.admin_fixture_overrides = load_db_string_keys("admin_fixtures", {}) # --- THE FALLBACK LOGIC --- # Apply baseline JSON overrides on top of the CSV data. If not in JSON, it naturally keeps the CSV data. for pid, overrides in app_data.admin_baseline_overrides.items(): if pid in app_data.finalized_df["id"].values: if "baseline_xMins" in overrides: app_data.finalized_df.loc[ app_data.finalized_df["id"] == pid, "baseline_xMins" ] = overrides["baseline_xMins"] # Admin baseline edits scale Solio's curve; without this they only moved the # baseline column, since every gameweek in the horizon has a Solio figure. effective_xmins = {pid: dict(gws) for pid, gws in app_data.admin_xmins_overrides.items()} effective_probs = {pid: dict(gws) for pid, gws in app_data.admin_mins_probs.items()} for pid, overrides in app_data.admin_baseline_overrides.items(): if "baseline_xMins" in overrides: pid = int(pid) effective_xmins[pid], effective_probs[pid] = _scaled_to_baseline( pid, overrides["baseline_xMins"], effective_xmins.get(pid, {}), effective_probs.get(pid, {})) print("Running initial FPL point engine...") app_data.output_df = engine.calculate_all_points( player_df_base=app_data.finalized_df, match_df=app_data.match_df, player_penalty_shares=app_data.player_penalty_shares, MINS_SCALING_BONUS=0.0, pos_map=POS_MAP, teams_dict_1=TEAMS_DICT_REVERSE, teams_dict=TEAMS_DICT, points_config=POINTS_CONFIG, effective_xmins_overrides=effective_xmins, MINS_THRESHOLD=app_data.MINS_THRESHOLD, RAMP_UP_PERIOD=app_data.RAMP_UP_PERIOD, decay_rates=app_data.decay_rates, ramp_up_rates=app_data.ramp_up_rates, user_player_status_overrides=app_data.player_status_overrides, team_skepticism={}, effective_availability_multipliers=app_data.availability_multipliers, effective_mins_probs=effective_probs, ) app_data.share_context = app_data.output_df.attrs.get("share_context") # Inject baseline_xMins into the output so the frontend can display it app_data.output_df["baseline_xMins"] = app_data.output_df["ID"].map( app_data.finalized_df.set_index("id")["baseline_xMins"] ) # --- BULLETPROOF ADVANCED STATS INJECTION --- try: finalized_idx = app_data.finalized_df.set_index("id") # 1. Add photo and Price so Transfer Market can display images and cost if "photo" in finalized_idx.columns: app_data.output_df["photo"] = app_data.output_df["ID"].map( finalized_idx["photo"] ) if "now_cost" in finalized_idx.columns: app_data.output_df["Price"] = app_data.output_df["ID"].map( finalized_idx["now_cost"] ) # 2. Safely map xG and xA app_data.output_df["xG"] = app_data.output_df["ID"].map( lambda pid: round( (finalized_idx.loc[pid, "baseline_xG_p90"] / 90) * finalized_idx.loc[pid, "baseline_xMins"], 2, ) if pid in finalized_idx.index and "baseline_xG_p90" in finalized_idx.columns else 0 ) app_data.output_df["xA"] = app_data.output_df["ID"].map( lambda pid: round( (finalized_idx.loc[pid, "baseline_xA_p90"] / 90) * finalized_idx.loc[pid, "baseline_xMins"], 2, ) if pid in finalized_idx.index and "baseline_xA_p90" in finalized_idx.columns else 0 ) # 3. Safely get CS% unique_gws = sorted(app_data.match_df["GW"].unique()) for gw in unique_gws: if f"{gw}_xG" in app_data.output_df.columns: # Format CS% for the GW app_data.output_df[f"{gw}_CS_Pct"] = ( app_data.output_df[f"{gw}_CS"] * 100 ).apply(lambda x: f"{x:.0f}%") # Calculate HIT% for the GW using the stored CBIT / CBITR def calc_hit_gw(row): pos = row["Pos"] if pos == "D": cbit = row.get(f"{gw}_cbit", 0) prob = engine.neg_binom_probability_at_least( cbit, 10, dispersion=3.2 ) elif pos == "M": cbitr = row.get(f"{gw}_cbitr", 0) prob = engine.neg_binom_probability_at_least( cbitr, 12, dispersion=2.8 ) elif pos == "F": cbitr = row.get(f"{gw}_cbitr", 0) prob = engine.neg_binom_probability_at_least( cbitr, 12, dispersion=1.7 ) else: return "-" return f"{prob * 100:.0f}%" app_data.output_df[f"{gw}_DefconHit"] = app_data.output_df.apply( calc_hit_gw, axis=1 ) except Exception as e: print(f"WARNING: Could not inject advanced stats. Reason: {e}") # How often the running app checks Supabase for new xMins. run_auto.py writes # them every few hours, but load_fpl_data only ran at startup, so a write sat # unseen until something happened to restart the Space. XMINS_POLL_SECONDS = int(os.environ.get("XMINS_POLL_SECONDS", "900")) _WATCHED_KEYS = ("admin_xmins", "admin_mins_probs") def _xmins_signature(): db = SessionLocal() try: rows = db.query(GlobalConfig).filter(GlobalConfig.key.in_(_WATCHED_KEYS)).all() blob = json.dumps({r.key: r.value for r in rows}, sort_keys=True, default=str) finally: db.close() return hashlib.sha256(blob.encode()).hexdigest() def _watch_xmins(): """Recalculate whenever the stored xMins change. Never raises.""" try: seen = _xmins_signature() except Exception as e: print(f"WARNING: xMins watcher could not read Supabase: {e}") seen = None while True: time.sleep(XMINS_POLL_SECONDS) try: current = _xmins_signature() if current != seen: print("xMins changed in Supabase - recalculating projections...") load_fpl_data() seen = current except Exception as e: print(f"WARNING: xMins reload failed, keeping current projections: {e}") @app.on_event("startup") def startup_event(): load_fpl_data() threading.Thread(target=_watch_xmins, daemon=True, name="xmins-watch").start() @app.get("/api/projections") def get_projections(): if app_data.output_df is None: raise HTTPException(status_code=503, detail="Loading") clean_df = app_data.output_df.where(pd.notnull(app_data.output_df), None) return clean_df.to_dict(orient="records") class UpdateRequest(BaseModel): player_id: int baseline_edit: Optional[float] = None gw_edits: Dict[str, float] = {} is_admin: bool = False admin_password: Optional[str] = None def _scaled_to_baseline(pid, baseline, gw_xmins, gw_probs): """Rescale a player's per-gameweek minutes (Solio's curve) to a hand-set baseline. Every gameweek in the horizon carries a Solio figure, and those outrank the baseline in the engine, so a baseline edit used to change nothing: on 2026-11-10 typing 90 for Gomez left him on 71 every week. The edit means "his minutes when fit", so the curve's peak is set to it and the other weeks keep their size relative to the peak: Gomez typed at 90 runs 90, 89, 88... instead of 71, 70, 69..., and a week Solio has him injured for stays at 0. (Scaling against the model's own baseline was tried first and only got him to 75.) Solio's start and 60+ probabilities belong to the old curve, so the rescaled weeks drop them and the engine falls back to thresholding the new mean. """ if not gw_xmins: return gw_xmins, gw_probs anchor = max((v for v in gw_xmins.values() if isinstance(v, (int, float))), default=0) if anchor <= 0: return {gw: float(baseline) for gw in gw_xmins}, {} ratio = float(baseline) / anchor if abs(ratio - 1.0) < 1e-6: return gw_xmins, gw_probs scaled = {gw: min(float(v) * ratio, 90.0) for gw, v in gw_xmins.items()} probs = {gw: v for gw, v in gw_probs.items() if gw not in scaled} return scaled, probs @app.post("/api/player/update") def update_player(req: UpdateRequest): if req.is_admin: if req.admin_password != ADMIN_PASSWORD: raise HTTPException(status_code=401, detail="Invalid admin password") # Save Admin Edits directly to the JSON files if req.baseline_edit is not None: if req.player_id not in app_data.admin_baseline_overrides: app_data.admin_baseline_overrides[req.player_id] = {} app_data.admin_baseline_overrides[req.player_id]["baseline_xMins"] = ( req.baseline_edit ) save_config_to_db("admin_baselines", app_data.admin_baseline_overrides) for gw_str, xmins in req.gw_edits.items(): gw_key = int(gw_str) if str(gw_str).isdigit() else str(gw_str) if req.player_id not in app_data.admin_xmins_overrides: app_data.admin_xmins_overrides[req.player_id] = {} app_data.admin_xmins_overrides[req.player_id][gw_key] = xmins # A hand-set minutes figure supersedes Solio's curve for that # gameweek. Leaving the old p_60 in place would score the new # number against the old distribution - 90 minutes typed in, but # still only a 40% chance of reaching 60. Dropping it falls the # engine back to the threshold on what was actually typed. app_data.admin_mins_probs.get(req.player_id, {}).pop(gw_key, None) if req.gw_edits: save_config_to_db("admin_xmins", app_data.admin_xmins_overrides) save_config_to_db("admin_mins_probs", app_data.admin_mins_probs) player_df = app_data.finalized_df[ app_data.finalized_df["id"] == req.player_id ].copy() if player_df.empty: raise HTTPException(status_code=404, detail="Player not found") current_baseline = player_df.iloc[0]["baseline_xMins"] if req.baseline_edit is not None: player_df["baseline_xMins"] = req.baseline_edit current_baseline = req.baseline_edit gw_xmins = app_data.admin_xmins_overrides.get(req.player_id, {}).copy() gw_probs = app_data.admin_mins_probs.get(req.player_id, {}).copy() admin_baseline = app_data.admin_baseline_overrides.get(req.player_id, {}).get("baseline_xMins") target = req.baseline_edit if req.baseline_edit is not None else admin_baseline if target is not None: gw_xmins, gw_probs = _scaled_to_baseline(req.player_id, target, gw_xmins, gw_probs) effective_overrides = {req.player_id: gw_xmins} effective_probs = {req.player_id: gw_probs} for gw_str, val in req.gw_edits.items(): gw_key = int(gw_str) if str(gw_str).isdigit() else str(gw_str) effective_overrides[req.player_id][gw_key] = val effective_probs[req.player_id].pop(gw_key, None) # Recalculate using all the existing status/penalty configs updated_row_df = engine.calculate_all_points( player_df_base=player_df, match_df=app_data.match_df, player_penalty_shares=app_data.player_penalty_shares, MINS_SCALING_BONUS=0.0, pos_map=POS_MAP, teams_dict_1=TEAMS_DICT_REVERSE, teams_dict=TEAMS_DICT, points_config=POINTS_CONFIG, effective_xmins_overrides=effective_overrides, MINS_THRESHOLD=app_data.MINS_THRESHOLD, RAMP_UP_PERIOD=app_data.RAMP_UP_PERIOD, decay_rates=app_data.decay_rates, ramp_up_rates=app_data.ramp_up_rates, user_player_status_overrides=app_data.player_status_overrides, team_skepticism={}, effective_availability_multipliers=app_data.availability_multipliers, effective_mins_probs=effective_probs, share_context=app_data.share_context, ) updated_row_df["baseline_xMins"] = current_baseline if req.is_admin and app_data.output_df is not None: idx = app_data.output_df[app_data.output_df["ID"] == req.player_id].index if not idx.empty: row_idx = idx[0] for col in updated_row_df.columns: if col in app_data.output_df.columns: app_data.output_df.at[row_idx, col] = updated_row_df[col].values[0] return updated_row_df.iloc[0].to_dict() @app.get("/api/ratings") def get_ratings(): if os.path.exists("team_ratings_dual_speed.csv"): df = pd.read_csv("team_ratings_dual_speed.csv") df.rename(columns=lambda x: x.strip(), inplace=True) # Strip trailing spaces just in case! df["Team"] = df["Team"].str.strip() return df.to_dict(orient="records") return [] @app.get("/api/fixtures") def get_fixtures(): if app_data.match_df is not None: cols = [ "GW", "home_team", "away_team", "home_win_prob", "draw_prob", "away_win_prob", "expected_home_goals", "expected_away_goals", "home_clean_sheet_odds", "away_clean_sheet_odds", ] df = app_data.match_df[cols].copy() for col in [ "home_win_prob", "draw_prob", "away_win_prob", "expected_home_goals", "expected_away_goals", "home_clean_sheet_odds", "away_clean_sheet_odds", ]: df[col] = df[col].astype(float).round(3) return df.to_dict(orient="records") return [] @app.get("/api/accuracy/players") def get_accuracy_players(): import os import pandas as pd file_path = "points_check.xlsx" if os.path.exists(file_path): df = pd.read_excel(file_path) df.columns = df.columns.str.strip() df.fillna(0, inplace=True) return df.to_dict(orient="records") # Fallback just in case csv_path = "points_check.xlsx - Sheet1.csv" if os.path.exists(csv_path): df = pd.read_csv(csv_path) df.columns = df.columns.str.strip() df.fillna(0, inplace=True) return df.to_dict(orient="records") return [] @app.get("/api/accuracy/matches") def get_accuracy_matches(): import os import pandas as pd file_path = "projections_check.xlsx" if os.path.exists(file_path): df = pd.read_excel(file_path) df.columns = df.columns.str.strip() df.fillna(0, inplace=True) return df.to_dict(orient="records") # Fallback just in case csv_path = "projections_check.xlsx - Sheet1.csv" if os.path.exists(csv_path): df = pd.read_csv(csv_path) df.columns = df.columns.str.strip() df.fillna(0, inplace=True) return df.to_dict(orient="records") return [] @app.get("/api/manager/{team_id}") async def get_manager_team(team_id: int): try: # 1. Run the precise open-fpl-solver logic to get ITB, FTs, and Selling Prices fpl_data = get_fpl_team_data(team_id) team_data = [] # 2. Merge the official FPL data with your local Projection Data for pick in fpl_data["squad"]: pid = pick["id"] # Find the player in your master projections dataframe proj_row = app_data.output_df[app_data.output_df["ID"] == pid] if proj_row.empty: continue proj_dict = proj_row.iloc[0].to_dict() # --- CRITICAL FIX: TRUE SELLING PRICE --- # Overwrite the global "Cost Price" with the user's personal "Selling Price" proj_dict["Price"] = pick["selling_price"] proj_dict["sell_price"] = pick["selling_price"] # Add photos if needed base_row = app_data.finalized_df[app_data.finalized_df["id"] == pid] proj_dict["photo"] = ( base_row.iloc[0]["photo"] if not base_row.empty and "photo" in base_row.columns else "" ) team_data.append(proj_dict) # 3. Send the exact payload React is expecting! return { "in_the_bank": fpl_data["in_the_bank"], "free_transfers": fpl_data["free_transfers"], "picks": team_data, } except Exception as e: print(f"Error fetching team: {e}") raise HTTPException(status_code=500, detail=str(e)) def _load_default_solver_settings() -> dict: path = os.path.join(os.path.dirname(__file__), "comprehensive_settings.json") if os.path.exists(path): with open(path, "r", encoding="utf-8") as f: return json.load(f) return {} @app.get("/api/solver/default-settings") def get_solver_default_settings(): return _load_default_solver_settings() @app.post("/api/solve") async def run_solver(payload: SolveRequest): try: data_dict = ( payload.model_dump() if hasattr(payload, "model_dump") else payload.dict() ) # 2. Run the Data Prepper solver_input = prep_solver_data(data_dict) # 3. Run the Math Engine optimal_moves = run_milp_model(solver_input) return optimal_moves except Exception as e: print(f"Error: {e}") raise HTTPException(status_code=400, detail=str(e)) class SensitivityRequest(SolveRequest): num_sims: int = 50 analysis_gw: Optional[int] = None # kept for compat, ignored internally @app.post("/api/sensitivity") async def run_sensitivity_analysis(payload: SensitivityRequest): """ Runs num_sims solves (iterations=1, per-player noise). For regular GWs aggregates buy/sell/move transfers; for WC/FH GWs aggregates squad selection % (PSB), lineup %, and positional lineup combinations. """ import random from collections import Counter try: data_dict = ( payload.model_dump() if hasattr(payload, "model_dump") else payload.dict() ) num_sims = int(data_dict.pop("num_sims", 50)) data_dict.pop("analysis_gw", None) horizon_gws = [int(g) for g in data_dict.get("horizon_gws", [])] base_settings = dict(data_dict.get("settings") or {}) base_settings["iterations"] = 1 # Detect which GWs have WC/FH active chip_free_gws: set[int] = set() for chip_key in ("use_wc", "use_fh"): for g in base_settings.get(chip_key) or []: chip_free_gws.add(int(g)) id_to_name: dict[int, str] = {} id_to_pos: dict[int, str] = {} for p in data_dict["market_players"]: pid = int(p["id"]) id_to_name[pid] = p["name"] id_to_pos[pid] = p["pos"] print( f"Sensitivity: running {num_sims} sims across {len(horizon_gws)} GWs " f"(chip-free GWs: {chip_free_gws or 'none'})..." ) # Regular GW accumulators gw_data: dict[int, dict] = { gw: {"buys": {}, "sells": {}, "moves": {}, "lineups": {}, "no_transfer": 0} for gw in horizon_gws if gw not in chip_free_gws } # WC/FH GW accumulators: per-player squad & lineup counts + combo counters wc_data: dict[int, dict] = { gw: { "squad": {}, # {name: count} (in the 15-man squad) "lineup": {}, # {name: count} (in the 11-man lineup) "combos": { "G": Counter(), "D": Counter(), "M": Counter(), "F": Counter(), }, } for gw in horizon_gws if gw in chip_free_gws } valid_runs = 0 for sim_idx in range(num_sims): noisy_players = [] for p in data_dict["market_players"]: noise = random.gauss(1.0, 0.12) noise = max(0.3, min(2.5, noise)) noisy_evs = {k: round(float(v) * noise, 4) for k, v in p["evs"].items()} noisy_players.append({**p, "evs": noisy_evs}) sim_data = { **data_dict, "market_players": noisy_players, "settings": {**base_settings}, } try: solver_input = prep_solver_data(sim_data) result = run_milp_model(solver_input) if result["status"] != "success" or not result["solutions"]: continue sol = result["solutions"][0] except Exception as sim_err: print(f" Sim {sim_idx + 1} failed: {sim_err}") continue valid_runs += 1 for gw_plan in sol["plan"]: gw = gw_plan["gw"] if gw in wc_data: # --- WC/FH GW: squad & lineup selection --- all_ids = list(gw_plan.get("lineup", [])) + list( gw_plan.get("bench", []) ) lineup_ids = set(gw_plan.get("lineup", [])) for pid in all_ids: name = id_to_name.get(pid, str(pid)) wc_data[gw]["squad"][name] = ( wc_data[gw]["squad"].get(name, 0) + 1 ) if pid in lineup_ids: wc_data[gw]["lineup"][name] = ( wc_data[gw]["lineup"].get(name, 0) + 1 ) # Lineup combos per position for pid in lineup_ids: pass # we accumulate below pos_players: dict[str, list[str]] = { "G": [], "D": [], "M": [], "F": [], } for pid in sorted(lineup_ids): pos = id_to_pos.get(pid, "M") name = id_to_name.get(pid, str(pid)) if pos in pos_players: pos_players[pos].append(name) for pos, names in pos_players.items(): combo = frozenset(names) if combo: wc_data[gw]["combos"][pos][combo] += 1 elif gw in gw_data: # --- Regular GW: buy/sell/move --- transfers_out_ids = gw_plan.get("transfers_out", []) transfers_in_ids = gw_plan.get("transfers_in", []) if not transfers_in_ids: gw_data[gw]["no_transfer"] += 1 else: buy_names = [ id_to_name.get(pid, str(pid)) for pid in transfers_in_ids ] sell_names = [ id_to_name.get(pid, str(pid)) for pid in transfers_out_ids ] for name in buy_names: gw_data[gw]["buys"][name] = ( gw_data[gw]["buys"].get(name, 0) + 1 ) for name in sell_names: gw_data[gw]["sells"][name] = ( gw_data[gw]["sells"].get(name, 0) + 1 ) sorted_buys = sorted(buy_names) sorted_sells = sorted(sell_names) if sorted_buys and sorted_sells: mk = ( f"{', '.join(sorted_sells)} -> {', '.join(sorted_buys)}" ) gw_data[gw]["moves"][mk] = ( gw_data[gw]["moves"].get(mk, 0) + 1 ) for pid in gw_plan.get("lineup", []): name = id_to_name.get(pid, str(pid)) gw_data[gw]["lineups"][name] = ( gw_data[gw]["lineups"].get(name, 0) + 1 ) if valid_runs == 0: raise Exception( "All sensitivity simulations failed. Check squad/budget settings." ) def to_pct_list(counter: dict, top_n: int = 20) -> list: return [ {"name": k, "count": v, "pct": round(v / valid_runs * 100, 1)} for k, v in sorted(counter.items(), key=lambda x: -x[1])[:top_n] ] name_to_pos: dict[str, str] = {v: id_to_pos[k] for k, v in id_to_name.items()} gw_results: dict[str, dict] = {} # --- Regular GW results --- for gw in horizon_gws: if gw in chip_free_gws: continue if gw not in gw_data: continue d = gw_data[gw] pos_groups: dict[str, list] = {"G": [], "D": [], "M": [], "F": []} for name, cnt in sorted(d["lineups"].items(), key=lambda x: -x[1]): pos = name_to_pos.get(name, "M") pct = round(cnt / valid_runs * 100, 1) if pos in pos_groups: pos_groups[pos].append({"name": name, "pct": pct, "count": cnt}) gw_results[str(gw)] = { "is_chip_free": False, "buys": to_pct_list(d["buys"]), "sells": to_pct_list(d["sells"]), "moves": to_pct_list(d["moves"]), "lineups": {pos: rows[:8] for pos, rows in pos_groups.items()}, "no_transfer_pct": round(d["no_transfer"] / valid_runs * 100, 1), } # --- WC/FH GW results --- for gw in horizon_gws: if gw not in wc_data: continue wd = wc_data[gw] # Per-player squad/lineup pct grouped by position player_data_by_pos: dict[str, list] = {"G": [], "D": [], "M": [], "F": []} all_names = set(list(wd["squad"].keys()) + list(wd["lineup"].keys())) for name in all_names: sq_cnt = wd["squad"].get(name, 0) lu_cnt = wd["lineup"].get(name, 0) pos = name_to_pos.get(name, "M") if pos in player_data_by_pos: player_data_by_pos[pos].append( { "name": name, "squad_pct": round(sq_cnt / valid_runs * 100, 1), "lineup_pct": round(lu_cnt / valid_runs * 100, 1), "squad_count": sq_cnt, "lineup_count": lu_cnt, } ) # Sort each position by squad_pct descending for pos in player_data_by_pos: player_data_by_pos[pos].sort(key=lambda x: -x["squad_pct"]) player_data_by_pos[pos] = player_data_by_pos[pos][:12] # Lineup combos per position (top 5) combo_data: dict[str, list] = {} for pos in ("G", "D", "M", "F"): combos = wd["combos"][pos] sorted_combos = sorted(combos.items(), key=lambda x: -x[1])[:5] combo_data[pos] = [ { "combination": ", ".join(sorted(combo)), "pct": round(cnt / valid_runs * 100, 1), "count": cnt, } for combo, cnt in sorted_combos ] gw_results[str(gw)] = { "is_chip_free": True, "players": player_data_by_pos, "combos": combo_data, } return { "status": "success", "num_sims": num_sims, "valid_runs": valid_runs, "horizon_gws": horizon_gws, "gw_results": gw_results, } except Exception as e: print(f"Sensitivity error: {e}") raise HTTPException(status_code=400, detail=str(e)) def _generate_chip_combos(chip_gw_options: dict) -> list[dict]: """ Generate all valid chip combinations from the per-chip GW option lists. Rules: - At most one chip per GW. - Each chip type used at most once. - Returns list of dicts like {"use_wc": [], "use_fh": [37], "use_bb": [33], "use_tc": []}. """ chip_types = ["wc", "fh", "bb", "tc"] options: list[list] = [] for c in chip_types: gws = [int(g) for g in (chip_gw_options.get(c) or [])] options.append([None] + gws) # None = don't use this chip valid: list[dict] = [] for combo in _itertools.product(*options): # combo = (wc_gw|None, fh_gw|None, bb_gw|None, tc_gw|None) used_gws = [g for g in combo if g is not None] if len(used_gws) != len(set(used_gws)): continue # Two chips assigned to same GW — invalid valid.append( { f"use_{c}": ([g] if g is not None else []) for c, g in zip(chip_types, combo) } ) return valid @app.post("/api/chip-solve") async def run_chip_solver(payload: ChipSolveRequest): """ Evaluate all valid chip combinations from the supplied option lists and return the top solutions ranked by objective score. Chip-solve uses fixed settings: decay=1.017, ft_value=0, ft_value_list all zeros, itb_value=0, ft_use_penalty=0. """ try: data_dict = ( payload.model_dump() if hasattr(payload, "model_dump") else payload.dict() ) chip_gw_options: dict = data_dict.pop("chip_gw_options", {}) # Fixed chip-solve settings (per run_parallel.py conventions) chip_fixed = { "decay_base": 1.017, "ft_value": 0.0, "ft_value_list": {}, "itb_value": 0.0, "ft_use_penalty": 0.0, "no_transfer_last_gws": 0, "iterations": 1, } base_settings = {**data_dict.get("settings", {}), **chip_fixed} combos = _generate_chip_combos(chip_gw_options) if not combos: raise Exception( "No valid chip combinations generated from the supplied GW options." ) # Cap at 30 combinations to keep runtime reasonable combos = combos[:30] print(f"Chip solve: evaluating {len(combos)} valid chip combination(s)...") all_solutions = [] for idx, combo in enumerate(combos): combo_settings = {**base_settings, **combo} combo_data = {**data_dict, "settings": combo_settings} try: solver_input = prep_solver_data(combo_data) result = run_milp_model(solver_input) if result["status"] == "success" and result["solutions"]: sol = result["solutions"][0] sol["chip_combo"] = combo # tag which chips were used sol["combo_id"] = idx + 1 all_solutions.append(sol) except Exception as combo_err: print(f" Combo {idx + 1} ({combo}) failed: {combo_err}") continue if not all_solutions: raise Exception("All chip combinations failed to find optimal solutions.") all_solutions.sort( key=lambda s: -(float(s.get("objective_score") or s.get("ev") or 0)) ) return {"status": "success", "solutions": all_solutions} except Exception as e: print(f"Chip solve error: {e}") raise HTTPException(status_code=400, detail=str(e)) @app.post("/api/settings/save") async def save_user_settings(payload: SettingsPayload, db: Session = Depends(get_db)): user = db.query(User).filter(User.default_team_id == payload.team_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") user.solver_settings = { "quick": payload.quick_settings, "advanced": payload.advanced_settings, } # THE FIX: Violently force SQLAlchemy to commit the JSON column flag_modified(user, "solver_settings") db.commit() return {"status": "success", "message": "Settings saved to cloud."} # 3. The LOAD Route (GET) @app.get("/api/settings/{team_id}") async def load_user_settings(team_id: int, db: Session = Depends(get_db)): # Find the user by their team_id user = db.query(User).filter(User.default_team_id == team_id).first() if not user or not user.solver_settings: # If no user or no settings, return nulls so React uses local defaults return {"status": "success", "quick_settings": None, "advanced_settings": None} return { "status": "success", "quick_settings": user.solver_settings.get("quick"), "advanced_settings": user.solver_settings.get("advanced"), } class FixtureOverrideRequest(BaseModel): overrides: Dict[str, Any] is_admin: bool = False admin_password: Optional[str] = None @app.get("/api/fixtures/overrides") def get_fixture_overrides(): # Serves the global fixtures to everyone who loads the website return app_data.admin_fixture_overrides @app.post("/api/fixtures/update") def update_fixture_overrides(req: FixtureOverrideRequest): if req.is_admin: if req.admin_password != ADMIN_PASSWORD: raise HTTPException(status_code=401, detail="Invalid admin password") # 1. Update Python's active RAM instantly! app_data.admin_fixture_overrides = req.overrides # 2. Save it to the Hard Drive permanently save_config_to_db("admin_fixtures", app_data.admin_fixture_overrides) return {"status": "success", "message": "Global fixtures updated!"} raise HTTPException(status_code=401, detail="Unauthorized") @app.get("/api/xmins/overrides") def get_xmins_overrides(): return app_data.admin_xmins_overrides @app.get("/api/ratings_history") def get_ratings_history(): try: # Reads your history CSV and sends it to React as a JSON array import pandas as pd df = pd.read_csv("team_ratings_history.csv") return df.to_dict(orient="records") except Exception: return []