From Notebook to Project

%%writefile bank_marketing_project/src/data/dataset.py"""Load, split and encode the UCI Bank Marketing dataset.Extracted from Module 2 Lesson 2.2. The `duration` column is dropped as a target leak:it is only known after the outcome it predicts."""from __future__ import annotationsimport urllib.requestimport zipfilefrom pathlib import Pathimport numpy as npimport pandas as pdimport torchfrom sklearn.model_selection import train_test_splitfrom sklearn.preprocessing import StandardScalerfrom torch.utils.data import DataLoader, TensorDatasetBANK_URL = "https://archive.ics.uci.edu/static/public/222/bank+marketing.zip"LEAKY_COLUMNS = ["duration"]def download(data_dir: Path) -> Path:    """Fetch ``bank-full.csv`` into ``data_dir`` unless it is already there.    The UCI archive nests a second zip, so both are downloaded and unpacked, the CSV is    kept, and the two archives are deleted. Later calls return the cached file at once.    Args:        data_dir: Folder to cache the dataset in; created if it does not exist.    Returns:        Path to ``bank-full.csv``.    Raises:        urllib.error.URLError: If the file is missing and UCI cannot be reached.        KeyError: If the archive no longer contains the expected files.    """    data_dir = Path(data_dir)    data_dir.mkdir(parents=True, exist_ok=True)    csv_path = data_dir / "bank-full.csv"    if csv_path.exists():        return csv_path    outer = data_dir / "bank+marketing.zip"    urllib.request.urlretrieve(BANK_URL, outer)    with zipfile.ZipFile(outer) as z:        z.extract("bank.zip", data_dir)    with zipfile.ZipFile(data_dir / "bank.zip") as z:        z.extract("bank-full.csv", data_dir)    outer.unlink()    (data_dir / "bank.zip").unlink()    return csv_pathdef load_splits(data_dir: Path, seed: int = 42) -> dict:    """Load Bank Marketing and return encoded, scaled train/validation/test arrays.    Drops the leaky ``duration`` column, makes a stratified 70/15/15 split with the    given seed, one-hot encodes the categorical columns and standardises the numeric    ones. Encoding columns and scaling statistics come from the training split only,    and validation and test are forced onto the same columns.    Args:        data_dir: Folder holding (or to receive) ``bank-full.csv``.        seed: Random seed for both splits, so the same seed gives the same rows.    Returns:        A dict with float32 feature matrices ``X_train``, ``X_val``, ``X_test``, int64        label vectors ``y_train``, ``y_val``, ``y_test`` (1 = subscribed) and the list of        encoded column names ``feature_names``.    """    df = pd.read_csv(download(data_dir), sep=";").drop(columns=LEAKY_COLUMNS)    y = (df.pop("y") == "yes").astype(np.int64).values    numeric = [c for c in df.columns if df[c].dtype.kind in "iuf"]    categorical = [c for c in df.columns if c not in numeric]    X_tr, X_tmp, y_tr, y_tmp = train_test_split(        df, y, test_size=0.30, stratify=y, random_state=seed)    X_va, X_te, y_va, y_te = train_test_split(        X_tmp, y_tmp, test_size=0.50, stratify=y_tmp, random_state=seed)    train_ohe = pd.get_dummies(X_tr, columns=categorical)    names = list(train_ohe.columns)    scaler = StandardScaler().fit(train_ohe[numeric])    def encode(frame: pd.DataFrame) -> np.ndarray:        """Encode one split exactly like the training split.        Args:            frame: Feature rows from one split of the original DataFrame.        Returns:            A float32 array whose columns match the training one-hot columns (missing            categories filled with 0), with numeric columns standardised by the training            scaler.        """        out = (pd.get_dummies(frame, columns=categorical)               .reindex(columns=names, fill_value=0)               .astype(np.float32))        out[numeric] = scaler.transform(out[numeric]).astype(np.float32)        return out.values.astype(np.float32)    return {        "X_train": encode(X_tr), "y_train": y_tr,        "X_val": encode(X_va), "y_val": y_va,        "X_test": encode(X_te), "y_test": y_te,        "feature_names": names,    }def make_loaders(splits: dict, batch_size: int = 256, eval_batch_size: int = 1024):    """Wrap encoded arrays in DataLoaders for training, validation and test.    Labels are converted to float and reshaped to ``(N, 1)`` so they match the model's    one-logit output and ``BCEWithLogitsLoss``. Only the training loader shuffles.    Args:        splits: The dict returned by ``load_splits``.        batch_size: Rows per training batch.        eval_batch_size: Rows per validation/test batch (larger: no gradients are kept).    Returns:        A tuple ``(train_loader, val_loader, test_loader)``.    """    def build(X, y, bs, shuffle):        """Build one DataLoader over a ``TensorDataset``.        Args:            X: Float32 feature matrix of shape ``(N, F)``.            y: Integer label vector of shape ``(N,)``.            bs: Batch size.            shuffle: Whether to reshuffle the rows every epoch.        Returns:            A ``DataLoader`` yielding ``(x, y)`` with ``y`` shaped ``(batch, 1)``.        """        ds = TensorDataset(            torch.from_numpy(X),            torch.from_numpy(y).float().unsqueeze(1),        )        return DataLoader(ds, batch_size=bs, shuffle=shuffle)    return (        build(splits["X_train"], splits["y_train"], batch_size, True),        build(splits["X_val"], splits["y_val"], eval_batch_size, False),        build(splits["X_test"], splits["y_test"], eval_batch_size, False),    )

Writing bank_marketing_project/src/data/dataset.py

%%writefile bank_marketing_project/src/training/train.py"""Training entry point.Run from the project root:    python -m src.training.train --config configs/baseline.yaml"""from __future__ import annotationsimport argparseimport copyimport jsonimport osimport randomimport timefrom pathlib import Pathimport numpy as npimport torchimport torch.nn as nnimport yamlfrom src.data.dataset import load_splits, make_loadersfrom src.evaluation.metrics import evaluate, recall_at_budgetfrom src.models.mlp import BankMLP, count_parametersdef seed_everything(seed: int) -> None:    """Seed every random-number generator that affects a run.    Seeds Python's ``random``, NumPy and PyTorch (all CUDA devices too when present), so    the split, weight initialisation and shuffling repeat for the same seed on the same    setup. Results can still differ slightly across CPU, CUDA and MPS. Setting    ``PYTHONHASHSEED`` here only affects processes started afterwards: Python reads it    at start-up.    Args:        seed: The seed value, normally ``cfg["seed"]``.    """    os.environ["PYTHONHASHSEED"] = str(seed)    random.seed(seed)    np.random.seed(seed)    torch.manual_seed(seed)    if torch.cuda.is_available():        torch.cuda.manual_seed_all(seed)def get_device(preference: str = "auto") -> torch.device:    """Choose the device to train on.    Args:        preference: ``"auto"`` to pick CUDA, then Apple MPS, then CPU; or an explicit            device string such as ``"cpu"``, ``"cuda"`` or ``"mps"``.    Returns:        A ``torch.device``.    """    if preference != "auto":        return torch.device(preference)    if torch.cuda.is_available():        return torch.device("cuda")    if torch.backends.mps.is_available() and torch.backends.mps.is_built():        return torch.device("mps")    return torch.device("cpu")def train_one_epoch(model, loader, loss_fn, optimizer, device) -> float:    """Train for one pass over the loader and return the mean loss per example.    For every batch: move it to the device, clear gradients, forward, compute the loss,    backpropagate and step. Each batch's loss is weighted by its real size, so a short    last batch does not distort the average.    Args:        model: The model to train (switched to training mode inside).        loader: DataLoader yielding ``(x, y)`` training batches.        loss_fn: Loss taking ``(logits, targets)``.        optimizer: Optimizer built from ``model.parameters()``.        device: Device the model lives on.    Returns:        Mean training loss per example over the epoch.    """    model.train()    running_loss, n_seen = 0.0, 0    for xb, yb in loader:        xb, yb = xb.to(device), yb.to(device)        optimizer.zero_grad()        loss = loss_fn(model(xb), yb)        loss.backward()        optimizer.step()        running_loss += loss.item() * xb.size(0)        n_seen += xb.size(0)    return running_loss / n_seendef main(config_path: str) -> dict:    """Run one complete experiment described by a YAML config file.    Seeds everything, picks a device, loads and encodes the data, builds ``BankMLP``,    Adam and ``BCEWithLogitsLoss``, trains with early stopping on validation loss,    restores the best weights, evaluates validation and then test, and saves a    checkpoint plus a JSON record of the run in ``checkpoint_dir``.    Args:        config_path: Path to a YAML file with ``run_name``, ``seed``, ``device``,            ``data_dir``, ``batch_size``, ``eval_batch_size``, ``hidden_sizes``,            ``learning_rate``, ``max_epochs``, ``patience``, ``min_delta`` and            ``checkpoint_dir``.    Returns:        The run record also written to ``<run_name>_results.json``: config, best epoch,        elapsed seconds, per-epoch history, validation and test metrics, torch version.    Raises:        KeyError: If the config is missing a required setting.    """    cfg = yaml.safe_load(Path(config_path).read_text())    seed_everything(cfg["seed"])    device = get_device(cfg.get("device", "auto"))    splits = load_splits(Path(cfg["data_dir"]), seed=cfg["seed"])    train_loader, val_loader, test_loader = make_loaders(        splits, batch_size=cfg["batch_size"], eval_batch_size=cfg["eval_batch_size"])    model = BankMLP(        in_features=splits["X_train"].shape[1],        hidden_sizes=tuple(cfg["hidden_sizes"]),    ).to(device)    optimizer = torch.optim.Adam(model.parameters(), lr=cfg["learning_rate"])    loss_fn = nn.BCEWithLogitsLoss()    print(f"device      : {device}")    print(f"features    : {splits['X_train'].shape[1]}")    print(f"train rows  : {splits['X_train'].shape[0]:,}")    print(f"parameters  : {count_parameters(model):,}")    print(f"max epochs  : {cfg['max_epochs']} (patience {cfg['patience']})")    print()    best_loss, best_epoch, best_weights = float("inf"), 0, None    stale = 0    history = []    started = time.time()    for epoch in range(1, cfg["max_epochs"] + 1):        train_loss = train_one_epoch(model, train_loader, loss_fn, optimizer, device)        val = evaluate(model, val_loader, loss_fn, device)        history.append({            "epoch": epoch,            "train_loss": train_loss,            "val_loss": val["loss"],            "val_roc_auc": val["roc_auc"],            "val_avg_precision": val["avg_precision"],        })        print(f"epoch {epoch:>3} | train {train_loss:.4f} | val {val['loss']:.4f} "              f"| ROC-AUC {val['roc_auc']:.4f} | AP {val['avg_precision']:.4f}")        if val["loss"] < best_loss - cfg["min_delta"]:            best_loss, best_epoch = val["loss"], epoch            best_weights = copy.deepcopy(model.state_dict())            stale = 0        else:            stale += 1            if stale >= cfg["patience"]:                print(f"\nEarly stop at epoch {epoch} (best was {best_epoch}).")                break    model.load_state_dict(best_weights)    elapsed = time.time() - started    val_final = evaluate(model, val_loader, loss_fn, device)    test_final = evaluate(model, test_loader, loss_fn, device)    print(f"\nBest epoch {best_epoch} | trained in {elapsed:.1f}s")    print(f"VAL   loss {val_final['loss']:.4f} | ROC-AUC {val_final['roc_auc']:.4f} "          f"| AP {val_final['avg_precision']:.4f}")    print(f"TEST  loss {test_final['loss']:.4f} | ROC-AUC {test_final['roc_auc']:.4f} "          f"| AP {test_final['avg_precision']:.4f}")    print(f"TEST  recall at 20% budget: "          f"{recall_at_budget(test_final['probs'], test_final['labels']):.4f}")    ckpt_dir = Path(cfg["checkpoint_dir"])    ckpt_dir.mkdir(parents=True, exist_ok=True)    torch.save({        "model_state_dict": model.state_dict(),        "optimizer_state_dict": optimizer.state_dict(),        "epoch": best_epoch,        "config": cfg,        "in_features": splits["X_train"].shape[1],        "torch_version": torch.__version__,    }, ckpt_dir / f"{cfg['run_name']}.pt")    record = {        "run_name": cfg["run_name"],        "config": cfg,        "best_epoch": best_epoch,        "elapsed_seconds": round(elapsed, 2),        "history": history,        "val": {k: v for k, v in val_final.items() if k not in ("probs", "labels")},        "test": {k: v for k, v in test_final.items() if k not in ("probs", "labels")},        "torch_version": torch.__version__,    }    (ckpt_dir / f"{cfg['run_name']}_results.json").write_text(json.dumps(record, indent=2))    print(f"\nSaved {ckpt_dir / (cfg['run_name'] + '.pt')}")    print(f"Saved {ckpt_dir / (cfg['run_name'] + '_results.json')}")    return recordif __name__ == "__main__":    parser = argparse.ArgumentParser(description="Train the Bank Marketing MLP.")    parser.add_argument("--config", required=True, help="path to a YAML config")    args = parser.parse_args()    main(args.config)

Writing bank_marketing_project/src/training/train.py

%%writefile bank_marketing_project/src/inference/predict.py"""Load a checkpoint and score new examples."""from __future__ import annotationsfrom pathlib import Pathimport numpy as npimport torchfrom src.models.mlp import BankMLPdef load_model(checkpoint_path: Path, device: torch.device | str = "cpu") -> torch.nn.Module:    """Rebuild a trained ``BankMLP`` from a checkpoint and switch it to evaluation mode.    The architecture comes from the checkpoint's own ``config`` and ``in_features``, so    the weights always load into the shape they were trained with.    Args:        checkpoint_path: A ``.pt`` file written by ``train.py``. Only load files you            trust: ``weights_only=False`` can execute pickled code.        device: Device to place the model on.    Returns:        The model, with weights loaded, in ``eval()`` mode.    """    ckpt = torch.load(checkpoint_path, map_location=device, weights_only=False)    model = BankMLP(        in_features=ckpt["in_features"],        hidden_sizes=tuple(ckpt["config"]["hidden_sizes"]),    ).to(device)    model.load_state_dict(ckpt["model_state_dict"])    model.eval()    return model@torch.no_grad()def predict_proba(model: torch.nn.Module, X: np.ndarray, device="cpu") -> np.ndarray:    """Return P(subscribe) for each row of already-encoded features.    Accepts a single row or a batch: a 1-D input is treated as one row. Runs without    gradients and applies the sigmoid to the model's logits.    Args:        model: A trained model, e.g. from ``load_model``.        X: Features encoded exactly as ``load_splits`` encodes them, shape            ``(in_features,)`` or ``(N, in_features)``. A raw CSV row will not work.        device: Device the model lives on.    Returns:        A 1-D NumPy array of probabilities, one per row.    """    tensor = torch.from_numpy(np.asarray(X, dtype=np.float32)).to(device)    if tensor.ndim == 1:                     # a single row -- restore the batch axis        tensor = tensor.unsqueeze(0)    return torch.sigmoid(model(tensor)).cpu().numpy().ravel()def rank_by_score(probs: np.ndarray, top_k: int) -> np.ndarray:    """Return the indices of the highest-scoring rows, best first.    Args:        probs: Scores, one per row.        top_k: How many rows to return (e.g. how many customers can be called).    Returns:        A NumPy array of up to ``top_k`` row indices, sorted by descending score.    """    return np.argsort(probs)[::-1][:top_k]

Writing bank_marketing_project/src/inference/predict.py

def run_in_project(command: list[str], timeout: int = 900) -> subprocess.CompletedProcess:    """Run a command from the project root, with the project importable.    Starts a separate process in ``PROJECT`` with ``PYTHONPATH`` pointing at it, prints    its standard output, and prints the end of standard error if the command fails.    Args:        command: The command and its arguments, e.g.            ``[sys.executable, "-m", "pytest", "tests/"]``.        timeout: Seconds to wait before giving up.    Returns:        The ``subprocess.CompletedProcess``; check ``returncode`` (0 means success).    Raises:        subprocess.TimeoutExpired: If the command runs longer than ``timeout``.    """    env = dict(os.environ, PYTHONPATH=str(PROJECT.resolve()))    result = subprocess.run(        command, cwd=PROJECT, env=env,        capture_output=True, text=True, timeout=timeout,    )    print(result.stdout)    if result.returncode != 0:        print("--- stderr ---")        print(result.stderr[-3000:])    return resultdef ensure_pytest():    # Some managed environments (PEP 668) refuse a plain pip install.    # Try the normal route first, then the override.    """Make sure pytest can be imported, installing it if necessary.    Tries a normal ``pip install`` first and then, for externally managed Pythons    (PEP 668), the same install with ``--break-system-packages``.    Returns:        ``True`` if pytest is available afterwards, ``False`` if both installs failed.    """    try:        import pytest  # noqa: F401        return True    except ImportError:        pass    print("pytest not found. Installing ...")    for extra in ([], ["--break-system-packages"]):        r = subprocess.run([sys.executable, "-m", "pip", "install", "-q", "pytest"] + extra,                           capture_output=True, text=True)        if r.returncode == 0:            return True    print("Could not install pytest automatically. Run: pip install pytest")    return Falsehave_pytest = ensure_pytest()test_run = run_in_project([sys.executable, "-m", "pytest", "tests/", "-v", "--no-header", "-q"])print(f"\nExit code: {test_run.returncode}  (0 means every test passed)")

============================= test session starts ==============================
collected 6 items

tests/test_shapes.py ......                                              [100%]

============================== 6 passed in 0.53s ===============================


Exit code: 0  (0 means every test passed)