System Architecture

def load_dataset(config: dict) -> pd.DataFrame:
    data_path = config["data"]["path"]
    df = pd.read_csv(data_path)
    # Feature engineering applied during load
    return add_engineered_features(df, fcfg)

data:
  path: ml_datasource.csv
  target: purchased
  test_size: 0.2

def build_preprocessor(X_train: pd.DataFrame, config: dict) -> ColumnTransformer:
    num_cols = X_train.select_dtypes(include=np.number).columns.tolist()
    cat_cols = X_train.select_dtypes(exclude=np.number).columns.tolist()
    
    numeric_transformer = Pipeline(steps=[("scaler", StandardScaler())])
    categorical_transformer = Pipeline(steps=[("onehot", OneHotEncoder(handle_unknown="ignore"))])
    
    return ColumnTransformer(
        transformers=[
            ("num", numeric_transformer, num_cols),
            ("cat", categorical_transformer, cat_cols),
        ]
    )

cv = StratifiedKFold(
    n_splits=5,
    shuffle=True,
    random_state=int(config["seed"]),
)
scoring = {"roc_auc": "roc_auc", "precision": "precision", "recall": "recall", "f1": "f1"}

target_precision = float(config["business"]["target_precision"])  # 0.9
candidates = [i for i, p in enumerate(precisions[:-1]) if p >= target_precision]
if candidates:
    idx = max(candidates, key=lambda i: recalls[i])  # Maximize recall
threshold = float(thresholds[idx])

artifacts/
├── best_model.joblib          # Trained sklearn pipeline
├── threshold.txt              # Calibrated decision threshold
├── metrics.json               # Performance metrics + CV ranking
├── lineage.json               # SHA-256 hashes for reproducibility
└── drift_baseline.json        # Training distribution statistics

python benchmarking/statistical_benchmark.py --runs 10 --batch-size 256

python hardware_aware_ml/tradeoff_experiments.py

from skl2onnx import to_onnx

initial_types = _build_initial_types(X_test)
onx = to_onnx(model, initial_types=initial_types, target_opset=15)
(out_dir / "model.onnx").write_bytes(onx.SerializeToString())

python deployment/parity_check.py --abs-tol 0.04 --mean-tol 0.01

app = FastAPI(title="Student Purchase Prediction API", version="1.2.0", lifespan=lifespan)

@app.post("/predict", response_model=PredictResponse)
def predict(payload: PredictRequest) -> PredictResponse:
    result = _predict_records([payload])[0]
    return result

class PredictRequest(BaseModel):
    student_country: str = Field(..., min_length=2, max_length=64)
    days_on_platform: int = Field(..., ge=0)
    minutes_watched: float = Field(..., ge=0)
    courses_started: int = Field(..., ge=0)
    practice_exams_started: int = Field(..., ge=0)
    practice_exams_passed: int = Field(..., ge=0)
    minutes_spent_on_exams: float = Field(..., ge=0)

class PredictResponse(BaseModel):
    predicted_purchase_probability: float
    predicted_purchase: int

@asynccontextmanager
async def lifespan(_: FastAPI):
    load_artifacts()  # Load model, threshold, drift baseline
    yield

with _LOCK:
    for col in numeric_cols:
        _MONITORING["feature_sums"][col] += float(feat_df[col].sum())
    _MONITORING["samples"] += len(feat_df)
    _MONITORING["predicted_positive"] += int(preds.sum())

current_mean = float(feature_sum) / samples
base_mean = float(baseline_stats[feature]["mean"])
base_std = max(float(baseline_stats[feature]["std"]), 1e-6)
abs_z = abs((current_mean - base_mean) / base_std)

if abs_z >= z_threshold:  # Default: 3.0
    drifted_features.append(feature)

{
  "timestamp_utc": "2026-03-04T19:30:45.123456Z",
  "threshold": 0.6524,
  "predicted_purchase_probability": 0.8234,
  "predicted_purchase": 1,
  "features": {
    "student_country": "US",
    "days_on_platform": 12,
    "minutes_watched": 366.7
  }
}

.
├── src/                          # Core training and serving code
│   ├── train.py                  # Training pipeline orchestrator
│   ├── api.py                    # FastAPI service
│   ├── data.py                   # Data loading and splitting
│   └── features.py               # Feature engineering
├── deployment/                   # ONNX export and validation
│   ├── export_onnx.py           # sklearn → ONNX conversion
│   ├── quantize_onnx.py         # INT8 quantization
│   └── parity_check.py          # Numerical equivalence testing
├── benchmarking/                 # Performance measurement
│   ├── statistical_benchmark.py # Latency/throughput benchmarks
│   └── dashboard.py             # Visualization
├── hardware_aware_ml/           # Trade-off analysis
│   └── tradeoff_experiments.py  # Accuracy vs latency curves
├── real_time_pipelines/         # Streaming simulation
│   └── unified_pipeline.py      # Event-driven inference
├── config/                      # Extended configuration
│   ├── datasets.yaml            # Dataset metadata
│   ├── experiments.yaml         # Experiment tracking
│   └── reproducibility.yaml     # Determinism settings
├── scripts/                     # Utility scripts
│   ├── reproducibility_check.py # Hash verification
│   └── cleanup_artifacts.py     # Artifact management
├── config.yaml                  # Main configuration file
├── requirements.txt             # Python dependencies
└── Makefile                     # Task automation

from src.data import load_config

config = load_config()  # Returns dict from config.yaml
seed = int(config["seed"])  # Global random seed
model_dir = config["artifacts"]["model_dir"]  # Artifact output path

train:
    python -m src.train

test:
    pytest

benchmark:
    python benchmarking/statistical_benchmark.py --runs 10 --batch-size 256
    python hardware_aware_ml/tradeoff_experiments.py

deploy-check:
    python deployment/export_onnx.py
    python deployment/quantize_onnx.py
    python deployment/parity_check.py --abs-tol 0.04 --mean-tol 0.01