import json import os import pickle import numpy as np import pandas as pd import streamlit as st from pathlib import Path from openai import OpenAI # Config _ROOT = Path(__file__).resolve().parent MODELS_DIR = _ROOT / "models" PROCESSED_DIR = _ROOT / "data" / "processed" _GPT_MODEL = "gpt-4o-mini" # Lookup tables DRIVER_ENC = { "aitken": 0, "albon": 1, "alonso": 2, "antonelli": 3, "bearman": 4, "bortoleto": 5, "bottas": 6, "colapinto": 7, "de_vries": 8, "doohan": 9, "gasly": 10, "giovinazzi": 11, "grosjean": 12, "hadjar": 13, "hamilton": 14, "hulkenberg": 15, "kevin_magnussen": 16, "kubica": 17, "kvyat": 18, "latifi": 19, "lawson": 20, "leclerc": 21, "max_verstappen": 22, "mazepin": 23, "mick_schumacher": 24, "norris": 25, "ocon": 26, "perez": 27, "piastri": 28, "pietro_fittipaldi": 29, "raikkonen": 30, "ricciardo": 31, "russell": 32, "sainz": 33, "sargeant": 34, "stroll": 35, "tsunoda": 36, "vettel": 37, "zhou": 38, } CONSTRUCTOR_ENC = { "alfa": 0, "alphatauri": 1, "alpine": 2, "aston_martin": 3, "ferrari": 4, "haas": 5, "mclaren": 6, "mercedes": 7, "racing_point": 8, "rb": 9, "red_bull": 10, "renault": 11, "sauber": 12, "williams": 13, } CIRCUIT_ENC = { "albert_park": 0, "americas": 1, "bahrain": 2, "baku": 3, "catalunya": 4, "hungaroring": 5, "imola": 6, "interlagos": 7, "istanbul": 8, "jeddah": 9, "losail": 10, "marina_bay": 11, "miami": 12, "monaco": 13, "monza": 14, "mugello": 15, "nurburgring": 16, "portimao": 17, "red_bull_ring": 18, "ricard": 19, "rodriguez": 20, "shanghai": 21, "silverstone": 22, "sochi": 23, "spa": 24, "suzuka": 25, "vegas": 26, "villeneuve": 27, "yas_marina": 28, "zandvoort": 29, } DRIVER_DISPLAY = { "aitken": "Jack Aitken", "albon": "Alexander Albon", "alonso": "Fernando Alonso", "antonelli": "Kimi Antonelli", "bearman": "Oliver Bearman", "bortoleto": "Gabriel Bortoleto", "bottas": "Valtteri Bottas", "colapinto": "Franco Colapinto", "de_vries": "Nyck de Vries", "doohan": "Jack Doohan", "gasly": "Pierre Gasly", "giovinazzi": "Antonio Giovinazzi", "grosjean": "Romain Grosjean", "hadjar": "Isack Hadjar", "hamilton": "Lewis Hamilton", "hulkenberg": "Nico Hulkenberg", "kevin_magnussen": "Kevin Magnussen", "kubica": "Robert Kubica", "kvyat": "Daniil Kvyat", "latifi": "Nicholas Latifi", "lawson": "Liam Lawson", "leclerc": "Charles Leclerc", "max_verstappen": "Max Verstappen", "mazepin": "Nikita Mazepin", "mick_schumacher": "Mick Schumacher", "norris": "Lando Norris", "ocon": "Esteban Ocon", "perez": "Sergio Perez", "piastri": "Oscar Piastri", "pietro_fittipaldi": "Pietro Fittipaldi", "raikkonen": "Kimi Raikkonen", "ricciardo": "Daniel Ricciardo", "russell": "George Russell", "sainz": "Carlos Sainz", "sargeant": "Logan Sargeant", "stroll": "Lance Stroll", "tsunoda": "Yuki Tsunoda", "vettel": "Sebastian Vettel", "zhou": "Guanyu Zhou", } CIRCUIT_DISPLAY = { "albert_park": "Albert Park", "americas": "Circuit of the Americas", "bahrain": "Bahrain International Circuit", "baku": "Baku City Circuit", "catalunya": "Circuit de Barcelona-Catalunya", "hungaroring": "Hungaroring", "imola": "Imola", "interlagos": "Interlagos", "istanbul": "Istanbul Park", "jeddah": "Jeddah Corniche Circuit", "losail": "Losail International Circuit", "marina_bay": "Marina Bay Street Circuit", "miami": "Miami International Autodrome", "monaco": "Circuit de Monaco", "monza": "Monza", "mugello": "Mugello", "nurburgring": "Nurburgring", "portimao": "Algarve International Circuit", "red_bull_ring": "Red Bull Ring", "ricard": "Circuit Paul Ricard", "rodriguez": "Autodromo Hermanos Rodriguez", "shanghai": "Shanghai International Circuit", "silverstone": "Silverstone", "sochi": "Sochi Autodrom", "spa": "Circuit de Spa-Francorchamps", "suzuka": "Suzuka International Racing Course", "vegas": "Las Vegas Street Circuit", "villeneuve": "Circuit Gilles Villeneuve", "yas_marina": "Yas Marina Circuit", "zandvoort": "Circuit Zandvoort", } # Data and model loading @st.cache_resource def load_model(): with open(MODELS_DIR / "f1_prediction_model.pkl", "rb") as f: model = pickle.load(f) with open(MODELS_DIR / "metadata.json") as f: metadata = json.load(f) return model, metadata @st.cache_data def load_data(): return pd.read_csv(PROCESSED_DIR / "f1_features.csv") # Feature engineering def _driver_history(df: pd.DataFrame, driver: str, circuit: str) -> dict: drv = df[df["driver_id"] == driver] crc = drv[drv["circuit_id"] == circuit] def _med(series, fallback=None): v = series.dropna() if len(v): return float(v.median()) if fallback is not None: v2 = fallback.dropna() return float(v2.median()) if len(v2) else np.nan return np.nan return { "driver_avg_change": _med(drv["driver_avg_change"]), "driver_circuit_avg": _med(crc["driver_circuit_avg"], drv["driver_avg_change"]), "constructor_avg_change": _med(drv["constructor_avg_change"]), "driver_dnf_rate": _med(drv["driver_dnf_rate"]), "constructor_dnf_rate": _med(drv["constructor_dnf_rate"]), "championship_position": _med(drv["championship_position"]), "quali_gap_to_pole": _med(crc["quali_gap_to_pole"], drv["quali_gap_to_pole"]), "quali_time_sec": _med(crc["quali_time_sec"], drv["quali_time_sec"]), "quali_speed_fl": _med(crc["quali_speed_fl"], drv["quali_speed_fl"]), "driver_race_pace": _med(drv["driver_race_pace"]), } def build_feature_row( df, driver, circuit, grid_position, temp_max, precipitation, windspeed_max, latest_season, latest_round, driver_constructor, ): constructor = driver_constructor.get(driver, "williams") history = _driver_history(df, driver, circuit) row = pd.DataFrame([{ "grid_position": grid_position, "driver_id_enc": DRIVER_ENC.get(driver, 0), "constructor_id_enc": CONSTRUCTOR_ENC.get(constructor, 0), "circuit_id_enc": CIRCUIT_ENC.get(circuit, 0), "season": latest_season, "round": latest_round, "temp_max": temp_max, "precipitation": precipitation, "windspeed_max": windspeed_max, "driver_avg_change": history["driver_avg_change"], "driver_circuit_avg": history["driver_circuit_avg"], "constructor_avg_change": history["constructor_avg_change"], "is_pitlane_start": 1 if grid_position == 0 else 0, "driver_dnf_rate": history["driver_dnf_rate"], "constructor_dnf_rate": history["constructor_dnf_rate"], "championship_position": history["championship_position"], "quali_time_sec": history["quali_time_sec"], "quali_gap_to_pole": history["quali_gap_to_pole"], "quali_speed_fl": history["quali_speed_fl"], "driver_race_pace": history["driver_race_pace"], }]) return row, history # LLM analysis _FEW_SHOT_EXAMPLES = [ { "driver": "Max Verstappen", "grid": 1, "prediction": "Held", "confidence": 0.72, "analysis": ( "Verstappen started from pole position and confidently defended his lead " "throughout the race. The model predicted 'Held' with high confidence, " "supported by Red Bull Racing's dominant pace and a flawless pit strategy. " "Dry conditions at 28°C favoured a stable race with no strategic surprises." ), }, { "driver": "Carlos Sainz", "grid": 8, "prediction": "Gained", "confidence": 0.61, "analysis": ( "Sainz moved up from P8 to P4, gaining four positions across the race. " "The model predicted 'Gained' based on his historically strong recovery " "drives and Ferrari's superior tyre management strategy. Light rain in the " "second half of the race further played into the Spaniard's hands." ), }, ] def _get_api_key() -> str | None: if key := os.environ.get("OPENAI_API_KEY"): return key try: return st.secrets["OPENAI_API_KEY"] except (KeyError, FileNotFoundError): return None def _call_llm(system_prompt: str, user_prompt: str) -> str: api_key = _get_api_key() if not api_key: raise EnvironmentError( "OPENAI_API_KEY is not set. " "Add it under Settings → Repository secrets in your HF Space." ) client = OpenAI(api_key=api_key) response = client.chat.completions.create( model=_GPT_MODEL, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], max_tokens=350, temperature=0.7, ) return response.choices[0].message.content.strip() def run_analysis( driver, grid, prediction, confidence, history, temp_max, precipitation, windspeed_max, scenario="", ) -> str: drv_avg = history.get("driver_circuit_avg") con_avg = history.get("constructor_avg_change") examples_text = "\n\n".join([ f"Example:\nDriver: {ex['driver']}, Grid: {ex['grid']}, " f"Prediction: {ex['prediction']} (Confidence: {ex['confidence']:.0%})\n" f"Analysis: {ex['analysis']}" for ex in _FEW_SHOT_EXAMPLES ]) system_prompt = ( "You are an F1 race analyst. Explain in 3–4 sentences why the ML model " "predicts this position change for the driver. Reference the specific data " f"provided. Follow the style of the examples below:\n\n{examples_text}" ) user_prompt = ( f"Driver: {driver}\n" f"Grid Position (Qualifying): {grid}\n" f"ML Prediction: {prediction} (Confidence: {confidence:.0%})\n" f"Weather: Temperature {temp_max}°C, Precipitation {precipitation} mm, " f"Wind {windspeed_max} km/h\n" f"Driver's historical avg position change at this circuit: " f"{f'{drv_avg:.1f} positions' if drv_avg is not None and not np.isnan(drv_avg) else 'n/a'}\n" f"Team avg position change: " f"{f'{con_avg:.1f} positions' if con_avg is not None and not np.isnan(con_avg) else 'n/a'}" ) if scenario.strip(): user_prompt += f"\n\nAdditional question / scenario: {scenario}" return _call_llm(system_prompt, user_prompt) # Prediction pipeline def run_prediction( df, model, feature_cols, label_inv, driver_constructor, driver_key, circuit_key, grid_pos, temp_max, precipitation, windspeed_max, latest_season, latest_round, ) -> dict: X_df, history = build_feature_row( df, driver_key, circuit_key, grid_pos, float(temp_max), float(precipitation), float(windspeed_max), latest_season, latest_round, driver_constructor, ) proba = model.predict_proba(X_df[feature_cols])[0].copy() # Physical constraints: P1 cannot gain, P20 cannot lose gained_idx = next((k for k, v in label_inv.items() if v == "Gained"), None) lost_idx = next((k for k, v in label_inv.items() if v == "Lost"), None) if grid_pos == 1 and gained_idx is not None: proba[gained_idx] = 0.0 proba /= proba.sum() if grid_pos == 20 and lost_idx is not None: proba[lost_idx] = 0.0 proba /= proba.sum() pred_label = label_inv[int(np.argmax(proba))] confidence = float(proba.max()) constructor = driver_constructor.get(driver_key, "–").replace("_", " ").title() return { "label": pred_label, "confidence": confidence, "proba": proba.tolist(), "constructor": constructor, "driver_name": DRIVER_DISPLAY.get(driver_key, driver_key), "circuit_name": CIRCUIT_DISPLAY.get(circuit_key, circuit_key), "grid": grid_pos, "history": history, "temp_max": float(temp_max), "precipitation": float(precipitation), "windspeed_max": float(windspeed_max), } # UI st.set_page_config( page_title="F1 Grid-to-Flag Predictor", page_icon="🏎️", layout="wide", ) st.markdown(""" """, unsafe_allow_html=True) st.markdown("""
F1 GRID-TO-FLAG PREDICTOR
ML-powered position change forecast · explained by AI
""", unsafe_allow_html=True) st.markdown("
", unsafe_allow_html=True) try: model, metadata = load_model() df = load_data() except FileNotFoundError: st.error( "Model or data files not found. " "Make sure `models/f1_prediction_model.pkl`, `models/metadata.json`, and " "`data/processed/f1_features.csv` exist." ) st.stop() feature_cols = metadata["feature_columns"] label_inv = {v: k for k, v in metadata["label_map"].items()} class_names = list(metadata["label_map"].keys()) latest_season = int(df["season"].max()) latest_round = int(df[df["season"] == latest_season]["round"].max()) min_season = int(df["season"].min()) driver_constructor = df.groupby("driver_id")["constructor_id"].last().to_dict() st.markdown("### Race Setup") col_d, col_c, col_g = st.columns([3, 3, 1]) with col_d: driver_key = st.selectbox( "Driver", options=sorted(DRIVER_ENC.keys()), format_func=lambda k: DRIVER_DISPLAY.get(k, k.replace("_", " ").title()), index=sorted(DRIVER_ENC.keys()).index("max_verstappen"), ) with col_c: circuit_key = st.selectbox( "Circuit", options=sorted(CIRCUIT_ENC.keys()), format_func=lambda k: CIRCUIT_DISPLAY.get(k, k.replace("_", " ").title()), index=sorted(CIRCUIT_ENC.keys()).index("bahrain"), ) with col_g: grid_pos = st.number_input("Grid", min_value=0, max_value=20, value=1, help="0 = pit-lane start") st.markdown("
", unsafe_allow_html=True) st.markdown("### Weather Conditions") col_t, col_r, col_w = st.columns(3) with col_t: temp_max = st.slider("Temperature (°C)", 5, 45, 28) with col_r: precipitation = st.slider("Precipitation (mm)", 0, 30, 0) with col_w: windspeed_max = st.slider("Wind Speed (km/h)", 5, 60, 18) st.markdown("
", unsafe_allow_html=True) predict_clicked = st.button("Run Prediction", type="primary", use_container_width=True) if predict_clicked or "prediction" in st.session_state: if predict_clicked: st.session_state["prediction"] = run_prediction( df, model, feature_cols, label_inv, driver_constructor, driver_key, circuit_key, grid_pos, temp_max, precipitation, windspeed_max, latest_season, latest_round, ) p = st.session_state["prediction"] st.markdown("
", unsafe_allow_html=True) st.divider() label_color = {"Gained": "#00C853", "Held": "#FFFFFF", "Lost": "#E10600"}.get(p["label"], "#FFFFFF") label_icon = {"Gained": "▲", "Held": "◆", "Lost": "▼"}.get(p["label"], "") st.markdown(f"""
{p['driver_name']} · {p['circuit_name']}
{label_icon} {p['label'].upper()}
Confidence
{p['confidence']:.0%}
""", unsafe_allow_html=True) m1, m2, m3 = st.columns(3) m1.metric("Grid Position", f"P{p['grid']}") m2.metric("Team", p["constructor"]) train_s = metadata["train_seasons"] m3.metric("Trained On", f"{min(train_s)} – {max(train_s)}") st.markdown("
", unsafe_allow_html=True) st.markdown("### Class Probabilities") proba_df = pd.DataFrame({"Probability": p["proba"]}, index=class_names) st.bar_chart(proba_df, color="#E10600") _label_colors = {"Gained": "#00C853", "Held": "#FFFFFF", "Lost": "#E10600"} prob_cols = st.columns(len(class_names)) for col, name, prob in zip(prob_cols, class_names, p["proba"]): color = _label_colors.get(name, "#FFFFFF") col.markdown( f"
" f"
{name}
" f"
{prob:.0%}
" f"
", unsafe_allow_html=True, ) st.divider() st.markdown("### Race Analysis") api_key_set = bool(_get_api_key()) if not api_key_set: st.info( "No OpenAI API key found. " "Add `OPENAI_API_KEY` to `.streamlit/secrets.toml` (local) " "or under **Settings → Repository secrets** on HF Space." ) scenario = st.text_area( "Scenario / Question (optional)", placeholder="e.g. 'What if there is a safety car?' or 'How does rain affect the outcome?'", ) if st.button("Analyse", type="secondary", disabled=not api_key_set, use_container_width=True): with st.spinner("Analysing ..."): try: analysis = run_analysis( driver=p["driver_name"], grid=p["grid"], prediction=p["label"], confidence=p["confidence"], history=p["history"], temp_max=p["temp_max"], precipitation=p["precipitation"], windspeed_max=p["windspeed_max"], scenario=scenario, ) st.markdown(f"""
{analysis}
""", unsafe_allow_html=True) except Exception as e: st.error(f"Analysis error: {e}") st.markdown("
", unsafe_allow_html=True) with st.expander("Model Details"): st.markdown(f"**Best Model:** {metadata['best_model_name']}") st.markdown(f"**Splits:** Train {metadata['train_seasons']} · Val {metadata['val_seasons']} · Test {metadata['test_seasons']}") if metadata.get("iteration_summary"): st.markdown("**CV Iteration Comparison (5-fold, all models):**") iter_df = pd.DataFrame(metadata["iteration_summary"]) iter_df = iter_df.rename(columns={ "iteration": "Iteration", "model": "Model", "cv_f1": "CV F1", "cv_std": "± Std", "cv_acc": "CV Acc", }) iter_df[["CV F1", "± Std", "CV Acc"]] = iter_df[["CV F1", "± Std", "CV Acc"]].round(4) st.dataframe(iter_df.set_index("Iteration"), use_container_width=True) st.markdown("**Final Holdout Evaluation (Val 2024 · Test 2025):**") for name, r in metadata["results"].items(): val = r.get("val_2024", {}) test = r.get("test_2025", {}) marker = " ✓" if name == metadata["best_model_name"] else "" st.markdown( f"- **{name}{marker}** — " f"Val 2024: Acc={val.get('accuracy', 0):.3f}, F1={val.get('f1_weighted', 0):.3f} · " f"Test 2025: Acc={test.get('accuracy', 0):.3f}, F1={test.get('f1_weighted', 0):.3f}" ) st.markdown(f"**Features ({len(feature_cols)}):** {', '.join(feature_cols)}")