---

name: airflow-dag-patterns description: Best practices and conventions for creating and maintaining Airflow DAGs, including task naming, flow patterns, and error handling. version: 1.0.0

Airflow DAG Patterns

Main DAG Location

dags/multi_sport_betting_workflow.py - Unified workflow for all sports.

Task Naming Convention

Format: {action}_{sport}

Examples:

• download_games_nba
• update_elo_nhl
• identify_bets_mlb
• place_bets_nfl

Sports Configuration

Add/modify sports in SPORTS_CONFIG:

CopySPORTS_CONFIG = {
    "nba": {
        "elo_module": "nba_elo_rating",
        "games_module": "nba_games",
        "kalshi_function": "fetch_nba_markets",
        "elo_threshold": 0.73,
    },
    # Add new sport here following same pattern
}

Task Flow Pattern

Copydownload_games → load_to_db → update_elo → fetch_markets → identify_bets → place_bets

Each sport runs independently in parallel.

Critical Rules

1. NEVER trigger manual DAG runs - Clear tasks and let Airflow schedule

2. Use PythonOperator for all tasks:

   Copyfrom airflow.providers.standard.operators.python import PythonOperator
   
   task = PythonOperator(
       task_id="update_elo_nba",
       python_callable=update_elo_ratings,
       op_kwargs={"sport": "nba"},
   )
   

3. Handle failures gracefully - Don't let one sport block others:

   Copytask = PythonOperator(
       task_id="download_games_nba",
       python_callable=download_games,
       retries=3,
       retry_delay=timedelta(minutes=5),
   )
   

Adding a New Task

1. Define function in DAG file or import from plugins
2. Create PythonOperator with task_id following naming convention
3. Set dependencies with >> operator
4. Test DAG parsing: python dags/multi_sport_betting_workflow.py

Common Imports

Copyfrom airflow import DAG
from airflow.providers.standard.operators.python import PythonOperator
from datetime import datetime, timedelta

Default Args Pattern

Copydefault_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "email_on_failure": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    "multi_sport_betting_workflow",
    default_args=default_args,
    schedule="0 10 * * *",  # 10 AM daily
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:
    ...

Files to Reference

• dags/multi_sport_betting_workflow.py - Main DAG
• dags/portfolio_hourly_snapshot.py - Simpler DAG example