Cloud Cost Anomaly Detector

Build an AI-powered API that detects and explains AWS cost anomalies.

Introduction

30 Second Summary

You check your cloud bill on Monday morning and something looks off. One service spiked to six times its usual cost overnight, but the spike is buried in a spreadsheet alongside hundreds of normal line items.

In this project, you will build a Cloud Cost Anomaly Detector API with FastAPI that scans sample AWS cost data, flags unusual spending using statistical analysis, and calls Claude to explain each anomaly with structured, actionable recommendations.

What You'll Build

You'll POST cost data to your running API and get back exactly three flagged anomalies, each with a Claude-generated severity rating, cause analysis, and recommended action plan.

By the end of this project, you'll have:

A detection endpoint where you POST cost data and get back statistically flagged anomalies with per-service Z-scores, catching spending spikes that a fixed dollar threshold completely misses.
An AI analysis endpoint that returns a validated severity rating, plain-English explanation, recommended action, and estimated cost impact for every detected anomaly.
A Docker container that packages the entire API for portable deployment, runnable on any machine with a single command.
Secret Mission: Add a /report endpoint that generates a full cost health report with per-service summaries and severity distribution.

Are there any prerequisites?

This project assumes you have Python 3.12 or higher and Docker Desktop installed on your machine. You will also need an Anthropic API key for the Claude integration.

Before We Start

Before you start coding, take a moment to set your intention for this project. Knowing what you're building and why helps you stay oriented as the pieces come together.

Set Up the Project and Sample Data

Before you can detect anomalies, you need a running API loaded with realistic cost data. Real cloud cost data comes from services like AWS Cost Explorer, which exports daily spending per service.

In this step, you'll scaffold the project, install dependencies, and load 40 sample cost records that mirror the structure of a real AWS export. By the end, your FastAPI server will be running with a health endpoint that confirms all 40 records loaded successfully.

In this step, get ready to:

Create the project directory, virtual environment, and install dependencies.
Create sample AWS cost data with 40 records across four services.
Build the FastAPI skeleton with a health endpoint that loads and validates the data.

Create the project directory and install dependencies

You need Python 3.12 or higher for this project. The dependencies rely on features introduced in Python 3.12.

Check your Python version by running this command:

python3 --version

✔️ I see version 3.12 or higher

You're all set. Move on to the next instruction below.

ⓧ I see an older version

Your system's default Python is older than 3.12. On macOS, the system Python is often outdated even after installing a newer version.

Install Python 3.12 or newer from the official downloads page.
After installing, try python3.12 --version to confirm the new version is available. If your system's python3 still points to the old version, use python3.12 everywhere this project says python3.

ⓧ Command not found

Python is not installed on your machine yet.

Download and install Python 3.12 or newer from the official downloads page.
Once installed, restart your terminal and run python3 --version again to confirm it prints 3.12 or higher.

Move to your Desktop by running this command:

cd ~/Desktop

Create your project folder and move into it by running these commands:

mkdir cost-anomaly-detector
cd cost-anomaly-detector

Your terminal prompt now shows cost-anomaly-detector as the current directory.

A virtual environment keeps this project's packages isolated from your system Python. This prevents version conflicts between projects.

Create and activate a virtual environment by running these commands:

macOS

python3 -m venv .venv
source .venv/bin/activate

Windows

python3 -m venv .venv
.venv\Scripts\activate

Your terminal prompt now starts with (.venv), confirming the virtual environment is active.

Don't see (.venv) in your prompt?

Make sure you ran the source (macOS) or .venv\Scripts\activate (Windows) command, not just the python3 -m venv line. The first command creates the environment. The second activates it.

help me activate my Python virtual environment

Install the four packages your project needs by running this command:

pip install fastapi uvicorn anthropic pydantic

You should see output ending with Successfully installed followed by the list of installed packages.

What are these packages?

fastapi is the web framework for building your API endpoints.
uvicorn is the server that runs your FastAPI application.
anthropic is the official Python SDK for calling Claude AI.
pydantic handles data validation and defines the shape of your request and response objects.

Seeing a pip error?

Make sure your virtual environment is active. You should see (.venv) in your terminal prompt. If not, re-run the activation command from the platform tabs above.

help me fix my pip install error

Create the project's folder structure by running these commands:

mkdir app data
touch app/__init__.py

Confirm the folders were created by running:

ls

You should see app and data listed in the output.

Why the empty init.py?

The app/__init__.py file tells Python that the app folder is a package it can import from. Without it, from app.models import CostRecord would fail.

Now open the project in your code editor.

Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
Type Cursor (or VS Code) and press Enter to open it.
Click File in the top menu bar.
Click Open Folder.
Select the cost-anomaly-detector folder on your Desktop, then click Open.

In the Explorer sidebar on the left, you should see the app and data folders.

Create the sample cost data

Your anomaly detector needs realistic data to analyze. You'll create a sample dataset of 40 daily cost records across four AWS services: EC2, S3, RDS, and Lambda. Three of those services have a hidden spending spike. One is stable.

The file has 40 records, so you'll paste it in two batches.

In the Explorer sidebar, right-click the data folder and select New File.
Name the file cost_data.json and press Enter.
Paste the first batch of cost records (EC2 and S3 data) into the empty file:

[
  {"date": "2024-07-01", "service": "Amazon EC2", "cost": 142.50, "region": "us-east-1"},
  {"date": "2024-07-02", "service": "Amazon EC2", "cost": 148.30, "region": "us-east-1"},
  {"date": "2024-07-03", "service": "Amazon EC2", "cost": 139.80, "region": "us-east-1"},
  {"date": "2024-07-04", "service": "Amazon EC2", "cost": 145.60, "region": "us-east-1"},
  {"date": "2024-07-05", "service": "Amazon EC2", "cost": 151.20, "region": "us-east-1"},
  {"date": "2024-07-06", "service": "Amazon EC2", "cost": 143.90, "region": "us-east-1"},
  {"date": "2024-07-07", "service": "Amazon EC2", "cost": 147.10, "region": "us-east-1"},
  {"date": "2024-07-08", "service": "Amazon EC2", "cost": 892.10, "region": "us-east-1"},
  {"date": "2024-07-09", "service": "Amazon EC2", "cost": 146.70, "region": "us-east-1"},
  {"date": "2024-07-10", "service": "Amazon EC2", "cost": 144.20, "region": "us-east-1"},
  {"date": "2024-07-01", "service": "Amazon S3", "cost": 23.40, "region": "us-east-1"},
  {"date": "2024-07-02", "service": "Amazon S3", "cost": 25.10, "region": "us-east-1"},
  {"date": "2024-07-03", "service": "Amazon S3", "cost": 24.60, "region": "us-east-1"},
  {"date": "2024-07-04", "service": "Amazon S3", "cost": 22.80, "region": "us-east-1"},
  {"date": "2024-07-05", "service": "Amazon S3", "cost": 24.90, "region": "us-east-1"},
  {"date": "2024-07-06", "service": "Amazon S3", "cost": 23.70, "region": "us-east-1"},
  {"date": "2024-07-07", "service": "Amazon S3", "cost": 156.30, "region": "us-east-1"},
  {"date": "2024-07-08", "service": "Amazon S3", "cost": 24.10, "region": "us-east-1"},
  {"date": "2024-07-09", "service": "Amazon S3", "cost": 25.40, "region": "us-east-1"},
  {"date": "2024-07-10", "service": "Amazon S3", "cost": 23.90, "region": "us-east-1"},

Paste the second batch (RDS and Lambda data) directly below the last line, on the very next line:

  {"date": "2024-07-01", "service": "Amazon RDS", "cost": 85.20, "region": "us-west-2"},
  {"date": "2024-07-02", "service": "Amazon RDS", "cost": 87.40, "region": "us-west-2"},
  {"date": "2024-07-03", "service": "Amazon RDS", "cost": 84.90, "region": "us-west-2"},
  {"date": "2024-07-04", "service": "Amazon RDS", "cost": 86.10, "region": "us-west-2"},
  {"date": "2024-07-05", "service": "Amazon RDS", "cost": 85.80, "region": "us-west-2"},
  {"date": "2024-07-06", "service": "Amazon RDS", "cost": 87.20, "region": "us-west-2"},
  {"date": "2024-07-07", "service": "Amazon RDS", "cost": 86.50, "region": "us-west-2"},
  {"date": "2024-07-08", "service": "Amazon RDS", "cost": 85.90, "region": "us-west-2"},
  {"date": "2024-07-09", "service": "Amazon RDS", "cost": 86.80, "region": "us-west-2"},
  {"date": "2024-07-10", "service": "Amazon RDS", "cost": 84.70, "region": "us-west-2"},
  {"date": "2024-07-01", "service": "AWS Lambda", "cost": 12.30, "region": "us-east-1"},
  {"date": "2024-07-02", "service": "AWS Lambda", "cost": 11.80, "region": "us-east-1"},
  {"date": "2024-07-03", "service": "AWS Lambda", "cost": 12.50, "region": "us-east-1"},
  {"date": "2024-07-04", "service": "AWS Lambda", "cost": 11.90, "region": "us-east-1"},
  {"date": "2024-07-05", "service": "AWS Lambda", "cost": 12.10, "region": "us-east-1"},
  {"date": "2024-07-06", "service": "AWS Lambda", "cost": 12.40, "region": "us-east-1"},
  {"date": "2024-07-07", "service": "AWS Lambda", "cost": 11.70, "region": "us-east-1"},
  {"date": "2024-07-08", "service": "AWS Lambda", "cost": 12.20, "region": "us-east-1"},
  {"date": "2024-07-09", "service": "AWS Lambda", "cost": 67.80, "region": "us-east-1"},
  {"date": "2024-07-10", "service": "AWS Lambda", "cost": 12.00, "region": "us-east-1"}
]

Save cost_data.json.
Verify the data loaded correctly by running this command in your terminal:

python3 -c "import json; print(len(json.load(open('data/cost_data.json'))), 'records')"

You should see 40 records printed in the terminal.

What's hiding in this data?

Each service has 10 days of cost data. Most days show normal spending patterns. Three services have a single-day spending spike hidden among the normal days. RDS is the control group with no anomalies.

Your detector's job is to find exactly which days are anomalous. Can you spot the spikes by scanning the data? You'll build code that does it automatically in the next step.

Not seeing 40 records?

Open cost_data.json in your editor and check that the second batch of records sits directly below the first, with no gap or duplicate bracket. The file should start with [ and end with ], with exactly 40 record lines between them.

help me fix my cost_data.json file

Build the FastAPI skeleton and health endpoint

Your API needs a data model that defines what a valid cost record looks like. Pydantic models enforce this shape. If the JSON data has a missing field or a negative cost, Pydantic rejects it at load time instead of letting bad data slip through.

In the Explorer sidebar, right-click the app folder and select New File.
Name the file models.py and press Enter.
Define the CostRecord model by pasting this code into models.py:

from pydantic import BaseModel, Field
from datetime import date


# Define the shape of a single cost record from AWS
class CostRecord(BaseModel):
    date: date
    service: str
    cost: float = Field(gt=0)
    region: str

What does this model do?

CostRecord extends Pydantic's BaseModel, which means any data you pass in gets validated automatically.
Field(gt=0) on the cost field ensures every cost value is greater than zero. A record with a negative cost would be rejected.
The date field accepts ISO 8601 date strings (like 2024-07-01) and converts them to Python date objects.

Save models.py.
Verify the model is importable by running this in your terminal:

python3 -c "from app.models import CostRecord; print('CostRecord ready')"

You should see CostRecord ready printed. This confirms the model imported successfully.

✔️ Awesome, I've got everything!

Great. Double check you saved your file before moving on.

ⓧ I'd like to double check the full code

from pydantic import BaseModel, Field
from datetime import date


class CostRecord(BaseModel):
    date: date
    service: str
    cost: float = Field(gt=0)
    region: str

Now you'll create the main application file. FastAPI uses a lifespan handler to run code once when the server starts. Your lifespan handler will load all 40 cost records from the JSON file and validate each one against the CostRecord model.

In the Explorer sidebar, right-click the app folder and select New File.
Name the file main.py and press Enter.
Add the imports and lifespan handler by pasting this code into main.py:

import json
from pathlib import Path
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.models import CostRecord


# Module-level list to hold validated cost records
cost_data: list[CostRecord] = []


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Load and validate cost data once when the server starts
    global cost_data
    data_path = Path("data/cost_data.json")
    if not data_path.exists():
        raise FileNotFoundError(f"Cost data file not found: {data_path}")
    with open(data_path) as f:
        raw_data = json.load(f)
    # Validate each record against the CostRecord schema
    cost_data = [CostRecord.model_validate(record) for record in raw_data]
    yield

What does the lifespan handler do?

The @asynccontextmanager decorator turns your function into a lifespan handler. Everything before yield runs once at startup. Everything after yield would run at shutdown.
CostRecord.model_validate(record) takes each raw dictionary from the JSON and validates it against the Pydantic model. If any record is missing a field or has a negative cost, the server refuses to start.
The validated records are stored in cost_data, a module-level list that all your endpoints can access.

Add the FastAPI app and health endpoint below the lifespan handler in the same file:



app = FastAPI(
    title="Cloud Cost Anomaly Detector",
    description="Detect and explain unusual AWS spending patterns",
    version="1.0.0",
    lifespan=lifespan,
)


# Health check endpoint to verify the API is running
@app.get("/health")
def health_check():
    return {"status": "healthy", "records_loaded": len(cost_data)}

What does this code do?

FastAPI() creates the application instance. The lifespan=lifespan parameter connects the data-loading function you just wrote so it runs at startup.
The /health endpoint returns the server status and how many records loaded. This gives you a quick way to verify the API is running and the data parsed correctly.

Save main.py.

✔️ Awesome, I've got everything!

Great. Double check you saved your file before moving on.

ⓧ I'd like to double check the full code

Your app/main.py has two sections you built above. Scroll up and confirm both are present in order:

The imports, cost_data list, and the lifespan handler (first chunk).
The app = FastAPI(...) block and the health_check function (second chunk).

Time to see if everything works. Before you run the server, what do you expect the /health endpoint to return if all 40 records loaded successfully?

Start the development server by running this command:

uvicorn app.main:app --reload

You should see output including Uvicorn running on http://127.0.0.1:8000. The --reload flag makes the server restart automatically when you edit files.

Open your browser and go to:

http://localhost:8000/health

You should see this JSON response in your browser:

{"status":"healthy","records_loaded":40}

The records_loaded: 40 confirms that all 40 cost records passed Pydantic validation and are ready for analysis.

Server won't start?

If you see ModuleNotFoundError, make sure your virtual environment is active. You should see (.venv) in your terminal prompt.
If you see FileNotFoundError, check that data/cost_data.json exists and is saved. Run ls data/ to confirm.
Make sure you're running the command from inside the cost-anomaly-detector folder, not from a subfolder like app/.

help me debug my FastAPI server startup

Your API skeleton is up and running with validated cost data. Next up, you'll build a detector that tries to find spending anomalies in that data. The first approach will seem logical, but the results will surprise you.

Build a Naive Fixed-Threshold Detector

Your API skeleton is running and 40 cost records are loaded across four AWS services. Somewhere in that data, three services have hidden spending spikes waiting to be found.

The simplest way to flag unusual spending is a fixed dollar threshold. If any day costs more than a set amount, flag it as an anomaly. In this step, you'll build that detector and see how it performs against real cost data.

In this step, get ready to:

Build a detection module that flags cost records above a dollar threshold.
Wire the detector into a POST /detect endpoint with request and response models.
Test the detector with curl and examine the results.

Create the detection module

The detector's job is straightforward. Take a list of cost records and a dollar threshold, then return every record whose cost exceeds that threshold.

Create a new file called detector.py inside your app folder, then add this code:

from app.models import CostRecord


# Flag any record that exceeds the dollar threshold
def detect_anomalies(records: list[CostRecord], threshold: float = 100.0) -> list[CostRecord]:
    return [record for record in records if record.cost > threshold]

What does this code do?

The detect_anomalies function takes a list of cost records and a threshold (defaulting to $100). It returns every record whose cost is above that threshold using a single list comprehension.

The logic is deliberately simple. If a day costs more than $100, it must be unusual. That's the assumption you're about to test.

Save app/detector.py.
Confirm the file exists by running:

ls app/

You should see detector.py alongside __init__.py, main.py, and models.py in the output.

Don't see detector.py?

Make sure you created the file inside the app/ directory, not in the project root.
Check that the filename is exactly detector.py with no extra spaces or typos.

Why isn't my detector.py file showing up in the app directory?

Wire the detector into a POST /detect endpoint

The detector function exists, but nothing in the API calls it yet. You need request and response models to structure the endpoint's input and output, then a route that connects the pieces.

In app/models.py, add these two new classes below the existing CostRecord class:

class DetectRequest(BaseModel):
    # Default threshold: flag anything over $100
    threshold: float = Field(default=100.0, gt=0)


class DetectResponse(BaseModel):
    total_records: int
    threshold_used: float
    anomalies_found: int
    anomalies: list[CostRecord]

What do these models do?

DetectRequest lets the caller override the dollar threshold. If they don't send one, it defaults to $100.
DetectResponse structures the output so the caller gets the total record count, the threshold that was used, how many anomalies were found, and the full list of flagged records.

Save app/models.py.

✔️ Awesome, I've got everything!

Great. Make sure you've saved app/models.py before moving on.

ⓧ I'd like to double check the full code

Here is the full app/models.py file. Confirm yours matches:

from pydantic import BaseModel, Field
from datetime import date


class CostRecord(BaseModel):
    date: date
    service: str
    cost: float = Field(gt=0)
    region: str


class DetectRequest(BaseModel):
    threshold: float = Field(default=100.0, gt=0)


class DetectResponse(BaseModel):
    total_records: int
    threshold_used: float
    anomalies_found: int
    anomalies: list[CostRecord]

Now update app/main.py to import the new models and the detector function, then add the endpoint that ties them together.

In app/main.py, find the import line near the top of the file:

from app.models import CostRecord

Replace that line with these two lines to import the new models and the detector:

from app.models import CostRecord, DetectRequest, DetectResponse
from app.detector import detect_anomalies

Scroll to the bottom of app/main.py, below the health_check function, and add the /detect endpoint:

@app.post("/detect", response_model=DetectResponse)
def detect_cost_anomalies(request: DetectRequest = DetectRequest()):
    # Run the threshold check against all loaded records
    anomalies = detect_anomalies(cost_data, threshold=request.threshold)
    return DetectResponse(
        total_records=len(cost_data),
        threshold_used=request.threshold,
        anomalies_found=len(anomalies),
        anomalies=anomalies,
    )

What does this endpoint do?

The @app.post("/detect") decorator registers a POST route. The caller can optionally send a JSON body with a custom threshold.
If no body is sent, DetectRequest() provides the default $100 threshold. The response includes the threshold used, total records scanned, and the list of flagged anomalies.

Save app/main.py.
Switch to the terminal where uvicorn is running. You should see a reload message confirming it picked up the changes with no errors.

Seeing an ImportError or ModuleNotFoundError?

Check that app/detector.py exists and is saved. The import in main.py will fail if the file is missing.
Verify that DetectRequest and DetectResponse are defined in app/models.py and that the file is saved.
Confirm the import line in main.py reads from app.models import CostRecord, DetectRequest, DetectResponse with commas between each name.

Why is uvicorn showing an ImportError after I updated main.py?

✔️ Awesome, I've got everything!

Great. Confirm you've saved app/main.py and that uvicorn reloaded without errors.

ⓧ I'd like to double check the full code

Your app/main.py should contain these sections in order:

Import block including CostRecord, DetectRequest, DetectResponse, and detect_anomalies.
The cost_data list and lifespan handler from Step 1.
The FastAPI() app creation and /health endpoint from Step 1.
The new /detect endpoint you just added at the bottom.

Scroll up through the substeps above to confirm each section is present in your file.

Test the detector and count the results

The detector is wired up and ready to run. You know your data has three hidden spending spikes buried across 40 records.

Before you run this, think about it: how many anomalies do you expect a $100 threshold to flag across EC2, S3, RDS, and Lambda?

Open a second terminal tab (keep uvicorn running in the first one).
Hit the /detect endpoint by running:

curl -X POST http://localhost:8000/detect | python3 -m json.tool

Look at the anomalies_found field in the response. You get 11 anomalies. Not 3. Eleven.

Scroll through the flagged records and look at the services. Something is very wrong.

What went wrong?

Look carefully at what the $100 threshold actually did to each service:

EC2 normally costs ~$145/day. Every single day is above $100, so all 10 EC2 records are flagged. The real spike at $892.10 is there, but it's buried among 9 false positives that look identical.
S3 has one spike at $156.30. This one is correctly flagged.
Lambda normally costs ~$12/day, but spiked to $67.80 on July 9. That's 5.6x its normal average. The $100 threshold completely missed it because $67.80 is below $100.
RDS is all ~$86/day with no spikes. It's correctly unflagged, but that's the only service the threshold handles well.

The core problem is that $100 means something completely different for a $145/day service (EC2) than for a $12/day service (Lambda). A fixed threshold drowns high-baseline services in noise while letting real spikes on low-baseline services fly under the radar.

Your detector works exactly as built. The problem is that a fixed dollar threshold treats every service the same, regardless of its normal spending. Next up, you'll replace this approach with one that compares each service's cost to its own baseline.

Fix It with Per-Service Z-Scores

Your fixed-threshold detector found 11 "anomalies" in the last step. Nine of them were false positives, and one real spike was completely invisible.

The core problem is that $100 means something completely different for EC2 (~$145/day baseline) than for Lambda (~$12/day baseline). A Z-score fixes this by comparing each cost to its own service's average, measured in standard deviations. A $67.80 Lambda day is 2.84 standard deviations above Lambda's mean, making it just as statistically unusual as the $892.10 EC2 day.

In this step, get ready to:

Replace the naive fixed-threshold detector with per-service Z-score detection.
Add Pydantic models for anomaly results with Z-score statistics.
Verify the detector catches exactly 3 true anomalies with zero false positives.

Add the Anomaly model to app/models.py

The Z-score detector needs richer output than a raw list of cost records. Each flagged anomaly should carry its Z-score, the service mean, and the service standard deviation so every detection is self-explanatory.

In app/models.py, find the DetectRequest class.
Select everything from DetectRequest to the end of the file and replace it with the following code:

# Holds a flagged anomaly with its statistical context
class Anomaly(BaseModel):
    record: CostRecord
    z_score: float
    service_mean: float
    service_std: float


# Accepts a configurable z-score threshold (default 2.0)
class AnalyzeRequest(BaseModel):
    z_score_threshold: float = Field(default=2.0, ge=1.0, le=4.0)


# Returns anomaly objects instead of raw cost records
class DetectResponse(BaseModel):
    total_records: int
    anomalies_found: int
    anomalies: list[Anomaly]

What does this code do?

Anomaly wraps a CostRecord with its Z-score, the service mean, and the service standard deviation. Anyone reading the response can see exactly why this record was flagged.

AnalyzeRequest replaces DetectRequest. Instead of a dollar threshold, it takes a z_score_threshold defaulting to 2.0 (flagging anything more than 2 standard deviations from the mean).

Save app/models.py.

In your terminal, uvicorn tries to reload and shows an import error. That is expected. app/main.py still imports DetectRequest, which no longer exists. You will fix this after rewriting the detector.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/models.py.

ⓧ I'd like to double check the full code

Here is the full app/models.py file at this point.

from pydantic import BaseModel, Field
from datetime import date


class CostRecord(BaseModel):
    date: date
    service: str
    cost: float = Field(gt=0)
    region: str


class Anomaly(BaseModel):
    record: CostRecord
    z_score: float
    service_mean: float
    service_std: float


class AnalyzeRequest(BaseModel):
    z_score_threshold: float = Field(default=2.0, ge=1.0, le=4.0)


class DetectResponse(BaseModel):
    total_records: int
    anomalies_found: int
    anomalies: list[Anomaly]

Rewrite the detector with per-service Z-scores

The old detector was a one-liner that compared every record against a single dollar amount. The new version groups records by service, calculates each service's mean and standard deviation, then flags any record whose cost sits too many standard deviations away from that service's average.

What is a Z-score?

A Z-score measures how far a value is from the average, in units of standard deviation. A Z-score of 0 means the value equals the mean. A Z-score of 2.0 means the value is 2 standard deviations above the mean.

By computing Z-scores per service, a $67.80 Lambda day (Z-score ~2.84 against Lambda's ~$12 average) gets flagged just as clearly as an $892.10 EC2 day (Z-score ~2.85 against EC2's ~$146 average). The dollar amounts are wildly different, but both are equally unusual for their service.

Replace the entire contents of app/detector.py with this code to set up the imports and record-grouping logic:

import statistics
from collections import defaultdict

from app.models import CostRecord, Anomaly


def detect_anomalies(records: list[CostRecord], threshold: float = 2.0) -> list[Anomaly]:
    # Group all cost records by their service name
    costs_by_service: dict[str, list[CostRecord]] = defaultdict(list)
    for record in records:
        costs_by_service[record.service].append(record)

    anomalies: list[Anomaly] = []

What does this code do?

statistics and defaultdict are Python standard library tools for calculating averages and grouping data.
costs_by_service builds a dictionary where each key is a service name (like "Amazon EC2") and each value is a list of that service's cost records.
The threshold parameter now defaults to 2.0 (standard deviations) instead of 100.0 (dollars).

Add the per-service Z-score calculation loop directly below the anomalies: list[Anomaly] = [] line by copying this code:

    for service, service_records in costs_by_service.items():
        costs = [r.cost for r in service_records]
        # Skip services with too few data points for a meaningful average
        if len(costs) < 3:
            continue

        # Calculate per-service mean and standard deviation
        mean = statistics.mean(costs)
        std = statistics.stdev(costs)
        if std == 0:
            continue

        # Flag records whose z-score exceeds the threshold
        for record in service_records:
            z_score = (record.cost - mean) / std
            if abs(z_score) > threshold:
                anomalies.append(
                    Anomaly(
                        record=record,
                        z_score=round(z_score, 2),
                        service_mean=round(mean, 2),
                        service_std=round(std, 2),
                    )
                )

    return anomalies

What does this code do?

statistics.mean() calculates the arithmetic mean of each service's costs. statistics.stdev() calculates the sample standard deviation.
The Z-score formula (record.cost - mean) / std converts each cost into "how many standard deviations from the service mean."
abs(z_score) > threshold uses the absolute value, so the detector catches costs that are unusually low as well as unusually high.
Each flagged Anomaly carries its Z-score plus the service mean and standard deviation, so anyone reading the result can see the math behind the flag.

Save app/detector.py. The server still shows an error in the terminal because app/main.py has not been updated yet.

✔️ Awesome, I've got everything!

Great. Confirm you have saved app/detector.py and move on to update app/main.py.

ⓧ I'd like to double check the full code

Your app/detector.py should contain two sections, in this order:

The imports (statistics, defaultdict, CostRecord, Anomaly) plus the detect_anomalies function signature and the costs_by_service grouping loop, ending with anomalies: list[Anomaly] = [].
The per-service loop that calculates mean, std, and z_score for each record, appending flagged records to anomalies, then returning the list.

Update the endpoint and verify the fix

The models and detector are ready, but app/main.py still imports DetectRequest and passes a dollar threshold to the detector. Two small edits bring everything into alignment.

In app/main.py, find this import line:

from app.models import CostRecord, DetectRequest, DetectResponse

Replace it with this multi-line import that swaps DetectRequest for AnalyzeRequest:

from app.models import (
    CostRecord,
    AnalyzeRequest,
    DetectResponse,
)

Now scroll down to the /detect endpoint function and find this code:

@app.post("/detect", response_model=DetectResponse)
def detect_cost_anomalies(request: DetectRequest = DetectRequest()):
    anomalies = detect_anomalies(cost_data, threshold=request.threshold)
    return DetectResponse(
        total_records=len(cost_data),
        threshold_used=request.threshold,
        anomalies_found=len(anomalies),
        anomalies=anomalies,
    )

Replace that entire endpoint with this updated version:

@app.post("/detect", response_model=DetectResponse)
def detect_cost_anomalies(request: AnalyzeRequest = AnalyzeRequest()):
    # Pass the z-score threshold to the detector
    anomalies = detect_anomalies(cost_data, threshold=request.z_score_threshold)
    return DetectResponse(
        total_records=len(cost_data),
        anomalies_found=len(anomalies),
        anomalies=anomalies,
    )

What changed?

DetectRequest is now AnalyzeRequest, which takes z_score_threshold instead of a dollar threshold.
The threshold_used field is gone from the response. The Z-score threshold is a tuning knob, not a per-response metric worth echoing back.

Save app/main.py.

Check your terminal. uvicorn should reload cleanly with no errors now that all three files are consistent.

Server still showing an error?

Make sure you saved all three files: app/models.py, app/detector.py, and app/main.py.
Check that app/main.py no longer references DetectRequest or threshold_used. Both names were removed in this step.
Confirm that the import in app/detector.py reads from app.models import CostRecord, Anomaly.

Why is my server still crashing after saving all three files?

Before you test, think for a moment. In Step 2, the fixed $100 threshold returned 11 results. How many do you expect the Z-score detector to return?

Open a new terminal tab and test the updated detector by running:

curl -X POST http://localhost:8000/detect

You should see exactly 3 anomalies in the response:

Amazon EC2 on 2024-07-08 at $892.10 with a Z-score of ~2.85.
Amazon S3 on 2024-07-07 at $156.30 with a Z-score of ~2.90.
AWS Lambda on 2024-07-09 at $67.80 with a Z-score of ~2.84.

Why is this so much better than the fixed threshold?

The fixed $100 threshold flagged 11 records. 9 of those were normal EC2 days that happened to cost more than $100. Meanwhile, Lambda's real spike at $67.80 (5.6x its normal ~$12 average) was invisible because it fell below the threshold.

The Z-score detector flags exactly 3 records. Zero false positives. Zero missed spikes. Amazon RDS is correctly left clean because its spending is stable (~$85-87/day with a max Z-score of ~1.47, well below the 2.0 threshold).

Seeing 0 anomalies or an error?

If you see 0 anomalies, check that app/detector.py uses threshold: float = 2.0 (not 100.0). A threshold of 100 standard deviations would flag nothing.
If you see a validation error in the response, make sure DetectResponse in app/models.py uses anomalies: list[Anomaly] (not list[CostRecord]).

Why does my Z-score detector return the wrong number of anomalies?

✔️ Awesome, I've got everything!

Great. Confirm you have saved all three files.

ⓧ I'd like to double check the full code

Your app/main.py should contain these sections, in this order:

The standard library imports (json, Path, asynccontextmanager) followed by the FastAPI import, then the model import pulling CostRecord, AnalyzeRequest, and DetectResponse.
The cost_data global, the lifespan handler, and the app = FastAPI(...) initialization (unchanged from Step 1).
The /health endpoint (unchanged) and the updated /detect endpoint using AnalyzeRequest and request.z_score_threshold.

Your detector now catches exactly the right anomalies with zero noise. Next up, you will call Claude AI to explain each flagged anomaly with a severity rating, a plain-English explanation, and a recommended action.

Add Structured AI Explanations

Your Z-score detector nails it. Three true anomalies, zero false positives, with per-service statistics backing every flag.

But a raw z-score doesn't tell a cloud team what happened or what to do next. In this step, you'll call Claude AI to explain each anomaly with a severity rating, a plain-English cause, and a recommended action. Claude's structured output API guarantees every response matches a Pydantic schema, so a frontend or alerting system can consume the results without parsing surprises.

In this step, get ready to:

Define Pydantic models for AI-generated anomaly explanations with severity, cause, action, and impact fields.
Create an analyzer module that calls Claude's structured output API for each anomaly.
Add a /analyze endpoint and test the full detection-to-explanation pipeline.

Define the explanation models

A cloud team receiving an anomaly alert needs four things: how severe is this, why is this unusual, what should I do about it, and what happens if I ignore it. Each of those becomes a field in a Pydantic model that Claude must fill out.

In app/models.py, find the blank line after the Anomaly class. Add these two new models above the AnalyzeRequest class:

class AnomalyExplanation(BaseModel):
    severity: str = Field(description="One of: low, medium, high, critical")
    explanation: str = Field(description="Why this spending is anomalous")
    recommended_action: str = Field(description="Specific action to investigate or fix")
    estimated_impact: str = Field(description="Estimated monthly cost impact if unaddressed")


class AnomalyReport(BaseModel):
    anomaly: Anomaly
    ai_analysis: AnomalyExplanation

What do these models do?

AnomalyExplanation defines the exact shape Claude must return. Each field has a description that tells Claude what to write there. When you use structured output, Claude is guaranteed to fill in every field.

AnomalyReport pairs an Anomaly (the statistical detection from your detector) with its AnomalyExplanation (the AI-generated analysis). This is what the API returns to the caller.

At the bottom of app/models.py, below the DetectResponse class, add the response model for the new endpoint:

class AnalyzeResponse(BaseModel):
    total_records: int
    anomalies_found: int
    reports: list[AnomalyReport]

What does this model do?

AnalyzeResponse wraps the full analysis result. It includes a count of total records and anomalies found, plus a list of AnomalyReport objects where each one pairs a detected anomaly with its AI explanation.

Save app/models.py.
Switch to the terminal running uvicorn. You should see a reload message confirming the server picked up your changes.

Server showing an import error?

Check that AnomalyReport is defined after AnomalyExplanation. It references AnomalyExplanation as a field type, so it must come second.
Confirm that AnalyzeResponse is at the very bottom of the file, after DetectResponse. It references AnomalyReport, which must be defined first.
help me fix a model ordering error in my models.py file

✔️ Awesome, I've got everything!

Make sure you've saved app/models.py before moving on.

ⓧ I'd like to double check the full code

Your app/models.py is now over 30 lines. Scroll through the file and confirm these classes exist in this order:

CostRecord (from Step 1).
Anomaly (from Step 3).
AnomalyExplanation (just added).
AnomalyReport (just added).
AnalyzeRequest (from Step 3).
DetectResponse (from Step 3).
AnalyzeResponse (just added).

Create the analyzer module

The Anthropic SDK provides a messages.parse() method that accepts a Pydantic model directly. Instead of getting raw text and hoping Claude's response is parseable, messages.parse() validates the output against your model before returning it. Every response is guaranteed to have exactly the fields you defined.

Create a new file called app/analyzer.py and add this code:

import anthropic

from app.models import Anomaly, AnomalyExplanation


# Initialize the Anthropic client (reads ANTHROPIC_API_KEY from environment)
client = anthropic.Anthropic()


def explain_anomaly(anomaly: Anomaly) -> AnomalyExplanation:
    # Build a prompt with all the anomaly's statistical context
    prompt = (
        f"Analyze this AWS cost anomaly:\n\n"
        f"Service: {anomaly.record.service}\n"
        f"Date: {anomaly.record.date}\n"
        f"Cost: ${anomaly.record.cost:.2f}\n"
        f"Region: {anomaly.record.region}\n"
        f"Z-Score: {anomaly.z_score} "
        f"(mean: ${anomaly.service_mean:.2f}, std: ${anomaly.service_std:.2f})\n\n"
        f"Explain why this is anomalous, assess severity, "
        f"recommend a specific action, and estimate the monthly cost impact if unaddressed."
    )

    # Use structured output to guarantee the response matches our Pydantic model
    parsed = client.messages.parse(
        model="claude-haiku-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
        output_format=AnomalyExplanation,
    )

    return parsed.parsed_output

What does this code do?

anthropic.Anthropic() creates a client that reads your ANTHROPIC_API_KEY environment variable automatically. No need to pass the key in code.
The prompt feeds Claude every piece of statistical context: the service name, date, dollar cost, region, z-score, and the service's mean and standard deviation. This gives Claude enough to reason about severity.
client.messages.parse() with output_format=AnomalyExplanation tells Claude to return its answer as a validated Pydantic object. The .parsed_output attribute is already typed to AnomalyExplanation.

Why structured output instead of raw text?

Without structured output, you would ask Claude for a text response and then try to parse it yourself. If Claude changes wording, adds extra fields, or formats differently, your parser breaks.

With messages.parse(), Claude is constrained to return exactly the fields your Pydantic model defines. The SDK validates the response before you ever see it. Every call returns the same shape, every time.

Save app/analyzer.py.
Confirm the file exists by running:

ls app/

You should see analyzer.py listed alongside __init__.py, detector.py, main.py, and models.py.

Don't see analyzer.py?

Make sure you created the file inside the app/ directory, not in the project root.
Check that the filename is exactly analyzer.py (not analyse.py or analysis.py).
help me find my analyzer.py file

✔️ Awesome, I've got everything!

Make sure you've saved app/analyzer.py before moving on.

ⓧ I'd like to double check the full code

import anthropic

from app.models import Anomaly, AnomalyExplanation


client = anthropic.Anthropic()


def explain_anomaly(anomaly: Anomaly) -> AnomalyExplanation:
    prompt = (
        f"Analyze this AWS cost anomaly:\n\n"
        f"Service: {anomaly.record.service}\n"
        f"Date: {anomaly.record.date}\n"
        f"Cost: ${anomaly.record.cost:.2f}\n"
        f"Region: {anomaly.record.region}\n"
        f"Z-Score: {anomaly.z_score} "
        f"(mean: ${anomaly.service_mean:.2f}, std: ${anomaly.service_std:.2f})\n\n"
        f"Explain why this is anomalous, assess severity, "
        f"recommend a specific action, and estimate the monthly cost impact if unaddressed."
    )

    parsed = client.messages.parse(
        model="claude-haiku-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
        output_format=AnomalyExplanation,
    )

    return parsed.parsed_output

Wire the /analyze endpoint and test

The analyzer module is ready, but nothing calls it yet. You need a new endpoint that runs the Z-score detector, then passes each flagged anomaly to Claude for explanation.

Your analyzer reads the ANTHROPIC_API_KEY environment variable when it loads. Before wiring it into the API, you need to make sure that variable is set in the terminal running your server.

In the terminal running uvicorn, stop the server by pressing Ctrl+C.
Export your Anthropic API key by running:

export ANTHROPIC_API_KEY=your-api-key-here

Replace your-api-key-here with your actual Anthropic API key.

Now update the imports and add the new endpoint. In app/main.py, find the import section at the top of the file:

from fastapi import FastAPI

from app.models import (
    CostRecord,
    AnalyzeRequest,
    DetectResponse,
)
from app.detector import detect_anomalies

Replace that entire import section with this updated version:

from fastapi import FastAPI, HTTPException

from app.models import (
    CostRecord,
    AnalyzeRequest,
    AnalyzeResponse,
    AnomalyReport,
    DetectResponse,
)
from app.detector import detect_anomalies
from app.analyzer import explain_anomaly

What changed in the imports?

HTTPException is added to the FastAPI import. You'll use it to return a 502 error if the AI analysis fails.
AnalyzeResponse and AnomalyReport are the new models you just created.
explain_anomaly is imported from your new analyzer module.

At the bottom of app/main.py, below the detect_cost_anomalies function, add the /analyze endpoint:

@app.post("/analyze", response_model=AnalyzeResponse)
def analyze_costs(request: AnalyzeRequest = AnalyzeRequest()):
    # Run the Z-score detector first
    anomalies = detect_anomalies(cost_data, threshold=request.z_score_threshold)

    if not anomalies:
        return AnalyzeResponse(
            total_records=len(cost_data),
            anomalies_found=0,
            reports=[],
        )

    # Call Claude for each detected anomaly
    reports: list[AnomalyReport] = []
    for anomaly in anomalies:
        try:
            explanation = explain_anomaly(anomaly)
            reports.append(AnomalyReport(anomaly=anomaly, ai_analysis=explanation))
        except Exception as e:
            raise HTTPException(
                status_code=502,
                detail=f"AI analysis failed for {anomaly.record.service} on {anomaly.record.date}: {str(e)}",
            )

    return AnalyzeResponse(
        total_records=len(cost_data),
        anomalies_found=len(anomalies),
        reports=reports,
    )

What does this endpoint do?

It first runs detect_anomalies() to get the Z-score flagged records. If none are found, it returns an empty report immediately.
For each anomaly, it calls explain_anomaly() which sends the anomaly details to Claude and gets back a validated AnomalyExplanation.
The try/except block catches API failures and returns HTTP 502. This tells the caller the AI service is unavailable, not that the request was bad.

Save app/main.py.
Restart the server by running:

uvicorn app.main:app --reload

You should see the server start with no errors. If you see an AuthenticationError, your ANTHROPIC_API_KEY is not set correctly.

Before you test, predict: will the /analyze response look different from the /detect response? What new fields do you expect to see?

Open a second terminal and test the new endpoint by running:

curl -X POST http://localhost:8000/analyze | python3 -m json.tool

Each of the 3 anomalies now has a structured ai_analysis object with exactly four fields: severity, explanation, recommended_action, and estimated_impact. The structure is identical every time you call it, even though the wording varies.

Run the same command again to confirm the response has the same structure with the same four fields.

Why does the wording change but the structure doesn't?

Claude generates the text content of each field fresh every time, so the explanation and recommended action might be phrased differently across requests. But messages.parse() guarantees the response always contains exactly the four fields your Pydantic model defines. The shape is locked. The content is dynamic.

Getting a 502 or connection error?

A 502 response means the Anthropic API call failed. Check that your ANTHROPIC_API_KEY is set in the terminal where uvicorn is running (not in a different terminal).
If you see a connection refused error, make sure uvicorn started without errors and is still running.
If the response takes a long time, that is normal. Claude is analyzing 3 anomalies sequentially, which takes a few seconds.
help me debug my /analyze endpoint

✔️ Awesome, I've got everything!

Make sure you've saved app/main.py before moving on.

ⓧ I'd like to double check the full code

Your app/main.py is now over 30 lines. Scroll through the file and confirm these sections exist in this order:

Imports: json, Path, asynccontextmanager, FastAPI, HTTPException, all models, detect_anomalies, and explain_anomaly.
The cost_data global variable and lifespan function (from Step 1).
The app FastAPI instance (from Step 1).
The /health GET endpoint (from Step 1).
The /detect POST endpoint (from Step 3).
The /analyze POST endpoint (just added).

Your API now detects anomalies statistically and explains them with AI. Next up, you'll package the whole thing into a Docker container so it runs anywhere.

Containerize with Docker

Your API now detects anomalies with Z-scores and explains each one with structured AI analysis. Anyone on your machine can hit /analyze and get validated results. But the moment you close your terminal, the server stops.

A local development server tied to your laptop is not portable or demo-ready. Docker solves this by packaging your entire API, its dependencies, and its data into a single image that runs identically on any machine.

In this step, get ready to:

Create a Dockerfile and supporting configuration files.
Build a Docker image for the cost anomaly detector.
Run and test the containerized API end-to-end.

Create the Dockerfile

A Dockerfile is a recipe that tells Docker how to build an image. It specifies the base environment, installs dependencies, copies your code, and defines how to start the application.

Create a file called Dockerfile (no extension) in the root of your cost-anomaly-detector folder, then paste in the following:

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

What does this Dockerfile do?

FROM python:3.12-slim starts from a lightweight Python 3.12 image. The slim variant keeps the image small by excluding build tools you don't need at runtime.
COPY requirements.txt and RUN pip install happen before COPY . . so that Docker caches the dependency layer. If you change your Python code later, Docker skips the install step on rebuild.
CMD sets the default command. It runs uvicorn on 0.0.0.0 so the API is reachable from outside the container.

Save Dockerfile.
Confirm the file exists by running:

ls Dockerfile

You should see Dockerfile listed in the output.

Don't see Dockerfile?

Make sure the file is named exactly Dockerfile with a capital D and no file extension. Not dockerfile.txt or Dockerfile.dockerfile.
Check that you created the file in the project root (cost-anomaly-detector/), not inside the app/ folder.

Why isn't Docker finding my Dockerfile?

✔️ Awesome, I've got everything!

Great. Make sure you've saved your Dockerfile.

ⓧ I'd like to double check the full code

Here is the complete Dockerfile. Compare it with yours line by line.

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Add build configuration and build the image

Docker needs to know which Python packages to install. A requirements.txt file lists them with pinned versions so the image builds identically every time.

Open requirements.txt in the project root and replace its contents with the following:

fastapi==0.141.1
uvicorn==0.52.4
anthropic==1.0.0
pydantic==2.13.4

Why pin exact versions?

Pinning versions with == ensures that the same versions install inside the container as on your local machine. Without pins, a newer version could install and break your app without any code changes.

Save requirements.txt.

Next, you need a .dockerignore file. It works like .gitignore, telling Docker which files to skip when copying your project into the image.

Create a file called .dockerignore in the project root, then paste in the following:

__pycache__
*.pyc
.venv
venv
.env
.git

What gets excluded?

Your virtual environment (.venv), Python cache files (__pycache__), your .env file containing your API key, and .git history all stay out of the image. This keeps the image small and avoids leaking secrets.

Save .dockerignore.

Now you have all three files Docker needs. Time to build the image.

Switch back to the terminal where uvicorn is running.
Press Ctrl+C to stop the development server.
Build the Docker image by running this command:

docker build -t cost-anomaly-detector .

What does this command do?

docker build reads your Dockerfile and executes each instruction to create an image.
-t cost-anomaly-detector tags the image with a name so you can reference it later.
The trailing . tells Docker to use the current directory as the build context.

The build takes a minute or two as Docker downloads the Python base image and installs your packages. You should see each step complete with output like Step 1/7 through Step 7/7.

Confirm the image was created by running:

docker images cost-anomaly-detector

You should see one row listing cost-anomaly-detector with the latest tag and a size around 200-300 MB.

Build failed?

Make sure Docker Desktop is running. You can check by looking for the Docker whale icon in your system tray (Windows) or menu bar (macOS).
If you see a pip error about a package not found, double-check that requirements.txt lists anthropic (not a misspelling). Package names must match exactly.
Make sure you are running the command from the cost-anomaly-detector folder, where Dockerfile lives.

Why did my Docker build fail?

Run and test the containerized API

The image is built. Now you will run a container from it and confirm the full detection and analysis pipeline works from inside Docker.

Start the container by running this command:

docker run -p 8000:8000 -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY cost-anomaly-detector

What do these flags do?

-p 8000:8000 maps port 8000 on your machine to port 8000 inside the container. Without this, the API would be unreachable from your browser or curl.
-e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY passes your local API key into the container as an environment variable. The Anthropic SDK inside the container reads it automatically.

You should see uvicorn startup output from inside the container. The terminal is now attached to the container, so you need a second terminal to send requests.

Open a new terminal window or tab.

Before you send a request, what do you think the /health endpoint will return? Will it show 40 records loaded, or will the container fail to find the data file?

Test the health endpoint by running:

curl http://localhost:8000/health

You should see the same healthy response as before, now served from inside the container.

{"status":"healthy","records_loaded":40}

The data file loaded successfully because the COPY . . instruction in your Dockerfile copied the data/ folder into the image along with your application code.

Now for the real test. Do you expect the full AI analysis pipeline to work from inside the container?

Test the analyze endpoint by running:

curl -X POST http://localhost:8000/analyze | python3 -m json.tool

You should see the same 3 anomalies, each with a structured ai_analysis object containing severity, explanation, recommended_action, and estimated_impact. The entire pipeline is now running from inside a Docker container.

Seeing connection refused or an API error?

If curl reports connection refused, make sure the first terminal still shows the container running. If it exited, check the logs for an error. A common cause is a missing or misspelled ANTHROPIC_API_KEY environment variable in your shell.
If you see a 502 error, the API key was not passed correctly. Verify the variable is set in your current shell by running echo $ANTHROPIC_API_KEY. If it's empty, export it again before re-running the docker run command.
If port 8000 is already in use, make sure you stopped the local uvicorn server with Ctrl+C before starting the container.

Why isn't my Docker container responding?

Secret mission

Add a Cost Health Report Endpoint

Your API detects anomalies and explains them one at a time. A real cloud team wants a single dashboard-ready report: total spend, per-service summaries ranked by impact, and a severity breakdown. Build a GET /report endpoint that aggregates everything into one structured response.

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. The Claude API usage (~$0.02-0.05) was a one-time cost that has already been charged.

Decide whether to keep your resources available, pause them to come back later, or delete them entirely.

Resources you used:

Docker image cost-anomaly-detector stored locally (no ongoing cost).
Project directory cost-anomaly-detector on your Desktop (includes source code, sample data, and virtual environment).

Keep everything running

No action needed. Choose this if you want to keep experimenting with the API or extend it further.

The Docker image stays on your machine at no cost.
Re-run the container anytime with the same docker run command from Step 5.
Your project files remain in cost-anomaly-detector on your Desktop.

Pause - I'll come back to this later

Stop any running containers to free up system resources. Your Docker image and project files stay in place, ready to pick up where you left off.

Press Ctrl+C in the terminal where your container is running to stop it.
Your Docker image and project directory remain on your machine at no cost.

Delete - I don't want to use this again

Remove all project resources and start fresh.

Stop the running container by pressing Ctrl+C in its terminal window.
Remove the Docker image by running:

docker rmi cost-anomaly-detector

Remove the project directory from your Desktop.

macOS

Run this command in your terminal:

rm -rf ~/Desktop/cost-anomaly-detector

Windows

Run this command in PowerShell:

Remove-Item -Recurse -Force ~\Desktop\cost-anomaly-detector

Image removal failed?

If docker rmi reports the image is in use by a stopped container, list the containers first:

docker ps -a --filter ancestor=cost-anomaly-detector

Copy the container ID from the output and remove it:

docker rm [[CONTAINER_ID="your container ID from the list above"]]

Then retry the docker rmi cost-anomaly-detector command.

Nice Work!

Nice work! You just built a production-style Cloud Cost Anomaly Detector that combines statistical analysis with AI-powered explanations, all running inside a Docker container.

You've learned how to:

Build a naive fixed-threshold detector, see it produce 9 false positives and miss a real anomaly, then replace it with per-service Z-score detection that correctly flags exactly 3 true anomalies.
Use Claude's structured output API via messages.parse() with a Pydantic model to generate validated AI explanations with severity, cause, action, and impact for each anomaly.
Design a FastAPI API with health checks, detection, and AI-powered analysis endpoints, then containerize the entire application with Docker for portable, demo-ready deployment.
Secret Mission: Added a /report endpoint that aggregates anomalies into a per-service cost health summary with severity distribution.

Ready to quiz yourself?

Cloud Cost Anomaly Detector

Introduction

30 Second Summary

What You'll Build

Are there any prerequisites?

Before We Start

Set Up the Project and Sample Data

In this step, get ready to:

Create the project directory and install dependencies

✔️ I see version 3.12 or higher

ⓧ I see an older version

ⓧ Command not found

macOS

Windows

Don't see (.venv) in your prompt?

What are these packages?

Seeing a pip error?

Why the empty __init__.py?

Create the sample cost data

What's hiding in this data?

Not seeing 40 records?

Build the FastAPI skeleton and health endpoint

What does this model do?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

What does the lifespan handler do?

What does this code do?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Server won't start?

Build a Naive Fixed-Threshold Detector

In this step, get ready to:

Create the detection module

What does this code do?

Don't see detector.py?

Wire the detector into a POST /detect endpoint

What do these models do?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

What does this endpoint do?

Seeing an ImportError or ModuleNotFoundError?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Test the detector and count the results

What went wrong?

Fix It with Per-Service Z-Scores

In this step, get ready to:

Add the Anomaly model to app/models.py

What does this code do?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Rewrite the detector with per-service Z-scores

What is a Z-score?

What does this code do?

What does this code do?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Update the endpoint and verify the fix

What changed?

Server still showing an error?

Why is this so much better than the fixed threshold?

Seeing 0 anomalies or an error?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Add Structured AI Explanations

In this step, get ready to:

Define the explanation models

What do these models do?

What does this model do?

Server showing an import error?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Create the analyzer module

What does this code do?

Why structured output instead of raw text?

Don't see analyzer.py?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Wire the /analyze endpoint and test

What changed in the imports?

Why the empty init.py?