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

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")

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

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

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

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 {{ ref('int_qb_game_stats') }}

import os, 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:
    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))