29.1 Modulhatár és tiszta függvény

A refaktorálás nem a notebook eldobása, hanem a benne felhalmozott tudás lefordítása olyan egységekre, amelyeket parancssorból és tesztből is meg tudunk hívni. Ebben az alfejezetben a Northstar-rendelések egyik időszerű feature-ét, az account_age_days értéket emeljük ki a notebookból egy src/northstar elrendezésű csomagba. A cél, hogy ugyanaz a matematikai transzformáció fusson a notebookban, a CLI-ben és a pytest alatt, miközben a fájlolvasás és -írás a modul szélén marad.

29.1.1 A célhelyzet és a refaktorálás szerződése

A helyzet: egy notebook cellája jelenleg nagyjából ezt csinálja.

# notebookbeli, állapotérzékeny változat vázlata
df = pd.read_csv("data/orders.csv", parse_dates=["order_ts"])
cust = pd.read_csv("data/customers.csv", parse_dates=["signup_date"])
joined = df.merge(cust[["customer_id", "signup_date", "segment", "region",
                        "acquisition_channel"]],
                  on="customer_id", how="left")
joined["account_age_days"] = (joined["order_ts"] - joined["signup_date"]).dt.days

A bemenetek: orders.csv rendelésenként, customers.csv ügyfelenként. A megfigyelési egység a rendelés; a cél egy rendelésenkénti numerikus oszlop, amely a rendelés pillanatában ismert ügyfélkort adja napokban. A kimeneti kontroll: a kézzel számolható példasornál a képletnek ugyanazt kell adnia, mint a notebookbeli számításnak, és az oszlop int64 dtype-ú, nemnegatív, a sorok száma megegyezik az orders.csv sorainak számával. A refaktorálás matematikai hídja az alábbi transzformáció:

$$a(t_s, t_o)=\left\lfloor\frac{t_o-t_s}{1\ \text{nap}}\right\rfloor,$$

ahol $t_s$ a signup_date, $t_o$ az order_ts; a kivonás időbélyeg-különbség, a floor pedig teljes napokra csonkít. A bemeneti sorozatok egyaránt $n$ hosszúak, a kimenet is $n$ hosszú egész napokból áll. Az adatszerződés szerint $t_s \le t_o$, ezért elvárjuk, hogy $a \ge 0$. A példában szereplő datetime-ok egyeznek a Northstar-szerződés UTC-kezelésével, így nincs rejtett időzóna-konverzió.

29.1.2 Fogalmi magyarázat: tiszta függvény, I/O-határ, src layout, import és CLI

Egy tiszta függvény (pure function) a bemenetéből kiszámít egy kimenetet, és nem módosítja a bemenetet, nem ír fájlt, nem olvas globális állapotot, és nem hív véletlenszám-generátort. Ugyanarra a bemenetre mindig ugyanazt a kimenetet adja. Ez nem csak stílus: a tiszta függvényt közvetlenül lehet tesztelni anélkül, hogy a fájlrendszert vagy egy notebook végrehajtási állapotát utánoznunk kellene.

Az I/O-határ (I/O boundary) a fájl, hálózat, adatbázis, óra és véletlen számára fenntartott réteg. A megoldásban a load_orders, load_customers és save_features tartoznak ide; ezek a dtype- és sémaszerződés tulajdonosai. A transzformációk csak a már betöltött DataFrame-ekkel dolgoznak. Így a pure függvény akkor is tesztelhető, ha az adatfájlt nem adjuk át neki.

Az src layout a Python-projekt javasolt szerkezete: a csomag forráskódja a src/northstar mappában van, a tesztek a tests/ mappában, a futtatási kimenetek az artifacts/ mappában. A pyproject.toml [tool.setuptools.packages.find] where = ["src"] beállítása mondja meg a csomagolónak, hogy a src alatt keresse a csomagot. Ez elkülöníti a forrást az adattól és a jegyzetfüzettől; a notebook a telepített vagy elérhető csomagot importálja.

lab/
├── data/
│   ├── orders.csv
│   └── customers.csv
├── artifacts/
├── src/
│   └── northstar/
│       ├── __init__.py
│       ├── features.py
│       ├── io.py
│       └── cli.py
├── tests/
│   └── test_features.py
└── pyproject.toml

Az import a modulhatár nyelve. A __init__.py határozza meg a publikus API-t; a notebook csak ezt használja:

from northstar.features import compute_account_age_days, build_order_features
from northstar.io import load_orders, load_customers, save_features

A CLI (command-line interface) az argparse-szal épített belépési pont. A notebook interaktív kutatáshoz jó; a CLI reprodukálható, naplózott, tesztelhető futtatást ad. A CLI nem számol közvetlenül: betölt, meghívja a tiszta transzformációkat, kiír, és a logging modulon keresztül beszámol. A típusellenőrzés itt elsősorban a pd.DataFrame és Path típushintjein keresztül segíti az olvasót; a futásidejű dtype-ellenőrzés az I/O-réteg és a pure függvény bemeneti feltétele.

29.1.3 Teljes minimális munkapélda

Először a tiszta feature-függvény. A dt.days a Timedelta.days viselkedését használja: teljes napokra csonkít. Ezt a viselkedést őrzi meg a refaktor, mert a kiinduló notebook is ezt használta.

# src/northstar/features.py
"""A Northstar-rendelések tiszta feature-transzformációi."""

from __future__ import annotations

import pandas as pd


def compute_account_age_days(
    signup_date: pd.Series,
    order_ts: pd.Series,
) -> pd.Series:
    """A rendelés pillanatában ismert ügyfélkort adja teljes napokban.

    Bemenet:
        signup_date: (n,) alakú datetime64 Series, az ügyfél regisztrációs dátuma.
        order_ts: ugyanolyan indexű, (n,) alakú datetime64 Series, a rendelés ideje.
    Kimenet:
        (n,) alakú int64 Series, nemnegatív teljes napok.
    """
    if not signup_date.index.equals(order_ts.index):
        raise ValueError("signup_date and order_ts must share the same index")
    if not (
        pd.api.types.is_datetime64_any_dtype(signup_date)
        and pd.api.types.is_datetime64_any_dtype(order_ts)
    ):
        raise TypeError("signup_date and order_ts must be datetime64 Series")

    result = (order_ts - signup_date).dt.days

    if (result < 0).any():
        raise ValueError(
            "account_age_days cannot be negative; check signup_date <= order_ts"
        )

    return result.astype("int64")


def build_order_features(
    orders: pd.DataFrame,
    customers: pd.DataFrame,
) -> pd.DataFrame:
    """A rendelés szintű predikciós feature-tábla előállítása.

    A join rendelésből ügyfélbe many-to-one; a kimenet nem módosítja a bemeneteket.
    """
    customer_cols = [
        "customer_id",
        "signup_date",
        "segment",
        "region",
        "acquisition_channel",
    ]
    order_cols = [
        "order_id",
        "customer_id",
        "order_ts",
        "distance_km",
        "priority_flag",
        "basket_minor",
        "quantity_sum",
    ]

    features = orders[order_cols].merge(
        customers[customer_cols],
        on="customer_id",
        how="left",
        validate="many_to_one",
    )
    features = features.copy()
    features["account_age_days"] = compute_account_age_days(
        features["signup_date"],
        features["order_ts"],
    )
    return features.drop(columns=["signup_date", "customer_id"]).copy()

A validate="many_to_one" a Join-kardinalitás és duplázódó célváltozó join-kardinalitási tanulsága: nem akarunk rejtett sortöbbszöröződést. A kimenet a Predikciós szerződés és hibateszt predikciós szerződés oszlopait követi, és nem tartalmazza a szivárgó delivery_days értéket.

Az I/O-réteg külön fájlban.

# src/northstar/io.py
"""Fájlműveletek a Northstar adatokhoz; a séma- és dtype-szerződés innen indul."""

from __future__ import annotations

from pathlib import Path

import pandas as pd


def load_orders(path: Path) -> pd.DataFrame:
    return pd.read_csv(path, parse_dates=["order_ts"])


def load_customers(path: Path) -> pd.DataFrame:
    return pd.read_csv(path, parse_dates=["signup_date"])


def save_features(df: pd.DataFrame, path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    df.to_csv(path, index=False)

A CLI az I/O-határon belül kombinálja a lépéseket.

# src/northstar/cli.py
"""CLI a feature-tábla reprodukálható újraépítéséhez."""

from __future__ import annotations

import argparse
import logging
import sys
from pathlib import Path

from .features import build_order_features
from .io import load_customers, load_orders, save_features


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Northstar rendelésszintű feature-tábla építése"
    )
    parser.add_argument("--orders", type=Path, required=True)
    parser.add_argument("--customers", type=Path, required=True)
    parser.add_argument("--out", type=Path, required=True)
    parser.add_argument("--log-level", default="INFO")
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)

    logging.basicConfig(
        level=args.log_level,
        format="%(asctime)s %(levelname)s %(message)s",
    )

    orders = load_orders(args.orders)
    customers = load_customers(args.customers)
    features = build_order_features(orders, customers)
    save_features(features, args.out)
    logging.info("wrote %s rows to %s", len(features), args.out)
    return 0


if __name__ == "__main__":
    sys.exit(main())

A publikus API rövid __init__.py.

# src/northstar/__init__.py
"""Northstar tiszta feature-csomag."""

from .features import build_order_features, compute_account_age_days
from .io import load_customers, load_orders, save_features

__all__ = [
    "build_order_features",
    "compute_account_age_days",
    "load_customers",
    "load_orders",
    "save_features",
]

A teszt nem fájlt gyárt, hanem kis DataFrame-ekkel ellenőrzi a transzformációt.

# tests/test_features.py
import pandas as pd

from northstar.features import compute_account_age_days


def test_account_age_days_matches_manual_calendar():
    signup = pd.to_datetime(
        pd.Series(["2025-01-10", "2025-03-01", "2025-03-02T00:00:00+00:00"])
    )
    order = pd.to_datetime(
        pd.Series(
            [
                "2025-02-10T08:00:00+00:00",
                "2025-03-05T09:30:00+00:00",
                "2025-03-02T00:00:00+00:00",
            ]
        )
    )

    result = compute_account_age_days(signup, order)

    assert result.dtype == "int64"
    assert result.tolist() == [31, 4, 0]

A futtatás a lab projektgyökérből, Linux/macOS shellben:

uv run pytest tests/test_features.py -q

A kimenet szemléltető, mert a szöveg írásakor nem futtattunk: 1 passed in 0.42s. A CLI parancs ugyanonnan:

uv run python -m northstar.cli \
  --orders data/orders.csv \
  --customers data/customers.csv \
  --out artifacts/features.csv \
  --log-level INFO

A features.csv várható sorainak száma 2400, mert az alapgenerált orders.csv ennyi rendelési sort tartalmaz; ezt a manifestből lehet ellenőrizni. A kimeneti naplósor várhatóan ilyesmi:

2026-10-20 12:00:00,000 INFO wrote 2400 rows to artifacts/features.csv

29.1.4 Eredmény értelmezése kézi kontrollal

A kézi kontroll a fenti teszt: 2025-01-10 és 2025-02-10 között 31 teljes nap van, 2025-03-01 és 2025-03-05 között 4, az azonos időpontok között 0. A refaktorálás invariantja: a notebookbeli gyors számítás és a modul kimenete azonos oszlopértékeket ad, tehát az OLS a lineáris algebra és statisztika találkozásánál alfejezetben bemutatott regressziós illesztés ugyanazt a dizájnmátrixot kapná. A shape kontroll: a bemeneti sorozatok (n,), a kimenet (n,); a build_order_features kimenete rendelésszintű, ezért a sorok száma nem nőhet a join során. A dtype kontroll: a kimenet int64, az I/O-réteg pedig datetime64 bemenetet ad a pure függvénynek.

Két további ellenőrzés a modulhatárról: a uv run python -m northstar.cli nem definiál rejtett globális DataFrame-et, és a futtatás közben a features.py nem nyit fájlt. Emiatt a notebook és a CLI ugyanazt a CSV-t kapva ugyanazt az account_age_days oszlopot adja. A Ruff és a típushintek itt a stílus- és olvashatóság-ellenőrzés részei:

uv run ruff check src tests
uv run ruff format src tests

Ezek kimenetét a tanuló helyi futtatása dönti el; itt nem állítunk mért eredményt.

A tiszta függvény fogalma nem egyenlő a mutációmentességgel: a függvény mellékhatásmentes, és kimenete csak argumentumaitól függ. Az időzóna és az üzleti nap számításának szemantikáját külön adatszerződésben rögzítsük.

29.1.5 Feladatok

29.1.5.1 1. Értelmezés: miért nem tiszta a beolvasó függvénybe rejtett számítás?

Egy kolléga így alakítja át a modult: a compute_account_age_days belsejében meghívja a pd.read_csv-t, hogy a notebookban egy sorba kerüljön a művelet. Sorolj fel legalább két hátrányt, és rajzold meg a helyes I/O-határt egy rövid függvényaláírással.

29.1.5.2 2. Megvalósítás: tiszta függvény az átlagos tételárra

Írj egy price_per_item_minor(basket_minor: pd.Series, quantity_sum: pd.Series) -> pd.Series tiszta függvényt a features.py-ba. A függvény a basket_minor / quantity_sum hányadost adja float64 dtype-tal. Ha bármelyik quantity_sum értéke nulla vagy negatív, dobjon ValueError-t, mert ez sérti a Northstar adatszerződést. A bemeneti sorozatok indexe egyezzen meg. Írj egy pytest esetet is, amely egy quantity_sum=0 sort kapva pytest.raises(ValueError)-t vár.

29.1.5.3 3. Hibadiagnózis: a CLI elfut, a notebook nem

A tanuló a CLI-t lefuttatja, és a notebook cellájában ezt a hibát kapja:

AttributeError: Can only use .dt accessor with datetimelike values

A notebook ezt hívja:

orders = pd.read_csv("data/orders.csv")
customers = pd.read_csv("data/customers.csv")
features["account_age_days"] = compute_account_age_days(
    features["signup_date"], features["order_ts"]
)

Mi a hiba gyökere, hol kell javítani, és miért ott? Add meg a javítás utáni notebooksort.

29.1.5.4 4. Megváltozott feltétel: kísérleti ügyfélszintű adat

A csoport most a assignments.csv és outcomes.csv kísérleti fájlokat is szeretné ugyanazzal a CLI-vel feldolgozni. A build_order_features rendelésszintű, a kísérleti kimenet ügyfélszintű. Indokold meg, miért nem szabad egyszerűen beilleszteni egy új mode ágat a meglévő függvénybe, és tervezd meg, milyen publikus függvényaláírás és CLI-bemenet őrizné meg a szemcsézettségbeli szerződést.

29.1.6 Megoldások és indoklás

29.1.6.1 1. megoldás

A beolvasás a fájlrendszerből olvas: mellékhatás, függ a munkakönyvtártól, és lassú. A tiszta függvény teszteléséhez fájlt kellene létrehozni, és a determinisztikus viselkedés nem garantálható, ha a fájl tartalma közben megváltozik. A helyes határ: az I/O-réteg load_orders(path) -> pd.DataFrame, a transzformációs réteg pedig compute_account_age_days(signup_date, order_ts). A kettőt a CLI vagy a notebook külön hívja; ezzel a teszt csak két darab Series-t ad át.

29.1.6.2 2. megoldás

def price_per_item_minor(
    basket_minor: pd.Series,
    quantity_sum: pd.Series,
) -> pd.Series:
    if not basket_minor.index.equals(quantity_sum.index):
        raise ValueError("basket_minor and quantity_sum must share the same index")
    if (quantity_sum <= 0).any():
        raise ValueError("quantity_sum must be positive")
    return (basket_minor / quantity_sum).astype("float64")

A teszt:

def test_price_per_item_rejects_non_positive_quantity():
    basket = pd.Series([1200, 900])
    quantity = pd.Series([3, 0])
    with pytest.raises(ValueError):
        price_per_item_minor(basket, quantity)

A hányados float64, mert a lebegőpontos osztás ezt adja; a basket_minor egész minor unit, a kimenet viszont nem egész, ezért a kerekítés nem alkalmazható.

29.1.6.3 3. megoldás

A hiba gyökere az I/O-réteg hiánya: a notebook pd.read_csv hívása nem használta a parse_dates paramétert, ezért a dátumoszlopok string típusúak maradtak. A .dt accessor csak datetime-jellegű oszlopon működik. A javítás nem a pure függvényben kell, mert annak a bemenete már kötelezően datetime64; a betöltésnek kell a szerződést teljesítenie. A javított notebooksor:

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

Ugyanez a dtype-szerződés van a io.py-ban, ezért a CLI nem hibázott.

29.1.6.4 4. megoldás

A build_order_features a rendelés szemcsézettségére épül: a merge many_to_one validációval rendelési sorokat kap. A kísérleti adatokban egy ügyfélhez egy assignment és egy kimenet tartozik; a rendelésszintű logika nem vihető át automatikusan. Egy külön build_experiment_features(assignments, outcomes) -> pd.DataFrame függvény jobb: ügyfélszintű, validate="one_to_one" join, és a conversion_7d, revenue_7d_minor oszlopokat nem keveri a rendelési tételsorok összegzésével. A CLI-ben külön alparancs, például build-experiment-features, tartja el. Így a meglévő rendelésszintű kimenet változatlan marad, és a szemcsézettségbeli hiba fordítási időben, nem futás közben derül ki.

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

A gyakori tévhit, hogy a notebook-problémát egyetlen nagy FeatureBuilder osztállyal kell helyettesíteni. Az osztály gyakran rejtett belső DataFrame-et tárol:

class FeatureBuilder:
    def __init__(self, data_dir):
        self.df = pd.read_csv(data_dir / "orders.csv")
    def build(self):
        self.df["account_age_days"] = self.df["order_ts"] - self.df["signup_date"]
        return self.df

Ez nem jobb szerkezet, csak máshová rejtette az állapotot. A tesztnek fájlt kell hoznia, a build kétszeri hívása másodszor már egy módosított DataFrame-et adhat, és a notebook állapotát nem szünteti meg. A tiszta függvényes változatban a bemenet és a kimenet a hívásban látható; az ellenőrzés pedig pd.testing.assert_series_equal-lel elvégezhető a notebook régi és a modul új kimenete között.

Önellenőrzés: egy függvény, amely a bemenetét nem módosítja, de minden hívásakor naplófájlba ír, nem tiszta, mert mellékhatása van. A modulhatár tesztje tehát nem csak az, hogy a számítás helyes-e, hanem az is, hogy a transzformáció során nem keletkezik rejtett I/O.

29.1.8 Rövid összefoglaló

A tiszta függvény ugyanahhoz a bemenethez ugyanazt adja, és nincs mellékhatása; a fájlolvasás és -írás az I/O-rétegbe kerül. A src/northstar csomagban a features.py a transzformációt, az io.py a betöltést és mentést, a cli.py az argparse-belépési pontot adja. A notebook már csak a publikus függvényeket hívja, ezért a CLI és a notebook ugyanazt az account_age_days feature-t állítja elő. A kézi naptári kontroll, a shape-, dtype- és indexellenőrzés a refaktorálás invariantja, a pytest pedig fájl nélkül rögzíti a várt viselkedést. A következő alfejezetben ezt a modulhatárt konfigurációval, futási naplóval és reprodukálható parancssori belépéssel bővítjük tovább.