39.2 Feature-pipeline, baseline és tanulás

A Probléma és adatszerződés befagyasztása alfejezetben befagyasztottuk a Northstar predikciós szerződést: a megfigyelési egység a rendelés, a predikció pillanata az order_ts, a célváltozó a target_late (1, ha delivery_days > promised_days = 7). Ebben az alfejezetben ellenőrzött joinnal feature-táblát építünk, két modelljelöltet futtatunk rögzített költségkerettel, és a tanulást úgy szervezzük, hogy a preprocessing minden CV-fold tanítórészén belül tanuljon. Az itt közölt kód nem futtatott ebben a generálási módban; a számszerű kimeneteket „szemléltető” jelöléssel kezeljük. A tényleges mérés a te gépeden, a uv run paranccsal történik.

39.2.1 Cél és konkrét helyzet

Bemenet: lab/data/orders.csv (2400 sor) és lab/data/customers.csv (300 sor). A predikciós idő a rendelés időpontja; a modell csak az ekkor ismert mezőket kaphatja. A Probléma és adatszerződés befagyasztása alfejezetben rögzített feature-lista:

distance_km, priority_flag, basket_minor, quantity_sum, account_age_days,
segment, region, acquisition_channel

A célváltozó target_late; a delivery_days, actual_delivery_at és label_available_at kizárt predictor. A felosztás: tanító (order_ts < cutoff) ÉS (label_available_at < cutoff); teszt order_ts >= cutoff; purged határsáv a többi. Az alap cutoff 2025-05-01T00:00:00+00:00.

Előzetes becslés: ha a késési arány a tanítóban körülbelül 25%, akkor a mindig negatív dummy pontossága körülbelül 0.75, log lossa

$$-(0.25\ln 0.25 + 0.75\ln 0.75)\approx 0.562.$$

Egy hasznos modelltől a teszt ROC AUC legalább 0.60–0.65 körüli értéket várunk; a tényleges szám a futásból derül ki.

39.2.2 Séma és ellenőrzött join

A séma (schema) az oszlopnév, dtype, tartomány és egység együttes rögzítése. A betöltés után ellenőrizzük:

A customers és orders összekapcsolása bal oldali join, validate="many_to_one" kapcsolóval. A pandas ezzel a jobb oldali customer_id-k egyediségét ellenőrzi. Nem létező ügyfélkulcs esetén a bal oldali join alapból nem ad hibát; ezt a csatlakozás eredményén külön, például indicator=True-val vagy az itt alkalmazott signup_date.notna() ellenőrzéssel kell kimutatni. A join után a sorok száma változatlanul 2400 kell legyen. Ugyanez a join DuckDB SQL-lel is elvégezhető, de a pandas validate paramétere a helyi kontroll-laborban közvetlen kardinalitás-ellenőrzést ad.

39.2.3 Point-in-time: account_age_days és label availability

A point-in-time elv azt mondja ki, hogy minden feature értékét abban az időpontban kell ismerni, amikor a predikciót valójában futtatnánk. Az account_age_days ezt teljesíti:

$$a_i = \frac{t_i^{\text{order}} - t_i^{\text{signup}}}{\text{1 nap}}$$

ahol $a_i$ egész nap, $t_i^{\text{order}}$ az $i$. rendelés UTC-időbélyege, $t_i^{\text{signup}}$ az ügyfél regisztrációs dátuma UTC szerint. A kivonás pandas Timedelta objektumot ad, a .dt.days egész napot. A szerződés garantálja, hogy a regisztráció minden rendelés előtt van, ezért $a_i \ge 0$.

A label availability a tanítócímke elérhetőségét jelenti: a label_available_at = actual_delivery_at, tehát a címke csak a teljes kézbesítéskor válik ismertté. A tanítóba csak azok a rendelések kerülnek, ahol label_available_at < cutoff. A purged sáv a cutoff előtti rendelések közül azok, ahol a címke még nem érkezett meg; ezeket kihagyjuk, mert a predikciós időpontban a valós rendszer sem ismerné a kimenetüket.

39.2.4 Pipeline és foldon belüli preprocessing

A Pipeline a scikit-learn-ben a preprocessing lépéseket és a modellt egyetlen becslővé fűzi. A ColumnTransformer numerikus oszlopokra SimpleImputer + StandardScaler, kategorikus oszlopokra SimpleImputer + OneHotEncoder láncot alkalmaz. A kulcspont: ha a keresztvalidációt a teljes pipeline-on futtatjuk, a scaler átlaga, a median imputált értéke és a one-hot kódolás kategóriakészlete minden fold tanítórészéből tanul, és a fold validációs részére alkalmazzuk. Ha ezt előre, a teljes tanítóhalmazon tennénk meg, a validációs fold statisztikája beszivárogna a transzformációba.

A logisztikus regresszió predikciója:

$$p_i = \sigma(\mathbf{x}_i^\top \mathbf{w} + b), \quad \sigma(z)=\frac{1}{1+e^{-z}}$$

ahol $\mathbf{x}_i$ a transzformált feature-vektor, shape $(p,)$; $\mathbf{w}$ a súlyvektor, shape $(p,)$; $b$ skalár; $p_i \in (0,1)$ a késés becsült valószínűsége. A HistGradientBoostingClassifier nem ezt a lineáris képletet használja, hanem döntési fák összegét; mindkét modell predict_proba-ja valószínűségi pontszámot ad. A logisztikus döntési határ a standardizált feature-térben lineáris, szemben az Embedding és hasonlósági keresés alfejezetben tárgyalt koszinusz-hasonlósággal, amely normalizált vektorok szögét méri.

Az AUC értelmezése:

$$\mathrm{AUC} = P(s(X_+) > s(X_-)),$$

ahol $s$ a modell pontszáma, $X_+$ egy pozitív, $X_-$ egy negatív megfigyelés. Az AUC csak a sorrendtől függ, a kalibrációtól nem; ezért a log loss-t is mérjük.

39.2.5 Baseline és controlled search

A baseline egy olyan egyszerű szabály, amelyhez a jelölteket mérjük. A DummyClassifier(strategy="prior") a tanító pozitív arányát adja meg a valószínűségkimenetben; predict() a többségi osztályt választja. A tesztpontosság ezért a teszt többségi osztályának tényleges részarányától függ, a log loss-hoz pedig a tanítóban becsült $q$ arány kerül. A log loss a dummy $p_i=q$ konstans esetén:

$$L = -\frac{1}{n}\sum_{i=1}^{n}\left[y_i\ln p_i + (1-y_i)\ln(1-p_i)\right],$$

ahol $q$ a pozitív osztály aránya, $y_i\in\{0,1\}$ a tényleges címke. A modellnek ennél kisebb log loss-t kell elérnie.

A controlled search (ellenőrzött keresés) rögzített fit-költségkerettel dolgozik. A példában a logisztikus grid $3\times2=6$ kombináció, a HGB grid $2\times2=4$ kombináció; 3-fold TimeSeriesSplit esetén a CV fitek $6\cdot3+1=19$ és $4\cdot3+1=13$, összesen 32 fit. A keret 50 fit, tehát egy további kisebb jelölt is belefér. A scoring="roc_auc" biztosítja, hogy a belső választás a célmetrikát optimalizálja; a tesztet a keresés nem érinti.

39.2.6 Tracking

A tracking a futás bemenetének, paramétereinek és kimenetének tárolása. A lab minimális JSON manifestet ír; MLflow esetén ugyanez mlflow.log_params, mlflow.log_metric és mlflow.sklearn.log_model hívásokkal történne. A manifest tartalmazza a cutoffot, a split darabszámokat, a feature-listát, a jelöltek legjobb paramétereit, a CV és teszt metrikákat, valamint a modell- és adatfájl SHA-256 hashét. A hash lehetővé teszi, hogy egy későbbi betöltésnél észleljük, ha az artifact megváltozott.

39.2.7 Teljes minimális munkapélda

A példa a lab/data útvonalat feltételezi; ha a CSV-k a lab gyökerében vannak, cseréld a DATA_DIR értékét Path(".")-ra. Helyezd a scriptet a lab/train_model.py fájlba, és futtasd a lab mappa gyökeréből:

cd lab
uv run python train_model.py --cutoff 2025-05-01T00:00:00+00:00 --artifact-dir artifacts/s154

PowerShell-ben ugyanez a parancs a cd lab után; a munkakönyvtár a script helye.

import argparse
import hashlib
import json
from pathlib import Path

import joblib
import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.dummy import DummyClassifier
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import average_precision_score, log_loss, roc_auc_score
from sklearn.model_selection import GridSearchCV, TimeSeriesSplit
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

DATA_DIR = Path("data")
FEATURES = [
    "distance_km", "priority_flag", "basket_minor", "quantity_sum",
    "account_age_days", "segment", "region", "acquisition_channel",
]
TARGET = "target_late"
NUM_FEATURES = ["distance_km", "priority_flag", "basket_minor", "quantity_sum", "account_age_days"]
CAT_FEATURES = ["segment", "region", "acquisition_channel"]


def file_hash(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--cutoff", default="2025-05-01T00:00:00+00:00")
    parser.add_argument("--artifact-dir", default="artifacts/s154")
    args = parser.parse_args()

    cutoff = pd.Timestamp(args.cutoff)
    artifact_dir = Path(args.artifact_dir)
    artifact_dir.mkdir(parents=True, exist_ok=True)

    orders = pd.read_csv(
        DATA_DIR / "orders.csv",
        parse_dates=["order_ts", "actual_delivery_at", "label_available_at"],
    )
    customers = pd.read_csv(DATA_DIR / "customers.csv", parse_dates=["signup_date"])

    # Séma-ellenőrzés a tiszta adaton
    assert orders["basket_minor"].isna().sum() == 0
    assert (orders["basket_minor"] > 0).all()
    assert (orders["distance_km"] >= 0).all()
    assert orders["priority_flag"].isin([0, 1]).all()
    assert orders["quantity_sum"].ge(1).all()
    assert orders["currency"].eq("EUR").all()

    # Ellenőrzött join: egy rendeléshez pontosan egy ügyfél
    df = orders.merge(
        customers[["customer_id", "signup_date", "segment", "region", "acquisition_channel"]],
        on="customer_id",
        how="left",
        validate="many_to_one",
    )
    assert df["signup_date"].notna().all()
    assert len(df) == len(orders)

    # Point-in-time feature
    df["signup_ts"] = pd.to_datetime(df["signup_date"]).dt.tz_localize("UTC")
    df["account_age_days"] = (df["order_ts"] - df["signup_ts"]).dt.days
    assert (df["account_age_days"] >= 0).all()

    # Split a s153 szerződés szerint
    df["split"] = "purged"
    df.loc[(df["order_ts"] < cutoff) & (df["label_available_at"] < cutoff), "split"] = "train"
    df.loc[df["order_ts"] >= cutoff, "split"] = "test"

    assert (df.loc[df["split"] == "train", "order_ts"] < cutoff).all()
    assert (df.loc[df["split"] == "train", "label_available_at"] < cutoff).all()
    assert (df.loc[df["split"] == "test", "order_ts"] >= cutoff).all()

    train = df[df["split"] == "train"].copy()
    test = df[df["split"] == "test"].copy()
    split_counts = df["split"].value_counts().to_dict()

    X_train, y_train = train[FEATURES], train[TARGET]
    X_test, y_test = test[FEATURES], test[TARGET]

    # Kizárt mezők ellenőrzése
    forbidden = {"delivery_days", "actual_delivery_at", "label_available_at", "order_id", "customer_id"}
    assert forbidden.isdisjoint(FEATURES)

    numeric_pipe = Pipeline([
        ("imputer", SimpleImputer(strategy="median")),
        ("scaler", StandardScaler()),
    ])
    categorical_pipe = Pipeline([
        ("imputer", SimpleImputer(strategy="most_frequent")),
        ("onehot", OneHotEncoder(handle_unknown="ignore", sparse_output=False)),
    ])
    preprocessor = ColumnTransformer([
        ("num", numeric_pipe, NUM_FEATURES),
        ("cat", categorical_pipe, CAT_FEATURES),
    ])

    # Két jelölt; a preprocessing a pipeline-on belül, foldonként tanul
    candidates = {
        "logreg": Pipeline([
            ("prep", preprocessor),
            ("model", LogisticRegression(max_iter=1000, solver="lbfgs")),
        ]),
        "hgb": Pipeline([
            ("prep", preprocessor),
            ("model", HistGradientBoostingClassifier(max_iter=100, early_stopping=True, random_state=42)),
        ]),
    }
    param_grid = {
        "logreg": {
            "model__C": [0.1, 1.0, 10.0],
            "model__class_weight": [None, "balanced"],
        },
        "hgb": {
            "model__learning_rate": [0.05, 0.1],
            "model__max_iter": [50, 100],
        },
    }

    cv = TimeSeriesSplit(n_splits=3)
    searches = {}
    budget = {}
    for name, model in candidates.items():
        search = GridSearchCV(
            model,
            param_grid[name],
            cv=cv,
            scoring="roc_auc",
            refit=True,
            n_jobs=2,
            error_score="raise",
        )
        search.fit(X_train, y_train)
        searches[name] = search
        # A refit +1 fit; a CV fitek: kombinációk * foldok
        budget[name] = len(search.cv_results_["params"]) * cv.get_n_splits() + 1

    # Dummy baseline
    dummy = DummyClassifier(strategy="prior")
    dummy.fit(X_train, y_train)

    def evaluate(name, model):
        p = model.predict_proba(X_test)[:, 1]
        return {
            "model": name,
            "auc": roc_auc_score(y_test, p) if y_test.nunique() == 2 else np.nan,
            "average_precision": average_precision_score(y_test, p),
            "log_loss": log_loss(y_test, p, labels=[0, 1]),
        }

    rows = [evaluate("dummy", dummy)]
    rows += [
        {
            "model": name,
            "auc": evaluate(name, search.best_estimator_)["auc"],
            "average_precision": evaluate(name, search.best_estimator_)["average_precision"],
            "log_loss": evaluate(name, search.best_estimator_)["log_loss"],
            "cv_mean_auc": search.best_score_,
            "best_params": search.best_params_,
        }
        for name, search in searches.items()
    ]
    metrics = pd.DataFrame(rows)
    print(metrics.to_string(index=False))

    # A CV szerinti legjobb jelölt mentése; a teszt AUC utólagos ellenőrzés
    best_name = max(searches, key=lambda name: searches[name].best_score_)
    best = searches[best_name].best_estimator_
    model_path = artifact_dir / "model.joblib"
    joblib.dump(best, model_path)

    # Betöltési integritás
    loaded = joblib.load(model_path)
    np.testing.assert_allclose(
        loaded.predict_proba(X_test.iloc[:10]),
        best.predict_proba(X_test.iloc[:10]),
        rtol=1e-6,
    )

    manifest = {
        "run_id": "s154-local-demo",
        "cutoff": cutoff.isoformat(),
        "split_counts": split_counts,
        "features": FEATURES,
        "target": TARGET,
        "fit_budget": budget,
        "selected_candidate": best_name,
        "candidates": {
            name: {
                "best_params": search.best_params_,
                "cv_mean_auc": search.best_score_,
                "test_auc": evaluate(name, search.best_estimator_)["auc"],
                "test_log_loss": evaluate(name, search.best_estimator_)["log_loss"],
            }
            for name, search in searches.items()
        },
        "model_sha256": file_hash(model_path),
        "orders_csv_sha256": file_hash(DATA_DIR / "orders.csv"),
        "customers_csv_sha256": file_hash(DATA_DIR / "customers.csv"),
    }
    manifest_path = artifact_dir / "manifest.json"
    manifest_path.write_text(json.dumps(manifest, indent=2))
    print(f"Wrote {model_path} and {manifest_path}")
    print(f"Split counts: {split_counts}")


if __name__ == "__main__":
    main()

Az alábbi kimenet szemléltető, nem rögzített futási jegyzékből származik; a te számaid eltérhetnek:

Split counts: {'train': 1512, 'test': 708, 'purged': 180}
model  auc    average_precision  log_loss  cv_mean_auc  best_params
dummy  0.500  0.234               0.612     NaN          NaN
logreg 0.619  0.351               0.589     0.608        {'model__C': 0.1, 'model__class_weight': None}
hgb    0.655  0.372               0.581     0.641        {'model__learning_rate': 0.1, 'model__max_iter': 100}

39.2.8 Eredmény értelmezése és ellenőrző invariánsok

A kézi ellenőrzés négy invariant köré épül:

1. len(df) == len(orders), tehát a join nem szaporította a sorokat.
2. train + test + purged = len(df), és a train minden sorára label_available_at < cutoff.
3. forbidden.isdisjoint(FEATURES), tehát a címke és a későbbi időbélyegek nem szerepelnek predictorban.
4. A mentett modell predict_proba-ja megegyezik a memóriában lévő modellével 1e-6 relatív tolerancián belül.

Az AUC 0.5 a véletlen szint; a dummy AUC pontosan 0.5, ha mindkét osztály jelen van a tesztben. A 0.60 körüli AUC szerény, a 0.65 már hasznos jelzés lehet, de a döntést a log loss és a gyakorlati küszöb alapján kell meghozni. A teszt AUC egyetlen splitből származó pontbecslés; a bizonytalanság becsléséhez bootstrap vagy nested CV kellene, amit a Keresztvalidáció a függőségi szerkezethez és Hiperparaméter és nested értékelés tárgyal. A CV-foldok szórása nem konfidenciaintervallum.

A fit-költség a manifest fit_budget mezőjében 32, a tervezett kereten belül. A HGB early stopping a falidőt csökkentheti, de a számlált fit-ek számát nem változtatja. A méréshez rögzítsd a gépet, a sorok számát és a futásidőt; e példában a 2400 soros Northstar adat a CPU, GPU és elosztott rendszer közötti döntés döntési lap szerint CPU-s, kis munkaterhelés, de a tényleges időt a te gépeden kell megmérned. A FastAPI végpont és ellenőrzött bemenet alfejezetben tanult predikciós szerződés itt batch CLI módban, a --cutoff és --artifact-dir argumentumokkal fut.

39.2.9 Feladatok

1. Értelmezés. Miért nem kerülhetnek a purged sorok sem a train, sem a test halmazba? Fogalmazd meg a predikció pillanata és a label availability szempontjából.
2. Megvalósítás. Bővítsd a scriptet egy RandomForestClassifier jelölttel n_estimators=[100, 200] és max_depth=[None, 5] rácson úgy, hogy a teljes fit-költség ne lépje túl az 50 fitet. Add meg a módosított kódrészletet.
3. Hibadiagnózis. A futtatás után a CV mean AUC 0.85, a teszt AUC 0.61. Sorold fel a két legvalószínűbb okot, és add meg az első ellenőrző lépést.
4. Megváltozott feltétel. Állítsd a cutoffot 2025-04-15T00:00:00+00:00 értékre, és futtasd újra. Mit vársz a train, test és purged halmazok arányától, és miért nőhet a modell varianciája?

39.2.10 Megoldások és indoklás

1. A purged sorok rendelése a cutoff előtti, tehát a teszt definíciója (order_ts >= cutoff) nem teljesül; a tesztbe téve a múltbeli rendeléseket kevernénk a jövőbeli alkalmazással. A trainbe sem tehetők, mert a label_available_at >= cutoff; a címkéjük a tanítás pillanatában a valós rendszerben nem lenne ismert. A modell így jövőbeli kimenetet tanulna, ami leakage. A helyes kezelés a kihagyás; az ellenőrzés a split_counts és a train feltétel assert-je.

2. A harmadik jelölt:

from sklearn.ensemble import RandomForestClassifier

candidates["rf"] = Pipeline([
    ("prep", preprocessor),
    ("model", RandomForestClassifier(random_state=42, n_jobs=1)),
])
param_grid["rf"] = {
    "model__n_estimators": [100, 200],
    "model__max_depth": [None, 5],
}

A fit-ek: $4\cdot3+1=13$; a meglévő 32-vel együtt 45 < 50. A n_jobs=1 azért fontos, mert a külső GridSearchCV(n_jobs=2) már párhuzamosít; a belső szálak túlfoglalnák a CPU-t. A RandomForestClassifier a predict_proba-t osztályarányból becsli, ezért az AUC és log loss ugyanúgy számítható.

3. A két leggyakoribb ok: (a) leakage, például egy későbbi időbélyeg vagy a label_available_at véletlenül a FEATURES-ben; (b) túlillesztés, amikor a modell a tanító foldokat memorizálja. Első lépés: futtasd le a forbidden.isdisjoint(FEATURES) ellenőrzést, és nézd meg a HGB max_iter/learning_rate értékeit. Ha a teszt AUC 0.61, a dummy 0.5-höz képest még van jel, de a 0.85-ös CV irreálisan optimista; csökkentsd a modell bonyolultságát, és ne a teszt alapján hangolj.

4. A korábbi cutoff kisebb train halmazt és nagyobb test halmazt eredményez. A purged sáv aránya a trainhez képest nőhet, mert a cutoff előtti rendelések közül többnek a teljes kézbesítése a cutoff után érkezik. A kevesebb tanító sor miatt a modell varianciája nő, a teszt AUC becslése bizonytalanabb. A manifest új run_id-t kap; a két cutoff eredménye nem független, mert a tesztablakok átfednek.

39.2.11 Gyakori tévedés ellenpéldával

Az időrend önmagában nem zárja ki a még meg nem érkezett tanítócímkéből eredő leakage-t. Tegyük fel, hogy a tanítót csak az order_ts < cutoff feltétellel definiáljuk, és a label_available_at szűrést elhagyjuk. Egy 2025-04-30-i rendelés, amelynek delivery_days=5, azaz a címkéje 2025-05-05-én válik ismertté, bekerül a tanítóba, pedig a predikció pillanatában (2025-04-30) a kimenete még nem volt ismert. A modell így a cutoff utáni kézbesítési információt használja, ami a teszt AUC-t mesterségesen felfelé torzítja. A helyes feltétel (order_ts < cutoff) & (label_available_at < cutoff), és ezt az assert őrzi.

39.2.12 Önellenőrzés

• A train minden sorára order_ts < cutoff és label_available_at < cutoff teljesül.
• A FEATURES nem tartalmazza a delivery_days, actual_delivery_at, label_available_at, order_id, customer_id mezőket.
• A GridSearchCV a teljes pipeline-on fut, nem előre transzformált adaton.
• A tesztet egyszer sem használtuk paraméter- vagy küszöbválasztásra.
• A manifest tartalmazza a split counts-ot, a paramétereket, a metrikákat és az artifact hash-eket.

39.2.13 Rövid összefoglaló

A Feature-pipeline, baseline és tanulás a Northstar feature-pipeline-ját, baseline-ját és tanulását építi fel a befagyasztott predikciós szerződésre. Az ellenőrzött many_to_one join után a point-in-time account_age_days feature készül, a train csak a cutoffnál már elérhető címkéjű sorokat kapja. A ColumnTransformer és a modell egy pipeline-ban van, így a scaler, imputer és one-hot minden TimeSeriesSplit fold tanítórészén tanul. Két jelölt, logisztikus regresszió és HGB, összesen 32 fit költségkerettel fut; a dummy baseline és a ROC AUC/log loss értékelés után a manifest JSON és a joblib artifact rögzíti a futást. A következő lépés Független értékelés és hibaelemzés alfejezetben a független értékelés és hibaelemzés.