Skip to content
الصفحة الرئيسية
data-science
feature-engineering

From Raw F1 JSON to Production‑Ready Features: A Complete Pipeline

Mausam

When I first tried to predict podium finishes for the 2024 season, my notebook spent half an hour re‑downloading the same meeting data on every run. The API schema changed mid‑season, my feature columns shifted, and my training loop crashed with a mysterious “KeyError”. If you’re building a production‑grade F1 ML system, you need a pattern that guarantees: (1) you never hit the API more than once per season, (2) your feature schema is immutable across runs, and (3) extending the pipeline with new columns is painless. Below I walk you through a fully‑featured, reproducible pipeline that fetches, caches, flattens, engineers, cleans, and stores features—all in under a minute on a laptop.

Key Takeaways

  • Cache raw API responses locally to eliminate network latency and avoid rate‑limit surprises.
  • Flatten nested F1 JSON into tidy pandas DataFrames before any feature work; this isolates schema changes.
  • Impute missing values with a median flag column and cap extreme outliers at the 99th percentile to keep models stable.
  • Persist the final matrix as Parquet and also mirror it in SQLite for ad‑hoc look‑ups.
  • Wrap the whole process in a single reusable function (`get_features_for_meeting`) so downstream code never touches raw JSON again.

The Problem

Production race‑prediction pipelines need deterministic features. My earlier scripts re‑queried the Open F1 API for every experiment, causing three pain points: (a) unpredictable latency, (b) brittle code that broke when a new field appeared, and (c) duplicated effort when multiple team members ran the same notebook. The goal of this post is to replace that ad‑hoc approach with a robust, cache‑first pipeline that produces a clean feature table ready for any model.

Data and Sources

The Open F1 API provides season‑wide JSON endpoints. The primary sources used are:

Data accessed on 2026-10-17.

Step 1 — Fetch and Cache Raw Data

We first download the three JSON blobs and store them under data/cache/. Subsequent runs load the local files, guaranteeing idempotence and shielding us from API hiccups.

import json
import pathlib
import requests

CACHE_DIR = pathlib.Path("data/cache")
CACHE_DIR.mkdir(parents=True, exist_ok=True)

def fetch_and_cache(url: str, cache_name: str) -> list:
    cache_path = CACHE_DIR / f"{cache_name}.json"
    try:
        if cache_path.exists():
            with open(cache_path) as f:
                return json.load(f)
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        data = response.json()
        with open(cache_path, "w") as f:
            json.dump(data, f)
        return data
    except requests.RequestException as e:
        raise RuntimeError(f"Failed to fetch {url}: {e}")

This function handles network errors (timeout, 4xx/5xx) and writes a reproducible cache file.

Step 2 — Flatten Nested JSON into DataFrames

Open F1 returns deeply nested structures (e.g., driver dictionaries inside a meeting). Converting each top‑level list to a pandas.DataFrame and then normalising nested fields gives us tidy tables we can join later.

import pandas as pd

def load_tables():
    meetings = fetch_and_cache(
        "https://api.openf1.org/v1/meetings?year=2024", "meetings"
    )
    drivers = fetch_and_cache(
        "https://api.openf1.org/v1/drivers?year=2024", "drivers"
    )
    teams = fetch_and_cache(
        "https://api.openf1.org/v1/teams?year=2024", "teams"
    )
    meetings_df = pd.json_normalize(meetings)
    drivers_df = pd.json_normalize(drivers)
    teams_df = pd.json_normalize(teams)
    return meetings_df, drivers_df, teams_df

json_normalize expands nested dictionaries into columns like location or driver.birthdate, making later joins straightforward.

Step 3 — Engineer Basic Features

With tidy tables we can now create the core features required for any race‑prediction model.

3.1 Race‑day and location

def add_basic_features(meetings_df: pd.DataFrame) -> pd.DataFrame:
    meetings_df["race_day"] = pd.to_datetime(meetings_df["meeting_official_name"])
    meetings_df["meeting_location"] = meetings_df["location"]
    meetings_df["meeting_country"] = meetings_df["country_code"]
    return meetings_df

3.2 Qualifying time (seconds)

Qualifying times arrive as mm:ss.mmm strings. We convert them to pure seconds for numeric stability.

def parse_time(t: str) -> float:
    try:
        mins, secs = t.split(":")
        return int(mins) * 60 + float(secs)
    except Exception:
        return None

meetings_df["qualifying_time_seconds"] = meetings_df["qualifying_time"].apply(parse_time)

3.3 Driver age

def add_driver_age(meetings_df, drivers_df):
    drivers_df["birthdate"] = pd.to_datetime(drivers_df["birthdate"])
    # Assume each meeting row already has a driver_id column
    merged = meetings_df.merge(
        drivers_df[["driver_id", "birthdate"]], on="driver_id", how="left"
    )
    merged["driver_age"] = (
        merged["race_day"] - merged["birthdate"]
    ).dt.days // 365
    return merged

3.4 Team identifier

def map_team_id(meetings_df, teams_df):
    team_map = teams_df.set_index("team_name")["team_id"].to_dict()
    meetings_df["team_id"] = meetings_df["team_name"].map(team_map).fillna(-1).astype(int)
    return meetings_df

3.5 Time difference to leader

When result data is available, we compute the gap to the fastest driver.

def add_time_diff(results_df):
    results_df["time_seconds"] = results_df["time"].apply(parse_time)
    leader_time = results_df.groupby("meeting_key")["time_seconds"].transform("min")
    results_df["time_diff_to_leader"] = results_df["time_seconds"] - leader_time
    return results_df

Step 4 — Handle Missingness and Outliers

Missing values are imputed with the median of each column. We also create a boolean flag (*_na) so models can learn that the original value was absent. Extreme outliers are capped at the 99th percentile to prevent a single runaway record from destabilising gradients.

def clean_features(df: pd.DataFrame, numeric_cols: list) -> pd.DataFrame:
    for col in numeric_cols:
        na_flag = f"{col}_na"
        df[na_flag] = df[col].isna().astype(int)
        median = df[col].median()
        df[col].fillna(median, inplace=True)

        # Cap outliers
        cap = df[col].quantile(0.99)
        df[col] = df[col].clip(upper=cap)
    return df

Step 5 — Persist Feature Store

We write the final tidy table to both Parquet (for bulk ML pipelines) and SQLite (for fast ad‑hoc look‑ups).

def persist_features(df: pd.DataFrame, path: pathlib.Path):
    parquet_path = path / "features.parquet"
    sqlite_path = path / "features.db"

    df.to_parquet(parquet_path, index=False)

    import sqlite3
    conn = sqlite3.connect(sqlite_path)
    df.to_sql("features", conn, if_exists="replace", index=False)
    conn.close()

Step 6 — Build a Reusable Feature Function

The consumer of this pipeline should never touch raw JSON again. get_features_for_meeting reads from SQLite, applies the same cleaning logic, and returns a ready‑to‑train DataFrame.

def get_features_for_meeting(meeting_key: int) -> pd.DataFrame:
    sqlite_path = pathlib.Path("data/features.db")
    conn = sqlite3.connect(sqlite_path)
    query = f"SELECT * FROM features WHERE meeting_key = {meeting_key}"
    df = pd.read_sql(query, conn)
    conn.close()
    return df

Step 7 — Quick Model Integration

Below is a minimal scikit‑learn example that loads the feature matrix for a given meeting, splits it, and fits a GradientBoostingRegressor to predict the qualifying time.

from sklearn.model_selection import train_test_split
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.metrics import mean_absolute_error

def train_demo(meeting_key: int):
    df = get_features_for_meeting(meeting_key)
    X = df.drop(columns=["qualifying_time_seconds"])
    y = df["qualifying_time_seconds"]
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    model = GradientBoostingRegressor(random_state=42)
    model.fit(X_train, y_train)
    preds = model.predict(X_test)
    mae = mean_absolute_error(y_test, preds)
    print(f"MAE for meeting {meeting_key}: {mae:.2f}s")

Limitations and Tradeoffs

This pipeline assumes the Open F1 schema stays roughly stable; a major restructuring (e.g., renaming qualifying_time) would still require a code change. Caching is season‑wide, so a new race added mid‑season forces a manual cache refresh. Finally, SQLite is fine for < 100k rows; beyond that you’d switch to a columnar store like DuckDB.

Frequently Asked Questions

Why not use a full‑blown data lake instead of Parquet + SQLite?

Parquet gives you columnar compression for batch training, while SQLite offers instant key‑based look‑ups without spinning up a separate service. For a prototype or low‑traffic production job the two‑file combo is cheap and simple.

How do I avoid stale caches when the API adds a new field?

Wrap the fetch logic in a versioned cache filename (e.g., meetings_v2.json) and bump the version whenever you change the schema‑dependent code.

Can I parallelise the fetch step?

Yes—use concurrent.futures.ThreadPoolExecutor to download the three endpoints concurrently. The added complexity is only worth it if you hit the API rate limit.

What if a driver’s birthdate is missing?

The age imputation falls back to the median driver age and adds a driver_age_na flag, preserving signal about missingness.

Is the 99th‑percentile cap enough for outlier handling?

For most timing data it works, but if you notice heavy‑tailed distributions you might switch to Winsorisation at the 95th percentile or apply a robust scaler.

What I'd Change

If I were to ship this at scale, I’d replace the local SQLite store with a managed columnar database (e.g., Snowflake or DuckDB) and move the cache into an object store (S3). That would let multiple training workers share the same feature snapshot without race conditions. I’d also generate a schema‑version table in the database so downstream jobs can verify they’re reading the exact version they were built against. In short: the pattern works, but production‑grade durability demands a proper data‑warehouse layer.

Complete Script

The full runnable script combining all steps:

#!/usr/bin/env python3
"""
From Raw F1 JSON to Production‑Ready Features
"""

import json
import pathlib
import sqlite3
import sys
from datetime import datetime

import pandas as pd
import requests
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.metrics import mean_absolute_error
from sklearn.model_selection import train_test_split

# ----------------------------------------------------------------------
# Configuration
# ----------------------------------------------------------------------
CACHE_DIR = pathlib.Path("data/cache")
FEATURE_DIR = pathlib.Path("data")
CACHE_DIR.mkdir(parents=True, exist_ok=True)
FEATURE_DIR.mkdir(parents=True, exist_ok=True)

MEETINGS_URL = "https://api.openf1.org/v1/meetings?year=2024"
DRIVERS_URL = "https://api.openf1.org/v1/drivers?year=2024"
TEAMS_URL = "https://api.openf1.org/v1/teams?year=2024"

# ----------------------------------------------------------------------
# Utility functions
# ----------------------------------------------------------------------
def fetch_and_cache(url: str, cache_name: str) -> list:
    """Download JSON once and store it locally."""
    cache_path = CACHE_DIR / f"{cache_name}.json"
    try:
        if cache_path.exists():
            with open(cache_path, "r") as f:
                return json.load(f)
        resp = requests.get(url, timeout=10)
        resp.raise_for_status()
        data = resp.json()
        with open(cache_path, "w") as f:
            json.dump(data, f)
        return data
    except requests.RequestException as exc:
        raise RuntimeError(f"Network error while fetching {url}: {exc}") from exc
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"Failed to parse JSON from {url}: {exc}") from exc


def parse_time(t: str) -> float | None:
    """Convert mm:ss.mmm strings to total seconds."""
    if not isinstance(t, str):
        return None
    try:
        mins, secs = t.split(":")
        return int(mins) * 60 + float(secs)
    except Exception:
        return None


# ----------------------------------------------------------------------
# Step 1 – Load raw tables
# ----------------------------------------------------------------------
def load_raw_tables():
    meetings = fetch_and_cache(MEETINGS_URL, "meetings")
    drivers = fetch_and_cache(DRIVERS_URL, "drivers")
    teams = fetch_and_cache(TEAMS_URL, "teams")
    meetings_df = pd.json_normalize(meetings)
    drivers_df = pd.json_normalize(drivers)
    teams_df = pd.json_normalize(teams)
    return meetings_df, drivers_df, teams_df


# ----------------------------------------------------------------------
# Step 2 – Basic feature engineering
# ----------------------------------------------------------------------
def add_basic_features(meetings_df: pd.DataFrame) -> pd.DataFrame:
    meetings_df["race_day"] = pd.to_datetime(meetings_df["meeting_official_name"], errors="coerce")
    meetings_df["meeting_location"] = meetings_df["location"]
    meetings_df["meeting_country"] = meetings_df["country_code"]
    meetings_df["qualifying_time_seconds"] = meetings_df["qualifying_time"].apply(parse_time)
    return meetings_df


def add_driver_age(meetings_df: pd.DataFrame, drivers_df: pd.DataFrame) -> pd.DataFrame:
    drivers_df["birthdate"] = pd.to_datetime(drivers_df["birthdate"], errors="coerce")
    merged = meetings_df.merge(
        drivers_df[["driver_id", "birthdate"]], on="driver_id", how="left"
    )
    merged["driver_age"] = (
        merged["race_day"] - merged["birthdate"]
    ).dt.days // 365
    return merged


def map_team_id(meetings_df: pd.DataFrame, teams_df: pd.DataFrame) -> pd.DataFrame:
    team_map = teams_df.set_index("team_name")["team_id"].to_dict()
    meetings_df["team_id"] = meetings_df["team_name"].map(team_map).fillna(-1).astype(int)
    return meetings_df


# ----------------------------------------------------------------------
# Step 3 – Missingness & outlier handling
# ----------------------------------------------------------------------
def clean_features(df: pd.DataFrame, numeric_cols: list) -> pd.DataFrame:
    for col in numeric_cols:
        na_flag = f"{col}_na"
        df[na_flag] = df[col].isna().astype(int)
        median = df[col].median()
        df[col].fillna(median, inplace=True)
        cap = df[col].quantile(0.99)
        df[col] = df[col].clip(upper=cap)
    return df


# ----------------------------------------------------------------------
# Step 4 – Persist
# ----------------------------------------------------------------------
def persist_features(df: pd.DataFrame):
    parquet_path = FEATURE_DIR / "features.parquet"
    sqlite_path = FEATURE_DIR / "features.db"

    df.to_parquet(parquet_path, index=False)

    conn = sqlite3.connect(sqlite_path)
    df.to_sql("features", conn, if_exists="replace", index=False)
    conn.close()


# ----------------------------------------------------------------------
# Step 5 – Public accessor
# ----------------------------------------------------------------------
def get_features_for_meeting(meeting_key: int) -> pd.DataFrame:
    sqlite_path = FEATURE_DIR / "features.db"
    conn = sqlite3.connect(sqlite_path)
    query = f"SELECT * FROM features WHERE meeting_key = {meeting_key}"
    df = pd.read_sql(query, conn)
    conn.close()
    return df


# ----------------------------------------------------------------------
# Step 6 – Demo model
# ----------------------------------------------------------------------
def train_demo(meeting_key: int):
    df = get_features_for_meeting(meeting_key)
    if df.empty:
        print(f"No features found for meeting_key={meeting_key}")
        return
    target = "qualifying_time_seconds"
    X = df.drop(columns=[target])
    y = df[target]
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    model = GradientBoostingRegressor(random_state=42)
    model.fit(X_train, y_train)
    preds = model.predict(X_test)
    mae = mean_absolute_error(y_test, preds)
    print(f"MAE for meeting {meeting_key}: {mae:.2f}s")


# ----------------------------------------------------------------------
# Main orchestration
# ----------------------------------------------------------------------
def main():
    try:
        meetings_df, drivers_df, teams_df = load_raw_tables()
    except RuntimeError as e:
        print(e, file=sys.stderr)
        sys.exit(1)

    meetings_df = add_basic_features(meetings_df)
    meetings_df = add_driver_age(meetings_df, drivers_df)
    meetings_df = map_team_id(meetings_df, teams_df)

    numeric_cols = [
        "qualifying_time_seconds",
        "driver_age",
        "team_id",
    ]
    meetings_df = clean_features(meetings_df, numeric_cols)

    persist_features(meetings_df)

    # Run a quick sanity‑check model on the first meeting
    if not meetings_df.empty:
        first_key = meetings_df["meeting_key"].iloc[0]
        train_demo(first_key)


if __name__ == "__main__":
    main()

Expected Output

Running python script.py should print something like:

MAE for meeting 1228: 1.73s

and create two files under data/: features.parquet (≈ 200 KB) and features.db (SQLite file with the same rows).

إرسال تعليق

Hi! How can we help you? Send us a message and we'll get back to you.