Software Engineering Fundamentals for Data and ML Projects

The Gap Between a Working Script and a Production System

You can build a model that trains correctly and still produce code that nobody else can run, that breaks silently when input data changes, and that you cannot debug three weeks later. The gap between "working in my notebook" and "trustworthy in a team environment" is software engineering.

This module covers the engineering habits that make ML code maintainable, testable, and auditable - the habits hiring teams look for when reviewing take-home projects.

Code Organization: Single Responsibility

Every function, class, and module should do one thing. ML code that violates this creates entangled logic that is impossible to test in isolation.

pythonCopy
# ENTANGLED: loads data, preprocesses, trains, and evaluates in one function
def run_experiment(data_path, n_estimators):
    df = pd.read_csv(data_path)
    df = df.dropna()
    df["log_income"] = np.log1p(df["income"])
    X = df.drop("label", axis=1).values
    y = df["label"].values
    X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2)
    model = RandomForestClassifier(n_estimators=n_estimators)
    model.fit(X_train, y_train)
    auc = roc_auc_score(y_val, model.predict_proba(X_val)[:, 1])
    print(f"AUC: {auc:.4f}")

pythonCopy
# SEPARATED: each step is testable and reusable independently
from pathlib import Path
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import roc_auc_score

def load_data(path: Path) -> pd.DataFrame:
    return pd.read_csv(path).dropna()

def build_features(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df["log_income"] = np.log1p(df["income"])
    return df

def split(df: pd.DataFrame, target: str, test_size: float = 0.2):
    X = df.drop(target, axis=1).values
    y = df[target].values
    return train_test_split(X, y, test_size=test_size, random_state=42)

def train(X_train, y_train, n_estimators: int = 100) -> RandomForestClassifier:
    model = RandomForestClassifier(n_estimators=n_estimators, random_state=42)
    return model.fit(X_train, y_train)

def evaluate(model, X_val, y_val) -> dict[str, float]:
    auc = roc_auc_score(y_val, model.predict_proba(X_val)[:, 1])
    return {"roc_auc": auc}

Now each function can be tested in isolation, reused across scripts, and debugged independently.

Logging: Replace Print Statements

Logging is configurable, timestamped, filterable, and goes to files. Print statements do none of these things.

pythonCopy
import logging
from pathlib import Path

def setup_logging(log_file: Path | None = None, level: str = "INFO") -> None:
    handlers = [logging.StreamHandler()]
    if log_file:
        log_file.parent.mkdir(parents=True, exist_ok=True)
        handlers.append(logging.FileHandler(log_file))

    logging.basicConfig(
        level=getattr(logging, level),
        format="%(asctime)s %(name)s %(levelname)s %(message)s",
        handlers=handlers,
    )

logger = logging.getLogger(__name__)

def train(X_train, y_train, config):
    logger.info("Starting training: n_estimators=%d", config.n_estimators)
    model = RandomForestClassifier(**config.to_dict())
    model.fit(X_train, y_train)
    logger.info("Training complete")
    return model

Log at INFO for normal progress, WARNING for recoverable issues (missing values above threshold, unexpected schema), and ERROR for failures that require intervention.

Configuration Management: No Hardcoded Values

Hardcoded hyperparameters and paths are invisible to your experiment tracker and fragile to change:

pythonCopy
# WRONG: hardcoded configuration
def train():
    model = XGBClassifier(n_estimators=200, max_depth=6, learning_rate=0.05)
    X_train = pd.read_csv("/home/user/data/train.csv")
    ...

pythonCopy
# CORRECT: configuration object
from dataclasses import dataclass, field, asdict
from pathlib import Path
import yaml

@dataclass
class TrainConfig:
    data_path: Path = Path("data/processed/train.parquet")
    model_type: str = "xgboost"
    n_estimators: int = 200
    max_depth: int = 6
    learning_rate: float = 0.05
    test_size: float = 0.2
    random_seed: int = 42
    output_dir: Path = Path("models/")

    @classmethod
    def from_yaml(cls, path: str | Path) -> "TrainConfig":
        with open(path) as f:
            data = yaml.safe_load(f)
        return cls(**{k: Path(v) if k.endswith("_path") or k.endswith("_dir") else v
                      for k, v in data.items()})

    def to_dict(self) -> dict:
        return asdict(self)

Log the config at the start of every training run so experiments are reproducible:

pythonCopy
import mlflow
config = TrainConfig.from_yaml("configs/train_v3.yaml")
with mlflow.start_run():
    mlflow.log_params(config.to_dict())

Testing ML Code

Testing ML code is different from testing deterministic software. You test:

• Data preprocessing functions (deterministic, easy to unit-test)
• Feature pipeline contract (input schema → output schema)
• Model training stability (loss decreases, output shape is correct)
• Evaluation utilities (metric calculations are correct)

pythonCopy
# tests/test_features.py
import pytest
import pandas as pd
import numpy as np
from myproject.features.pipeline import build_features

class TestBuildFeatures:
    def setup_method(self):
        self.df = pd.DataFrame({
            "income": [50000, None, 75000, 30000],
            "age": [25, 35, 45, 28],
            "label": [0, 1, 0, 1],
        })

    def test_log_income_is_created(self):
        result = build_features(self.df)
        assert "log_income" in result.columns

    def test_log_income_is_non_negative(self):
        result = build_features(self.df.dropna())
        assert (result["log_income"] >= 0).all()

    def test_original_df_is_not_modified(self):
        original_cols = set(self.df.columns)
        build_features(self.df)
        assert set(self.df.columns) == original_cols  # No in-place modification

    def test_handles_missing_income(self):
        result = build_features(self.df)
        # Missing income rows should not cause an exception
        assert len(result) == len(self.df)

Run tests with pytest tests/ -v. Aim for tests on every public function in your feature and evaluation modules.

Data Validation: Catching Problems at the Boundary

Validate data at the point it enters your pipeline - before any processing:

pythonCopy
from dataclasses import dataclass
from typing import Any
import pandas as pd

@dataclass
class ColumnSpec:
    dtype: str
    nullable: bool = False
    min_value: float | None = None
    max_value: float | None = None
    allowed_values: list[Any] | None = None

EXPECTED_SCHEMA = {
    "user_id": ColumnSpec(dtype="object", nullable=False),
    "age": ColumnSpec(dtype="int64", nullable=False, min_value=0, max_value=120),
    "income": ColumnSpec(dtype="float64", nullable=True, min_value=0),
    "label": ColumnSpec(dtype="int64", nullable=False, allowed_values=[0, 1]),
}

def validate_dataframe(df: pd.DataFrame, schema: dict[str, ColumnSpec]) -> list[str]:
    errors = []
    for col, spec in schema.items():
        if col not in df.columns:
            errors.append(f"Missing column: {col}")
            continue
        if not spec.nullable and df[col].isna().any():
            null_pct = df[col].isna().mean()
            errors.append(f"{col}: unexpected nulls ({null_pct:.1%})")
        if spec.allowed_values and not df[col].dropna().isin(spec.allowed_values).all():
            bad = df[col].dropna()[~df[col].dropna().isin(spec.allowed_values)].unique()
            errors.append(f"{col}: unexpected values {bad}")
    return errors

# Use in the data loading step
errors = validate_dataframe(df, EXPECTED_SCHEMA)
if errors:
    raise ValueError(f"Data validation failed:\n" + "\n".join(errors))

Refactoring a Notebook Into a Package

The workflow:

1. Start in a Jupyter notebook - explore, prototype, visualize
2. Once an approach works, identify the reusable logic
3. Extract it into Python modules with clear function boundaries
4. Add type hints and logging
5. Write tests for the extracted functions
6. Create a CLI entry point (argparse or click)

The notebook stays in notebooks/ as documentation and exploration history. The extracted code in src/ is what runs in production.

Common Mistakes and Bad Instincts

One giant function that does everything. If you cannot test a function in isolation with a small synthetic input, it is too big. Break it up.

Testing against the real database or large production files. Tests should use small, in-memory fixtures (like the self.df above). Tests that depend on external systems are slow, fragile, and cannot run in CI.

Not testing edge cases. Empty DataFrames, all-null columns, single-row inputs - these are the cases that crash production systems. Add a test for each one you can think of.

Ignoring validation errors. If your validation function finds problems and you print them instead of raising an exception, the problems propagate silently downstream and produce incorrect model outputs.

Where to Go Next

Module 9 (Data Visualization and Exploratory Analysis With Judgment) shows how to use EDA not as a gallery of charts but as a systematic investigation that answers specific modeling questions - which features matter, where the data is broken, and what the model will struggle with.