This capstone project challenges you to build, package, deploy, and monitor a complete end-to-end production machine learning system for IPL player performance analytics. You will implement all three production phases in sequence: a reproducible data pipeline and experiment-tracked training workflow (Phase 1), a containerised REST API with prediction and SHAP explainability endpoints (Phase 2), and an Evidently-powered drift monitoring system with alerting (Phase 3). The finished artefact — a working Docker service with an attached monitoring layer — constitutes a portfolio piece that demonstrates the full MLOps engineering discipline required by industry roles.
Learning Objectives
- Implement a reproducible data ingestion and feature engineering pipeline that generates consistent training datasets from raw IPL match records.
- Track multiple XGBoost training runs with MLflow, apply a two-gate quality criterion, and promote the best model to the Staging registry stage.
- Wrap the registered model in a FastAPI application with /health, /predict, and /explain endpoints, validated by Pydantic schemas with field-level constraints.
- Containerise the full service stack with a production-grade Dockerfile and docker-compose.yml, passing all automated endpoint tests.
- Integrate Evidently drift detection to monitor live inference traffic and trigger an alert when feature or prediction drift exceeds configured thresholds.
Technical Requirements
Your submission must include all source files committed to a GitHub repository with a clear README. The repository must contain: a src/ directory with ingestion.py, features.py, and train.py; an api/ directory with main.py, schemas.py, and explainer.py; a monitoring/ directory with drift_report.py; a Dockerfile and docker-compose.yml at the project root; and a requirements.txt pinning all dependency versions. The MLflow tracking store must be volumised so experiment data persists across container restarts. All three pytest test files (test_train.py, test_api.py, test_explain.py) must pass with zero failures.
Architecture
# Complete project architecture — annotated folder structure
architecture_diagram = """
ipl-performance-predictor/ # Project root
├── data/
│ ├── raw/ # Unprocessed IPL match CSV
│ │ └── ipl_matches_raw.csv
│ ├── processed/
│ │ └── ipl_features.parquet # Feature-engineered Parquet
│ └── reference/
│ └── reference_distribution.parquet # Evidently reference window
├── src/
│ ├── ingestion.py # Load + validate raw data
│ ├── features.py # Feature engineering functions
│ └── train.py # MLflow training + promotion
├── api/
│ ├── __init__.py
│ ├── main.py # FastAPI app (lifespan, routes)
│ ├── schemas.py # Pydantic input/output models
│ └── explainer.py # SHAP TreeExplainer wrapper
├── monitoring/
│ ├── drift_report.py # Evidently drift detection
│ └── alert.py # Threshold alert logic
├── tests/
│ ├── test_train.py # MLflow training tests
│ ├── test_api.py # FastAPI endpoint tests
│ └── test_explain.py # SHAP endpoint tests
├── mlruns/ # MLflow local tracking store (volumised)
├── reports/ # Evidently HTML/JSON reports
│ └── drift_report_<timestamp>.html
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── README.md
"""
print(architecture_diagram)Phase 1: Data Pipeline and MLflow Training
Phase 1 establishes the data foundation and experiment tracking infrastructure. The ingestion module loads raw IPL match records, validates that expected columns are present, removes rows with null values in target or feature columns, and saves a cleaned Parquet file. The features module derives aggregate per-player statistics (rolling batting average, boundary percentage, dot-ball percentage) and the opponent bowling quality score. The train module runs three XGBoost configurations, logs all params and metrics to MLflow, and promotes the best model to Staging using the two-gate quality criterion (RMSE and R²).
# src/train.py — complete Phase 1 training script
import numpy as np
import pandas as pd
import xgboost as xgb
import mlflow
import mlflow.xgboost
from mlflow import MlflowClient
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
MLFLOW_URI = "http://localhost:5000"
EXPERIMENT = "ipl-strike-rate-experiment"
MODEL_NAME = "ipl-strike-rate-predictor"
RMSE_GATE = 18.0
R2_GATE = 0.75
FEATURE_COLS = [
"batting_average", "innings_count", "boundary_pct",
"dot_ball_pct", "opponent_bowling_quality", "venue_encoded", "is_knockout"
]
xgb_configs = [
{"run_name": "conservative", "n_estimators": 200, "learning_rate": 0.03, "max_depth": 4, "colsample_bytree": 0.8},
{"run_name": "high-capacity", "n_estimators": 300, "learning_rate": 0.10, "max_depth": 8, "colsample_bytree": 0.9},
{"run_name": "regularised", "n_estimators": 150, "learning_rate": 0.05, "max_depth": 5, "colsample_bytree": 0.75},
]
def load_features(parquet_path: str) -> tuple:
ipl_df = pd.read_parquet(parquet_path)
X = ipl_df[FEATURE_COLS].values
y = ipl_df["strike_rate"].values
return train_test_split(X, y, test_size=0.2, random_state=42)
def run_training_pipeline(parquet_path: str) -> str:
mlflow.set_tracking_uri(MLFLOW_URI)
mlflow.set_experiment(EXPERIMENT)
client = MlflowClient()
X_train, X_test, y_train, y_test = load_features(parquet_path)
for cfg in xgb_configs:
params = {k: v for k, v in cfg.items() if k != "run_name"}
with mlflow.start_run(run_name=cfg["run_name"]):
mlflow.log_params(params)
ipl_model = xgb.XGBRegressor(objective="reg:squarederror", random_state=42, **params)
ipl_model.fit(X_train, y_train)
batting_preds = ipl_model.predict(X_test)
mlflow.log_metric("mae", mean_absolute_error(y_test, batting_preds))
mlflow.log_metric("rmse", np.sqrt(mean_squared_error(y_test, batting_preds)))
mlflow.log_metric("r2", r2_score(y_test, batting_preds))
mlflow.xgboost.log_model(ipl_model, artifact_path="model")
# Promote best model to Staging
experiment = client.get_experiment_by_name(EXPERIMENT)
best_runs = client.search_runs(
[experiment.experiment_id],
filter_string=f"metrics.rmse < {RMSE_GATE} and metrics.r2 > {R2_GATE}",
order_by=["metrics.rmse ASC"], max_results=1
)
if not best_runs:
raise RuntimeError("No run passed quality gates.")
best_run_id = best_runs[0].info.run_id
mv = mlflow.register_model(f"runs:/{best_run_id}/model", MODEL_NAME)
client.transition_model_version_stage(MODEL_NAME, mv.version, "Staging", archive_existing_versions=True)
print(f"Phase 1 complete: version {mv.version} promoted to Staging")
return mv.version
if __name__ == "__main__":
run_training_pipeline("data/processed/ipl_features.parquet")Phase 2: FastAPI + Docker Deployment with /predict and /explain
Phase 2 packages the Phase 1 model into a production-grade REST service. The FastAPI application loads both the pyfunc model and the SHAP explainer at startup from the same Staging model URI, exposing /health, /predict, and /explain endpoints. The Dockerfile uses a slim Python 3.11 base, separates dependency installation from code copy for build-cache efficiency, runs as a non-root user, and includes a Docker HEALTHCHECK. The docker-compose.yml wires the API service to the MLflow service via a named network. All three endpoints must return correct responses within 200 ms under the test load.
# Phase 2 acceptance tests — must all pass before Phase 3
import httpx
import pytest
BASE_URL = "http://localhost:8000"
# Virat Kohli profile — consistent, high average
kohli_match_features = {
"batting_average": 37.21,
"innings_count": 237,
"boundary_pct": 0.49,
"dot_ball_pct": 0.34,
"opponent_bowling_quality": 7.8,
"venue_encoded": 3,
"is_knockout": 1,
}
# Shubman Gill profile — young, aggressive
gill_match_features = {
"batting_average": 34.85,
"innings_count": 78,
"boundary_pct": 0.56,
"dot_ball_pct": 0.29,
"opponent_bowling_quality": 7.2,
"venue_encoded": 8,
"is_knockout": 0,
}
def test_health_returns_model_loaded():
r = httpx.get(f"{BASE_URL}/health")
assert r.status_code == 200
assert r.json()["model_loaded"] is True
print(f"[PASS] /health: model loaded, version {r.json()['model_version']}")
def test_predict_kohli_in_range():
r = httpx.post(f"{BASE_URL}/predict", json=kohli_match_features)
assert r.status_code == 200
sr = r.json()["predicted_strike_rate"]
assert 80 <= sr <= 250, f"Predicted SR {sr} out of realistic range"
print(f"[PASS] /predict Virat Kohli SR = {sr}")
def test_explain_shap_consistency_gill():
r = httpx.post(f"{BASE_URL}/explain", json=gill_match_features)
assert r.status_code == 200
body = r.json()
shap_sum = sum(a["shap_value"] for a in body["feature_attributions"])
reconstructed = body["base_value"] + shap_sum
assert abs(reconstructed - body["predicted_strike_rate"]) < 0.05
print(f"[PASS] /explain Shubman Gill SR = {body['predicted_strike_rate']}, SHAP consistent")
def test_invalid_input_rejected():
bad_input = kohli_match_features.copy()
bad_input["batting_average"] = -10 # invalid: ge=0
r = httpx.post(f"{BASE_URL}/predict", json=bad_input)
assert r.status_code == 422, f"Expected 422, got {r.status_code}"
print("[PASS] Invalid input correctly rejected with 422")
if __name__ == "__main__":
test_health_returns_model_loaded()
test_predict_kohli_in_range()
test_explain_shap_consistency_gill()
test_invalid_input_rejected()
print("\nAll Phase 2 acceptance tests PASSED.")Phase 3: Evidently Drift Monitoring and Alerting
Phase 3 adds observability to the deployed system. You will create a reference dataset from a representative sample of the training data, log all incoming /predict requests to a JSON file, run Evidently's DataDriftPreset and TargetDriftPreset against a rolling window of recent inference traffic, generate an HTML drift report, and trigger an alert if any feature's drift score exceeds 0.15 (a commonly used PSI threshold). The alert module should print a warning message and could be extended to send an email or Slack notification in a production system.
# monitoring/drift_report.py — complete Evidently drift monitoring
import pandas as pd
import numpy as np
import json
from pathlib import Path
from datetime import datetime
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, TargetDriftPreset
from evidently.metrics import DatasetDriftMetric
FEATURE_COLS = [
"batting_average", "innings_count", "boundary_pct",
"dot_ball_pct", "opponent_bowling_quality", "venue_encoded", "is_knockout"
]
DRIFT_THRESHOLD = 0.15
REPORTS_DIR = Path("reports")
REPORTS_DIR.mkdir(exist_ok=True)
def load_reference_data(parquet_path: str) -> pd.DataFrame:
"""Load training data as reference distribution for drift detection."""
df = pd.read_parquet(parquet_path)
return df[FEATURE_COLS + ["strike_rate"]].rename(columns={"strike_rate": "target"})
def load_inference_log(log_path: str) -> pd.DataFrame:
"""Load recent inference requests from JSON log file."""
records = []
with open(log_path) as f:
for line in f:
row = json.loads(line.strip())
records.append(row)
inference_df = pd.DataFrame(records)
# Rename prediction column to 'target' for Evidently
if "predicted_strike_rate" in inference_df.columns:
inference_df = inference_df.rename(columns={"predicted_strike_rate": "target"})
return inference_df[FEATURE_COLS + ["target"]]
def run_drift_detection(reference_df: pd.DataFrame, current_df: pd.DataFrame) -> dict:
"""Run Evidently drift report and return drift metrics."""
ipl_drift_report = Report(metrics=[
DatasetDriftMetric(),
DataDriftPreset(),
TargetDriftPreset(),
])
ipl_drift_report.run(reference_data=reference_df, current_data=current_df)
# Save HTML report
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
report_path = REPORTS_DIR / f"drift_report_{timestamp}.html"
ipl_drift_report.save_html(str(report_path))
print(f"[monitoring] Drift report saved: {report_path}")
# Extract drift metrics as dict
report_dict = ipl_drift_report.as_dict()
dataset_drift = report_dict["metrics"][0]["result"]
return {
"dataset_drift_detected": dataset_drift.get("dataset_drift", False),
"drift_share": dataset_drift.get("drift_share", 0.0),
"number_of_drifted_features": dataset_drift.get("number_of_drifted_features", 0),
"report_path": str(report_path),
}
def check_and_alert(drift_results: dict) -> None:
"""Trigger alert if dataset drift detected."""
if drift_results["dataset_drift_detected"]:
drifted = drift_results["number_of_drifted_features"]
share = drift_results["drift_share"]
print(f"""\n{'='*60}
ALERT: Data drift detected in IPL inference traffic!
Drifted features: {drifted}
Drift share : {share:.2%}
Report : {drift_results['report_path']}
Action required : Investigate feature distributions and consider
retraining with recent IPL match data.
{'='*60}""")
else:
print(f"[monitoring] No drift detected. Drift share: {drift_results['drift_share']:.2%}")
def simulate_drifted_inference_log(n_samples: int = 200) -> pd.DataFrame:
"""Simulate drifted inference traffic for testing — higher boundary_pct distribution."""
np.random.seed(99)
return pd.DataFrame({
"batting_average": np.random.normal(40, 8, n_samples).clip(10, 70),
"innings_count": np.random.randint(50, 300, n_samples),
"boundary_pct": np.random.beta(8, 3, n_samples), # drifted: higher boundary %
"dot_ball_pct": np.random.beta(2, 7, n_samples), # drifted: lower dot %
"opponent_bowling_quality": np.random.normal(6.0, 0.8, n_samples).clip(4, 9), # drifted
"venue_encoded": np.random.randint(0, 12, n_samples),
"is_knockout": np.random.randint(0, 2, n_samples),
"target": np.random.normal(150, 20, n_samples).clip(80, 220),
})
if __name__ == "__main__":
reference_data = load_reference_data("data/processed/ipl_features.parquet")
# In production: load from inference log file
# current_data = load_inference_log("logs/inference.jsonl")
# For demo: use simulated drifted data
current_data = simulate_drifted_inference_log(200)
drift_results = run_drift_detection(reference_data, current_data)
check_and_alert(drift_results)
print(f"Drift summary: {drift_results}")Evaluation Rubric
- Code Quality (20 pts): PEP 8 compliance, meaningful variable names using cricket-themed conventions, docstrings on all public functions, no bare except clauses, and requirements.txt with pinned versions.
- MLflow Tracking Completeness (20 pts): At least 3 experiment runs logged with all hyperparameters and metrics (MAE, RMSE, R²), Staging model version present in registry, and quality-gate promotion script passing with correct thresholds.
- API Functionality (20 pts): /health, /predict, and /explain endpoints all return correct responses, Pydantic validation rejects invalid inputs with 422, and all Phase 2 acceptance tests pass within 200 ms median latency.
- Explainability (20 pts): /explain endpoint returns SHAP attributions for all 7 features sorted by absolute value, SHAP consistency check passes (|base + sum - prediction| < 0.05), and sensibility check passes (high boundary_pct → positive SHAP).
- Monitoring Setup (20 pts): Evidently drift report generated from a simulated drifted dataset, HTML report saved to reports/ directory, alert printed when drift_share > 0, and README documents how to run the monitoring script with real inference data.
Pro Tip
For maximum portfolio impact, add a GitHub Actions workflow (.github/workflows/ci.yml) that runs pytest tests/test_train.py, tests/test_api.py, and tests/test_explain.py on every push to main. Reviewers and hiring managers immediately see a green CI badge, which signals professional engineering standards — the same signal that IPL franchises look for when evaluating team infrastructure: 'Is this system reliable and self-verifying, or does it depend on manual checking?'
Warning: Never commit secrets (MLflow passwords, container registry tokens, or database URLs) to your GitHub repository — even in a private repo. Use GitHub Actions secrets for the CI workflow and environment variables loaded at container runtime for the API. The model card must explicitly state what data was used for training and whether it contains any PII: submitting a model card that claims no PII when player names or match IDs are included in the training set is a data governance failure that disqualifies the project from the documentation rubric regardless of technical correctness.