Build an AI Tactical Analyst with NFL Data, dbt, and RAG: A Full Data Engineering Pipeline

import os
import subprocess
from pathlib import Path

PROJECT_NAME = "nfl_ai_bootcamp"

def create_project(base_path):
    """Scaffold the dbt + DuckDB project."""
    base = Path(base_path) / PROJECT_NAME
    base.mkdir(parents=True, exist_ok=True)

    (base / "requirements.txt").write_text(
        "dbt-duckdb>=1.7\n"
        "nfl_data_py>=0.3\n"
        "pandas\nnumpy\nopenai>=1.0\n"
    )

    subprocess.run(
        ["dbt", "init", PROJECT_NAME, "--skip-profile-setup"],
        cwd=base, check=True,
    )
    return base

import duckdb
import nfl_data_py as nfl

# Pull 2024 regular season play-by-play
pbp = nfl.import_pbp_data([2024])

con = duckdb.connect("nfl_analytics.duckdb")
con.execute("CREATE SCHEMA IF NOT EXISTS raw")
con.register("pbp_df", pbp)
con.execute("CREATE OR REPLACE TABLE raw.plays AS SELECT * FROM pbp_df")
print(f"Loaded {con.execute('SELECT COUNT(*) FROM raw.plays').fetchone()[0]:,} plays")

# ~/.dbt/profiles.yml
nfl_ai_bootcamp:
  target: dev
  outputs:
    dev:
      type: duckdb
      path: /absolute/path/to/nfl_ai_bootcamp/nfl_analytics.duckdb
      schema: analytics
      threads: 4

# packages.yml
packages:
  - package: dbt-labs/dbt_utils
    version: [">=1.1.0", "<2.0.0"]

dbt deps

-- models/staging/stg_plays.sql

with source as (

    select * from {{ source('raw', 'plays') }}

),

renamed as (

    select
        play_id,
        game_id,
        posteam                 as possession_team,
        defteam                 as defense_team,
        qtr                     as quarter,
        down,
        ydstogo                 as yards_to_go,
        yardline_100,
        play_type,
        passer_player_id        as qb_id,
        passer_player_name      as qb_name,
        passing_yards,
        pass_attempt,
        complete_pass,
        epa,
        cpoe,
        success,
        week
    from source
    where play_type in ('pass', 'run')
      and down is not null

)

select * from renamed

# models/staging/_sources.yml
version: 2

sources:
  - name: raw
    schema: raw
    tables:
      - name: plays
        description: "Raw NFL play-by-play from nfl_data_py"

-- models/intermediate/int_qb_game_stats.sql

with plays as (

    select * from {{ ref('stg_plays') }}
    where pass_attempt = 1
      and qb_id is not null

),

qb_game as (

    select
        qb_id,
        qb_name,
        game_id,
        week,
        possession_team,
        count(*)                    as total_attempts,
        sum(complete_pass)          as completions,
        sum(passing_yards)          as passing_yards,
        avg(epa)                    as avg_epa,
        avg(cpoe)                   as avg_cpoe,
        avg(success::int)           as success_rate
    from plays
    group by 1, 2, 3, 4, 5

)

select * from qb_game

-- models/marts/mart_qb_performance.sql

with game_stats as (

    select * from {{ ref('int_qb_game_stats') }}

),

season_agg as (

    select
        qb_id,
        qb_name,
        count(distinct game_id)                         as games_played,
        sum(total_attempts)                             as attempts,
        sum(completions) * 1.0 / sum(total_attempts)    as completion_pct,
        avg(avg_epa)                                    as season_epa_per_play,
        avg(avg_cpoe)                                   as season_cpoe,
        avg(success_rate)                               as success_rate
    from game_stats
    group by 1, 2
    having sum(total_attempts) >= 100

)

select
    *,
    case
        when season_epa_per_play >= 0.20 then 'Elite'
        when season_epa_per_play >= 0.10 then 'Above Average'
        when season_epa_per_play >= 0.00 then 'Average'
        else 'Below Replacement'
    end as tier
from season_agg
order by season_epa_per_play desc

# models/marts/_marts.yml
version: 2

models:
  - name: mart_qb_performance
    description: "Season-level QB performance with EPA/CPOE tiers"
    columns:
      - name: qb_id
        tests:
          - unique
          - not_null
      - name: attempts
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
      - name: completion_pct
        tests:
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 1

dbt deps   # one-time, installs dbt_utils
dbt run
dbt test

-- models/marts/mart_qb_rolling_form.sql

with game_stats as (

    select * from {{ ref('int_qb_game_stats') }}

)

select
    qb_id,
    qb_name,
    game_id,
    week,
    avg_epa,
    avg(avg_epa) over (
        partition by qb_id
        order by week
        rows between 4 preceding and current row
    ) as rolling_5_game_epa,
    avg(avg_cpoe) over (
        partition by qb_id
        order by week
        rows between 4 preceding and current row
    ) as rolling_5_game_cpoe
from game_stats

# ai/tactical_analyst.py
import os
import duckdb
from openai import OpenAI

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
con = duckdb.connect("nfl_analytics.duckdb")

def get_qb_context(qb_name: str) -> str:
    """Pull rolling form + season tier from dbt marts."""
    rolling = con.execute("""
        select week, rolling_5_game_epa, rolling_5_game_cpoe
        from analytics.mart_qb_rolling_form
        where qb_name = ?
        order by week desc
        limit 1
    """, [qb_name]).fetchone()

    season = con.execute("""
        select season_epa_per_play, season_cpoe, tier
        from analytics.mart_qb_performance
        where qb_name = ?
    """, [qb_name]).fetchone()

    if not rolling or not season:
        return f"No data found for {qb_name}."

    return (
        f"{qb_name} — Tier: {season[2]}. "
        f"Season EPA/play: {season[0]:.3f}, CPOE: {season[1]:.2f}. "
        f"Last 5-game rolling EPA: {rolling[1]:.3f}, "
        f"rolling CPOE: {rolling[2]:.2f}."
    )

def tactical_call(qb_name: str, down: int, yards_to_go: int, yardline: int) -> str:
    context = get_qb_context(qb_name)

    prompt = f"""
You are an NFL tactical analyst with access to real play-by-play data.

Context from dbt marts:
{context}

Situation:
- Down: {down}
- Yards to go: {yards_to_go}
- Yardline (distance to opponent end zone): {yardline}

Should the team go for it or punt/kick?
Justify using EPA and rolling form. Give a probability of success
based on historical patterns at this down/distance.
"""

    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
    )
    return resp.choices[0].message.content

# 4th and 2, opponent 40, with Mahomes hot
print(tactical_call("P.Mahomes", down=4, yards_to_go=2, yardline=40))

# run_pipeline.py
import subprocess
import sys

def run_step(name, cmd):
    print(f"\n▶ {name}")
    r = subprocess.run(cmd, shell=True)
    if r.returncode != 0:
        print(f"✖ {name} failed")
        sys.exit(1)
    print(f"✓ {name} ok")

if __name__ == "__main__":
    run_step("0. Install dbt packages", "dbt deps")
    run_step("1. Ingest NFL data",      "python ingest_nfl.py")
    run_step("2. dbt run",              "dbt run")
    run_step("3. dbt test",             "dbt test")
    run_step("4. Tactical analyst",     "python ai/tactical_analyst.py")