Ein Stack, der skaliert — und den man versteht

-- dbt-Modell: bereinigte Bestellungen
with source as (
    select * from {{ source('shop', 'orders') }}
),
renamed as (
    select
        order_id,
        customer_id,
        cast(order_ts as date)      as order_date,
        revenue,
        revenue - cost                as margin,
        round((revenue - cost) / nullif(revenue, 0) * 100, 1) as margin_pct
    from source
    where revenue > 0
)
select * from renamed

# Feature-Aufbereitung für das Prognosemodell
import pandas as pd

def build_features(orders: pd.DataFrame) -> pd.DataFrame:
    df = orders.copy()
    df["month"]   = df["order_date"].dt.month
    df["weekday"] = df["order_date"].dt.weekday
    df["rolling"] = (
        df.groupby("customer_id")["revenue"]
          .transform(lambda s: s.rolling(3, min_periods=1).mean())
    )
    return df.dropna()

# Airflow-DAG: täglicher Lauf
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

with DAG(
    "datalab_daily",
    schedule="0 5 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:
    ingest    = PythonOperator(task_id="ingest",    python_callable=load_sources)
    transform = PythonOperator(task_id="dbt_run",   python_callable=run_dbt)
    forecast  = PythonOperator(task_id="forecast",  python_callable=score_models)
    ingest >> transform >> forecast