Performance Reporting

Path: 3 — ML Engineering on AWS Position: 9 of 12 Difficulty: 🔴 Time: 3h AWS Cost: ~$0.05 (CloudWatch API calls only — endpoint from p3-07 must still be running, OR use the included sample data fixtures for testing)

---

Agent Pickup Instructions

# Bootstrap
cd projects/confident-prep/p3-09-performance-reporting
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

# Option A: run against a live endpoint (from p3-07)
python src/main.py \
  --endpoint-name p3-07-adult-income-endpoint \
  --days 7 \
  --output report.md

# Option B: dry run using sample data fixtures (no AWS needed)
python src/main.py \
  --endpoint-name mock \
  --days 7 \
  --output report.md \
  --dry-run

# Run tests (no AWS needed)
pytest tests/ -v

Done when:

• python src/main.py --output report.md writes a non-empty report.md
• report.md contains: current endpoint accuracy, baseline AUC from MLflow registry, delta value
• python src/main.py --dry-run works without any AWS credentials
• pytest tests/ -v shows 4/4 passing
• Script is structured to be safe as a cron job (no user interaction, clear exit codes, writes output to a file)

---

What this project is

A weekly performance monitoring script that pulls CloudWatch invocation metrics from a running SageMaker endpoint (InvocationCount, ModelLatency p50/p95, ErrorCount), runs the held-out test set through the endpoint to compute live accuracy, fetches the baseline AUC from p3-05's MLflow registry, and outputs a Markdown report showing current vs baseline with a delta. The script is designed to run unattended as a cron job.

What the learner achieves

"I can write an automated model performance report that compares live endpoint accuracy to the registered baseline, is safe to run as a cron job, and surfaces drift before it becomes a customer-facing problem."

---

Folder structure

p3-09-performance-reporting/
├── data/
│   └── adult.data              # reference dataset for live evaluation
├── fixtures/
│   ├── sample_cloudwatch_response.json   # fake CloudWatch API response for dry-run
│   └── sample_monitoring_metrics.json    # fake live eval metrics for dry-run
├── output/
│   └── report.md               # generated report
├── src/
│   ├── main.py                # CLI entry: --endpoint-name, --days, --output, --dry-run
│   ├── cloudwatch_fetcher.py  # fetch InvocationCount, ModelLatency, ErrorCount from CloudWatch
│   ├── live_evaluator.py      # send test CSV rows to endpoint, compute AUC and accuracy
│   ├── baseline_fetcher.py    # fetch "production" model's val_auc from MLflow registry
│   └── reporter.py            # generate Markdown report from fetched metrics
├── tests/
│   ├── test_cloudwatch_fetcher.py  # empty response handling
│   └── test_reporter.py           # delta calculation, markdown output correctness
├── .env.example
├── requirements.txt
└── README.md

---

.env.example

# AWS region
AWS_REGION=us-east-1

# MLflow tracking URI (to fetch baseline metrics)
MLFLOW_TRACKING_URI=http://localhost:5000

# MLflow model name (must match p3-05)
MLFLOW_MODEL_NAME=adult-income-xgboost

# Default endpoint name (overridable via --endpoint-name)
DEFAULT_ENDPOINT_NAME=p3-07-adult-income-endpoint

# Default report output path
REPORT_OUTPUT_PATH=output/report.md

# Dry run mode (uses fixtures instead of AWS — set to "true" for CI)
DRY_RUN=false

---

requirements.txt

boto3==1.35.93
scikit-learn==1.6.0
pandas==2.2.3
mlflow==2.19.0
numpy==2.2.1
python-dotenv==1.0.1
pytest==8.3.4

---

src/ — what to implement

src/cloudwatch_fetcher.py

fetch_endpoint_metrics(endpoint_name: str, days: int, region: str) -> dict

Fetches three CloudWatch metrics for the specified endpoint over the last days days.

For each metric, calls boto3.client('cloudwatch').get_metric_statistics():

• Namespace: "AWS/SageMaker"
• Dimensions: [{"Name": "EndpointName", "Value": endpoint_name}]
• StartTime: datetime.utcnow() - timedelta(days=days)
• EndTime: datetime.utcnow()
• Period: days * 86400 (aggregate over entire window)
• Statistics: ["Sum"] for InvocationCount and Errors; ["Average"] for ModelLatency

Metrics to fetch:

1. InvocationCount — total requests in window
2. ModelLatency — average in microseconds (SageMaker's unit); convert to ms by dividing by 1000
3. Errors — total error count

Returns:

{
    "endpoint_name": str,
    "period_days": int,
    "invocation_count": int,   # 0 if no datapoints
    "model_latency_p50_ms": float,  # 0.0 if no data
    "error_count": int,
    "error_rate": float,        # error_count / max(invocation_count, 1)
}

Edge cases:

• If Datapoints list is empty (endpoint has no traffic), return 0 for all metrics
• Do NOT raise on empty response — this is a normal case (new endpoint, or no traffic in window)
• If invocation_count == 0 and error_count == 0, set error_rate = 0.0

---

src/live_evaluator.py

evaluate_endpoint(endpoint_name: str, test_data_path: str, sample_size: int = 200) -> dict

Sends sample_size rows from the test set to the live endpoint and computes accuracy and AUC.

1. Load test_data_path (UCI Adult): assign column names, strip whitespace, replace ? with NaN, drop NaN rows, pd.get_dummies
2. Take the last 20% as test set (same split as training: random_state=42)
3. Sample sample_size rows from the test set (use sample_size if test set is larger, else use all)
4. For each row, call boto3.client('sagemaker-runtime').invoke_endpoint() with single-row CSV payload
5. Parse probability from response
6. Convert probabilities to binary predictions (threshold=0.5)
7. Compute accuracy_score and roc_auc_score vs ground-truth y_test

Returns:

{
    "accuracy": float,
    "auc": float,
    "n_samples": int,
    "endpoint_name": str,
}

Edge cases:

• If endpoint invocation fails for a row (non-200 status), skip the row and count it in a failed counter; print warning
• If all rows fail, raise RuntimeError("All invocations failed. Is the endpoint running?")
• Cap sample_size at available test rows silently

---

src/baseline_fetcher.py

get_baseline_from_registry(model_name: str, alias: str = "production") -> dict

Fetches the registered model's logged metrics from MLflow.

1. Constructs URI: f"models:/{model_name}@{alias}"
2. Uses mlflow.MlflowClient() to get model version by alias
3. Gets the run_id from the model version
4. Calls mlflow.get_run(run_id) to fetch the run's data
5. Returns:

{
    "val_auc": float,
    "val_accuracy": float,
    "run_id": str,
    "model_name": str,
    "alias": str,
}

Edge cases:

• If alias not found: return {"val_auc": None, "val_accuracy": None, "run_id": None, "model_name": model_name, "alias": alias}
• If val_auc metric not in run data: return None for that field
• Never raise — callers check for None values

---

src/reporter.py

compute_delta(current_metric: float | None, baseline_metric: float | None) -> dict

• Returns {"delta": float, "direction": str, "percent_change": float}
• direction is "improvement" if delta > 0, "regression" if delta < 0, "no change" if delta == 0
• If either metric is None, return {"delta": None, "direction": "unknown", "percent_change": None}

generate_weekly_report(cloudwatch_metrics: dict, live_eval: dict, baseline: dict, output_path: str) -> str

Generates and writes a Markdown performance report.

Required sections in order:

# Weekly Model Performance Report
Generated: {datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")}
Endpoint: {endpoint_name}
Period: Last {n_days} days

## Summary

| Metric | Current | Baseline | Delta |
|---|---|---|---|
| AUC | {live_auc:.4f} | {baseline_auc:.4f} | {delta:+.4f} |
| Accuracy | {live_accuracy:.4f} | {baseline_accuracy:.4f or "N/A"} | {delta:+.4f or "N/A"} |

**Status:** {OK/WARNING/UNKNOWN}
- OK: current AUC within 0.02 of baseline
- WARNING: current AUC drops more than 0.02 below baseline
- UNKNOWN: baseline or live metrics not available

## Endpoint Health (CloudWatch)

| Metric | Value |
|---|---|
| Invocations (last {n_days}d) | {invocation_count} |
| Avg Model Latency | {latency_ms:.1f} ms |
| Error Count | {error_count} |
| Error Rate | {error_rate:.2%} |

## Live Evaluation

Evaluated {n_samples} samples from held-out test set.
- AUC: {live_auc:.4f}
- Accuracy: {live_accuracy:.4f}

## Baseline (MLflow Registry)

- Model: {model_name} @ {alias}
- Run ID: {run_id}
- Baseline val_AUC: {baseline_auc:.4f}

## Recommendation

{recommendation text based on status}

Returns the report as a string AND writes to output_path.

---

src/main.py

CLI arguments:

• --endpoint-name (str, default from DEFAULT_ENDPOINT_NAME env var)
• --days (int, default 7)
• --output (str, default from REPORT_OUTPUT_PATH env var or output/report.md)
• --dry-run (flag) — uses fixture data instead of AWS API calls
• --sample-size (int, default 200) — rows to evaluate from live endpoint

load_fixtures() -> tuple[dict, dict, dict]

• Loads fixtures/sample_cloudwatch_response.json → cloudwatch_metrics
• Loads fixtures/sample_monitoring_metrics.json → live_eval
• Returns hardcoded baseline dict {"val_auc": 0.883, "val_accuracy": 0.872, "run_id": "fixture_run", "model_name": "adult-income-xgboost", "alias": "production"}

main()

1. Parse args
2. If --dry-run: call load_fixtures() for all three data sources
3. Else: call cloudwatch_fetcher.fetch_endpoint_metrics, live_evaluator.evaluate_endpoint, baseline_fetcher.get_baseline_from_registry
4. reporter.generate_weekly_report(...) → write output_path
5. Print: Report written to {output_path}
6. Exit 0 on success, exit 1 on any uncaught exception
7. Exceptions are caught at the top level and printed cleanly (no stack trace to stdout in cron mode)

---

tests/ — what to test

tests/test_cloudwatch_fetcher.py

test_fetch_returns_zero_when_no_datapoints

• Mock boto3.client('cloudwatch').get_metric_statistics to return {"Datapoints": []}
• Call cloudwatch_fetcher.fetch_endpoint_metrics("test-endpoint", days=7, region="us-east-1")
• Assert result["invocation_count"] == 0
• Assert result["error_count"] == 0
• Assert result["error_rate"] == 0.0

test_fetch_does_not_raise_on_empty_response

• Same mock as above
• Assert no exception is raised

tests/test_reporter.py

test_delta_is_positive_when_current_better_than_baseline

• compute_delta(current_metric=0.895, baseline_metric=0.880)
• Assert result["delta"] == pytest.approx(0.015, abs=0.001)
• Assert result["direction"] == "improvement"

test_delta_is_negative_when_current_worse

• compute_delta(current_metric=0.860, baseline_metric=0.880)
• Assert result["delta"] < 0
• Assert result["direction"] == "regression"

test_delta_returns_none_when_metrics_missing

• compute_delta(current_metric=None, baseline_metric=0.880)
• Assert result["delta"] is None
• Assert result["direction"] == "unknown"

test_report_markdown_contains_all_sections

• Call generate_weekly_report(mock_cw_metrics, mock_live_eval, mock_baseline, tmp_path / "report.md")
• Assert file size > 200 bytes
• Assert "## Summary" in content
• Assert "## Endpoint Health" in content
• Assert "## Baseline (MLflow Registry)" in content

---

README.md content

# P3-09 — Performance Reporting

Weekly model performance report: CloudWatch metrics + live evaluation + MLflow baseline.

## Prerequisites

- Python 3.11+
- AWS credentials configured (or use --dry-run for local testing)
- p3-05 MLflow server running with "production" alias registered
- Optional: p3-07 endpoint running for live evaluation

## Quick start (dry run — no AWS)

```bash
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
python src/main.py --dry-run --output output/report.md
cat output/report.md

Run against a live endpoint

cp .env.example .env  # fill in endpoint name and region
python src/main.py --endpoint-name p3-07-adult-income-endpoint --days 7 --output output/report.md

Set up as a cron job

0 9 * * 1  cd /path/to/p3-09-performance-reporting && .venv/bin/python src/main.py --output output/report-$(date +%Y%m%d).md >> logs/cron.log 2>&1

Runs every Monday at 9am. Output file is date-stamped.

Tests

pytest tests/ -v

Tests use mocked AWS responses — no credentials needed.

---

## GUIDE.md content

```markdown
# Guide — P3-09 Performance Reporting

## Why performance reporting matters

Deployed models degrade silently. The data distribution shifts, the world changes, and your
model keeps returning predictions — just increasingly wrong ones. Without a regular performance
check, you find out about degradation from users or from downstream business metrics, not from
your ML system.

## The three data sources

**CloudWatch:** Tells you about endpoint health — how many requests, what latency, how many errors.
It does NOT tell you if the model is accurate. An endpoint can have 0 errors and 100% wrong predictions.

**Live evaluation:** You send your held-out test set (or a sample) to the actual endpoint and
compare predictions to ground truth. This tells you actual current accuracy, but requires having
ground-truth labels available — which is not always the case.

**MLflow baseline:** The AUC logged when you registered the "production" model. This is your
reference point. The delta between live AUC and baseline AUC is the signal you monitor.

## The status thresholds

The ±0.02 AUC threshold is illustrative. In production, calibrate it to:
- Historical AUC variation between training runs (if your model naturally varies ±0.01, a 0.02 drop may not be meaningful)
- Business impact (for fraud detection, a 0.01 AUC drop may be critical; for content ranking, it may not be)

## Cron safety requirements

The script must:
- Exit with code 0 on success, non-zero on failure (cron can alert on non-zero exit)
- Write output to a file (not just stdout — cron captures stdout/stderr, but file output is more durable)
- Print nothing sensitive to stdout (logs go to CloudWatch or a file, not a user terminal)
- Handle all exceptions at the top level — never leave a partial JSON or broken Markdown file

## Interview framing

"I treat model monitoring as a regular reporting job: pull CloudWatch health metrics, evaluate
live accuracy on a sample of the test set, compare to the registered baseline, and surface the
delta as a status. The script runs as a cron job every week — so AUC degradation is caught
in days, not months."

---

Failure modes to handle

Failure	Where	How to handle
CloudWatch returns no datapoints (endpoint has zero traffic)	cloudwatch_fetcher.py	Return 0 for all metrics; report shows Invocations: 0
MLflow registry has no "production" alias	baseline_fetcher.py	Return None values; report shows Baseline: not available
Endpoint not running when live_evaluator runs	live_evaluator.py	Catch ClientError, include failed count in result, raise RuntimeError if all fail
AUC cannot be computed (all predictions same class)	live_evaluator.py	roc_auc_score raises ValueError — catch, set auc=None, report as N/A
Output directory does not exist	reporter.py	os.makedirs(dirname, exist_ok=True)
Script interrupted mid-write (partial report file)	reporter.py	Write to a temp file, then os.replace(temp, output_path) for atomic write

---

The metric this project measures

What: AUC delta between live endpoint evaluation and MLflow registry baseline

Format: Delta: +0.003 (improvement) or Delta: -0.025 (regression — WARNING)

Target: Delta within ±0.02 of baseline = status OK. Delta < -0.02 = status WARNING (should trigger investigation or retraining pipeline).