# BEFORE: notebook cell
# Cell 5: Clean the data
df = pd.read_csv('sales.csv')
df = df.dropna(subset=['revenue'])
df['date'] = pd.to_datetime(df['date'])
df = df[df['revenue'] > 0]
df['month'] = df['date'].dt.to_period('M')

# AFTER: extracted function
def clean_sales_data(raw_df: pd.DataFrame) -> pd.DataFrame:
    """Remove nulls, parse dates, filter positive revenue."""
    df = raw_df.dropna(subset=['revenue']).copy()
    df['date'] = pd.to_datetime(df['date'])
    df = df[df['revenue'] > 0]
    df['month'] = df['date'].dt.to_period('M')
    return df

cells 1-3   (load data)       → src/ingestion.py
cells 4-8   (clean/transform) → src/preprocessing.py
cells 9-12  (feature eng)     → src/features.py
cells 13-16 (model train)     → src/training.py
cells 17-20 (evaluate)        → src/evaluation.py
cells 21-23 (report/plot)     → src/reporting.py

# src/pipeline.py
from src.ingestion import load_sales_data
from src.preprocessing import clean_sales_data
from src.features import build_features
from src.training import train_model
from src.evaluation import evaluate_model
from src.reporting import generate_report

def run_pipeline(config: dict) -> dict:
    raw = load_sales_data(config['data_path'])
    clean = clean_sales_data(raw)
    features = build_features(clean, config['feature_params'])
    model = train_model(features, config['model_params'])
    metrics = evaluate_model(model, features)
    report = generate_report(metrics, config['output_dir'])
    return {'model': model, 'metrics': metrics, 'report': report}

CLI arguments  →  Environment variables  →  config.yaml  →  defaults in code

# config.yaml
data:
  input_path: "s3://bucket/sales/"
  output_path: "s3://bucket/results/"

preprocessing:
  min_revenue: 0
  date_column: "date"

features:
  window_sizes: [7, 14, 30]
  categorical_columns: ["region", "product_type"]

model:
  algorithm: "xgboost"
  hyperparameters:
    max_depth: 6
    learning_rate: 0.1
    n_estimators: 500

random_seed: 42

# src/config.py
from pathlib import Path
import yaml

def load_config(config_path: str = "config.yaml", overrides: dict | None = None) -> dict:
    with open(config_path) as f:
        config = yaml.safe_load(f)
    if overrides:
        _deep_merge(config, overrides)
    return config

def _deep_merge(base: dict, override: dict) -> None:
    for key, value in override.items():
        if isinstance(value, dict) and isinstance(base.get(key), dict):
            _deep_merge(base[key], value)
        else:
            base[key] = value

import os

# Secrets and environment-specific values ONLY
DATABASE_URL = os.environ["DATABASE_URL"]  # fail fast if missing
API_KEY = os.environ["API_KEY"]
ENV = os.environ.get("ENV", "development")  # default for non-secrets

# tests/conftest.py
import pandas as pd
import pytest

@pytest.fixture
def sample_sales_df():
    return pd.DataFrame({
        'date': ['2025-01-01', '2025-01-02', '2025-01-03', None],
        'revenue': [100.0, -50.0, 200.0, 150.0],
        'region': ['US', 'EU', 'US', 'EU'],
    })

@pytest.fixture
def clean_sales_df(sample_sales_df):
    from src.preprocessing import clean_sales_data
    return clean_sales_data(sample_sales_df)

def test_feature_pipeline_output(clean_sales_df, tmp_path):
    result = build_features(clean_sales_df, window_sizes=[7])
    output_path = tmp_path / "features.csv"
    result.to_csv(output_path, index=False)

    golden = Path("tests/golden/features.csv")
    if not golden.exists():
        # First run: create golden file
        result.to_csv(golden, index=False)
        pytest.skip("Golden file created, re-run to validate")

    expected = pd.read_csv(golden)
    pd.testing.assert_frame_equal(result, expected)

def test_clean_data_contract(clean_sales_df):
    """Validate the output schema of the cleaning step."""
    assert set(clean_sales_df.columns) >= {'date', 'revenue', 'month', 'region'}
    assert clean_sales_df['revenue'].min() > 0, "Negative revenue should be filtered"
    assert clean_sales_df['date'].dtype == 'datetime64[ns]'
    assert clean_sales_df['revenue'].isna().sum() == 0, "No null revenue allowed"

# pipelines/daily_sales.py
from dagster import asset, Definitions

@asset
def raw_sales():
    return load_sales_data("s3://bucket/sales/")

@asset
def clean_sales(raw_sales):
    return clean_sales_data(raw_sales)

@asset
def sales_features(clean_sales):
    return build_features(clean_sales, window_sizes=[7, 14, 30])

@asset
def sales_model(sales_features):
    return train_model(sales_features, {"max_depth": 6})

defs = Definitions(assets=[raw_sales, clean_sales, sales_features, sales_model])

# crontab -e
0 6 * * * cd /opt/pipeline && python -m src.pipeline --config config.yaml >> /var/log/pipeline.log 2>&1

# src/__main__.py
import argparse
from src.pipeline import run_pipeline
from src.config import load_config

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--config", default="config.yaml")
    args = parser.parse_args()
    config = load_config(args.config)
    result = run_pipeline(config)
    print(f"Pipeline complete. Metrics: {result['metrics']}")

project/
├── config.yaml
├── pyproject.toml
├── README.md
├── notebooks/           # Exploratory only, not part of prod
│   └── exploration.ipynb
├── src/
│   ├── __init__.py
│   ├── __main__.py      # CLI entry point
│   ├── config.py
│   ├── pipeline.py      # Orchestration
│   ├── ingestion.py
│   ├── preprocessing.py
│   ├── features.py
│   ├── training.py
│   ├── evaluation.py
│   └── reporting.py
├── tests/
│   ├── conftest.py
│   ├── golden/          # Snapshot test data
│   ├── test_preprocessing.py
│   ├── test_features.py
│   ├── test_training.py
│   └── test_pipeline.py
├── pipelines/           # DAG definitions (Airflow/Dagster)
│   └── daily_sales.py
└── data/                # Local dev data only, gitignored
    ├── raw/
    └── processed/