Azure FinOps GenAI Cost Pipeline
Build a Python pipeline that detects GenAI cost anomalies in Azure billing data.
Introduction
30 Second Summary
Imagine your company's AI spending tripled overnight and nobody noticed until the bill arrived. As organizations adopt generative AI, tracking these unpredictable costs alongside traditional cloud spend has become a critical engineering challenge.
In this project, you will build a Python data pipeline that pulls real Azure billing data, generates realistic LLM token usage logs, merges everything into an industry-standard cost format, and automatically detects spending anomalies.
What You'll Build
You'll create a local Python FinOps data pipeline that connects to Azure Cost Management, produces simulated API gateway token logs, and runs statistical anomaly detection to surface cost spikes.
By the end of this project, you'll have:
- A working Python script that authenticates to Azure and queries the Cost Management API for real billing data.
- A simulated LLM token usage dataset with realistic patterns and injected cost spikes modeled on Azure API Management gateway logs.
- A merged FOCUS-format dataset with rolling Z-score anomaly detection that flags runaway GenAI spend.
- Secret Mission: Tag-based cost allocation that splits shared GenAI costs across business units and outputs a chargeback summary.
Are there any prerequisites?
No prior experience with Azure or FinOps is required. This guide walks you through setting up every tool from scratch, including Python, VS Code, and an Azure free account with $200 credit for 30 days.
Before We Start
Building a FinOps data pipeline requires several tools working together. You need Python to write the pipeline scripts, the Azure CLI to authenticate with your Azure account, and VS Code as your code editor.
In this step, you will install and configure all of these tools so you have a clean, ready-to-go development environment for the rest of the project.
In this step, get ready to:
- Install Python 3.11 and VS Code.
- Install the Azure CLI and create an Azure free account.
- Create a project folder with a virtual environment and install dependencies.
Install Python 3.11 and VS Code
This project requires Python 3.11 or higher. Some of the libraries we use need at least this version to work correctly.
- Open a terminal and check your current Python version by running this command:
macOS
python3 --versionWindows
python --version✔️ I see version 3.11 or higher
You are all set with Python. Move on to installing VS Code below.
ⓧ I see an older version
You have Python installed, but it is an older version. You need to install Python 3.11 alongside it.
- Go to the Python 3.11 downloads page.
- On Windows, download the Windows installer (64-bit). On macOS, download the macOS 64-bit universal2 installer.
- Run the installer. On Windows, check the box labeled Add Python to PATH before clicking Install Now.
- Close and reopen your terminal after installation.
- Run the version check command again. You should now see 3.11 or higher.
macOS path shadowing
macOS may still resolve the older system Python after installing 3.11. Use python3.11 --version to confirm the new version is available. If it is, use python3.11 instead of python3 throughout this project.
ⓧ Command not found
Python is not installed yet. Follow these steps to install Python 3.11.
- Go to the Python 3.11 downloads page.
- On Windows, download the Windows installer (64-bit). On macOS, download the macOS 64-bit universal2 installer.
- Run the installer. On Windows, check the box labeled Add Python to PATH before clicking Install Now.
- Close and reopen your terminal after installation.
- Run the version check command again. You should now see 3.11 or higher.
Now install VS Code, the code editor you will use throughout this project.
- Go to code.visualstudio.com and download the installer for your operating system.
- Run the installer and follow the prompts to complete setup.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Visual Studio Code and press Enter to open it.
VS Code not appearing in search?
On Windows, the installer may ask if you want to add VS Code to your PATH. Make sure this is checked. If you missed it, restart your computer and try again.
On macOS, drag the VS Code app into your Applications folder if you downloaded the .zip version.
help me open VS Code on my machine
Install the Azure CLI and set up your Azure account
The Azure CLI lets you authenticate and manage Azure resources from your terminal. You will use it to connect your Python scripts to Azure Cost Management.
- Check if the Azure CLI is already installed by running this command in your terminal:
az --version✔️ I see a version number
The Azure CLI is already installed. Move on to creating your Azure account below.
ⓧ Command not found
The Azure CLI is not installed yet. Follow the instructions for your operating system.
macOS
- Install the Azure CLI using Homebrew by running:
brew install azure-cliIf you do not have Homebrew, visit the Azure CLI install page for alternative methods.
Windows
- Install the Azure CLI using the Windows Package Manager by running this in PowerShell or Command Prompt:
winget install -e --id Microsoft.AzureCLIIf WinGet is not available, download the MSI installer from the Azure CLI install page.
- Close and reopen your terminal after installation.
- Run az --version again to confirm the install was successful. You should see a version number in the output.
Now create an Azure free account so your pipeline can query real billing data.
- Go to azure.microsoft.com/free and click Start free.
- Sign in with a Microsoft or GitHub account.
- Complete the signup form. You will need a phone number and a credit or debit card (non-prepaid) for identity verification.
- Once signup is complete, sign into Azure from your terminal by running:
az loginWhat does az login do?
This command opens your browser and asks you to sign in with your Azure credentials. Once you authenticate, the Azure CLI stores a token locally so your Python scripts can access Azure services without re-entering credentials.
What does the Azure free account include?
The free account gives you $200 of credit valid for 30 days. No charges occur unless you explicitly deploy paid services beyond the free tier.
az login opens a browser but nothing happens?
Try running az login --use-device-code instead. This gives you a code to enter manually at microsoft.com/devicelogin.
help me sign into Azure through the CLI
Create your project folder and install dependencies
With your tools installed, you can now create a dedicated folder for the project and set up an isolated Python environment.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir azure-finops-pipeline
cd azure-finops-pipeline- Confirm the folder exists by running:
lsYou should see the azure-finops-pipeline folder listed in the output (if you ran ls from the Desktop) or an empty directory listing (if you ran it from inside the folder).
Next, create a virtual environment to isolate this project's packages from your system Python.
- Create the virtual environment by running:
macOS
python3 -m venv .venvWindows
python -m venv .venvWhat is a virtual environment?
A virtual environment is an isolated Python installation inside your project folder. Packages you install here stay separate from your system Python, preventing version conflicts between projects.
- Activate the virtual environment by running:
macOS
source .venv/bin/activateWindows
.venv\Scripts\activateYou should see (.venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.
Activation failed or no (.venv) prefix?
On Windows, if you see an execution policy error, run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned in PowerShell first, then try activating again.
On macOS, make sure you used source before the path. Without it, the command will not work.
help me activate my Python virtual environment
Now create the requirements.txt file that lists all the Python packages this project needs.
- In VS Code, click File in the top menu bar.
- Click Open Folder.
- Select the azure-finops-pipeline folder on your Desktop, then click Open.
- Create a new file by pressing Cmd+N (macOS) or Ctrl+N (Windows).
- Paste the following dependency list into the file:
azure-identity==1.25.3
azure-mgmt-costmanagement==4.0.1
pandas==2.3.3
numpy==1.26.4What are these packages?
azure-identity handles authentication to Azure services. azure-mgmt-costmanagement is the SDK for querying Azure billing data.
pandas is a data manipulation library for working with tabular datasets. numpy provides the math operations needed for anomaly detection.
- Save the file as requirements.txt by pressing Cmd+S (macOS) or Ctrl+S (Windows). Type requirements.txt as the filename and save it in the project root.
- Install all the dependencies by running this command in your terminal (make sure you see (.venv) in the prompt):
pip install -r requirements.txtYou should see pip downloading and installing four packages. When it finishes, you will see a success message confirming all packages were installed.
Seeing install errors?
Make sure your virtual environment is active (you see (.venv) in your prompt). If you closed your terminal, navigate back to the project folder and activate the environment again before running pip install.
If you see a permissions error on macOS, confirm you are inside the virtual environment rather than using the system pip.
help me fix pip install errors in my virtual environment
✔️ Awesome, I've got everything!
Great. Double check you have saved requirements.txt and that all four packages installed without errors.
ⓧ I'd like to double check the full code
Here is the complete requirements.txt file for comparison:
azure-identity==1.25.3
azure-mgmt-costmanagement==4.0.1
pandas==2.3.3
numpy==1.26.4Your environment is ready. Next up, you will write a script that connects to Azure Cost Management and pulls real billing data.
Query Azure Cost Management
Your environment is ready with Python, the Azure CLI, and all dependencies installed. Now it's time to connect to Azure and pull real billing data.
Every FinOps pipeline starts with a data source. In this step, you will write a Python script that authenticates to Azure Cost Management and queries your subscription's spending. If your account is brand new and has no billing history, the script will gracefully fall back to realistic synthetic data so you can keep building.
In this step, get ready to:
- Write a script that authenticates to Azure and queries the Cost Management API.
- Build a QueryDefinition that requests daily cost data grouped by service.
- Add a synthetic fallback so the pipeline works even on brand-new accounts.
Create the file and set up authentication
Before you can query billing data, the script needs to prove who you are. DefaultAzureCredential automatically picks up your az login session, so there's no extra configuration needed.
- In VS Code, create a new file called azure_costs.py in your project folder.
- Add the imports and the subscription ID helper by pasting this code:
"""Query Azure Cost Management API for billing data."""
import pandas as pd
from azure.identity import DefaultAzureCredential
from azure.mgmt.costmanagement import CostManagementClient
from azure.mgmt.costmanagement.models import (
QueryDefinition,
QueryDataset,
QueryTimePeriod,
QueryAggregation,
QueryGrouping,
)
from datetime import datetime, timedelta
import json
def get_subscription_id():
"""Get subscription ID from Azure CLI login context."""
import subprocess
result = subprocess.run(
["az", "account", "show", "--query", "id", "-o", "tsv"],
capture_output=True,
text=True,
)
sub_id = result.stdout.strip()
if not sub_id:
raise RuntimeError("No Azure subscription found. Run 'az login' first.")
return sub_idWhat does this code do?
The imports pull in the Azure SDK classes you need to authenticate and query costs. DefaultAzureCredential automatically uses whatever login method is available (your az login session, environment variables, or managed identity).
The get_subscription_id() function shells out to az account show --query id -o tsv to grab your active subscription ID. If it returns empty (you're not logged in), the function raises an error with a helpful message.
- Save azure_costs.py.
- You won't see a change yet. The next chunk adds the function that actually queries Azure.
Seeing red squiggles under the imports?
Make sure your virtual environment is activated (you should see (.venv) in your terminal prompt). VS Code also needs to be using the .venv interpreter. Press Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows) and select Python: Select Interpreter, then choose the .venv option.
help me fix import errors in my Azure Python project
Build the cost query and parse the response
The Cost Management API accepts a structured query definition that specifies what data you want. You will request actual (not amortized) costs for the current month, grouped by service and resource.
- In azure_costs.py, add the query_azure_costs() function below get_subscription_id():
def query_azure_costs(subscription_id):
"""Query Azure Cost Management for month-to-date costs."""
credential = DefaultAzureCredential()
client = CostManagementClient(credential=credential)
scope = f"/subscriptions/{subscription_id}"
query_def = QueryDefinition(
type="ActualCost",
timeframe="MonthToDate",
dataset=QueryDataset(
granularity="Daily",
aggregation={
"totalCost": QueryAggregation(name="Cost", function="Sum"),
},
grouping=[
QueryGrouping(type="Dimension", name="ServiceName"),
QueryGrouping(type="Dimension", name="ResourceId"),
],
),
)
try:
result = client.query.usage(scope=scope, parameters=query_def)
if result and result.rows:
columns = [col.name for col in result.columns]
df = pd.DataFrame(result.rows, columns=columns)
print(f"[LIVE] Retrieved {len(df)} cost rows from Azure.")
return df
else:
print("[INFO] No cost data yet (new account). Using synthetic fallback.")
return None
except Exception as e:
print(f"[WARN] Could not query Azure: {e}")
print("[INFO] Using synthetic fallback data.")
return NoneWhat does this code do?
QueryDefinition tells the API you want ActualCost data (real spend, not amortized reservations) for the current month. The dataset groups results by ServiceName and ResourceId so you can see which services cost the most.
The client.query.usage() call sends this query to Azure at subscription scope. If the response has rows, they get parsed into a pandas DataFrame. If the account is too new to have data (or the API call fails), the function returns None so the caller can fall back gracefully.
- Save azure_costs.py.
- You won't see a change yet. The next chunk adds the fallback data and ties everything together.
Not sure about the scope format?
The scope must be /subscriptions/{subscriptionId} for subscription-level queries. The get_subscription_id() function grabs this ID automatically from your az login session. If you need to target a specific resource group instead, the scope would be /subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}.
help me understand Azure Cost Management API scopes
Add a synthetic fallback and run the pipeline
A brand-new Azure free account has no billing data for the first day or two. Instead of blocking the project, you will add a function that generates realistic synthetic cost data matching the same schema.
- In azure_costs.py, add the generate_synthetic_costs() function below query_azure_costs():
def generate_synthetic_costs():
"""Generate realistic synthetic Azure cost data as fallback."""
import numpy as np
np.random.seed(42)
days = 30
base_date = datetime.now() - timedelta(days=days)
services = [
("Microsoft.Compute", "Virtual Machines", 2.50),
("Microsoft.Storage", "Storage Accounts", 0.30),
("Microsoft.CognitiveServices", "Azure OpenAI", 5.00),
("Microsoft.Network", "Virtual Network", 0.10),
("Microsoft.Sql", "SQL Database", 1.80),
]
rows = []
for day_offset in range(days):
date = base_date + timedelta(days=day_offset)
for service_name, resource_name, base_cost in services:
daily_cost = base_cost * np.random.uniform(0.7, 1.3)
rows.append({
"UsageDate": date.strftime("%Y-%m-%d"),
"ServiceName": service_name,
"ResourceId": f"/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/rg-finops/providers/{service_name}/{resource_name.lower().replace(' ', '')}",
"Cost": round(daily_cost, 4),
"Currency": "USD",
})
df = pd.DataFrame(rows)
print(f"[SYNTHETIC] Generated {len(df)} cost rows.")
return dfWhat does this code do?
This function creates 30 days of fake-but-realistic cost data for five Azure services. Each service has a base daily cost that varies randomly between 70% and 130% to simulate real fluctuations.
The output matches the same column schema (UsageDate, ServiceName, ResourceId, Cost, Currency) that live Azure data would return. This means the rest of your pipeline can process either source identically.
- Now add the main() function and entry point at the bottom of azure_costs.py:
def main():
print("=" * 60)
print("Azure Cost Management - Data Ingestion")
print("=" * 60)
try:
subscription_id = get_subscription_id()
print(f"Subscription: {subscription_id}")
df = query_azure_costs(subscription_id)
except RuntimeError as e:
print(f"[WARN] {e}")
df = None
if df is None:
df = generate_synthetic_costs()
print(f"\nDataset shape: {df.shape}")
print(f"\nFirst 5 rows:")
print(df.head())
print(f"\nCost summary by service:")
if "ServiceName" in df.columns:
print(df.groupby("ServiceName")["Cost"].sum().sort_values(ascending=False))
return df
if __name__ == "__main__":
main()What does main() do?
main() orchestrates the full ingestion flow. It tries to get live data from Azure first. If that fails (no subscription or no data yet), it falls back to the synthetic generator.
Either way, you get a DataFrame printed with its shape, first rows, and a cost summary grouped by service name. This pattern of "try live, fall back to synthetic" is common in production data pipelines.
- Save azure_costs.py.
- Run the script by entering this command in your terminal:
python azure_costs.pyYou should see a banner, followed by either [LIVE] data from Azure or [SYNTHETIC] fallback data with 150 rows. Both paths print the DataFrame shape, first 5 rows, and a cost summary by service.
Seeing a RuntimeError about no subscription?
This means az login has expired or wasn't completed. Run az login again in your terminal, complete the browser authentication, then re-run the script.
If you see [WARN] Could not query Azure followed by [SYNTHETIC] output, that is expected behavior for new accounts. The script is working correctly.
help me fix Azure authentication errors in my Python script
✔️ Awesome, I've got everything!
Great. Double check you have saved azure_costs.py and that it ran without crashing.
ⓧ I'd like to double check the full code
Here is the complete azure_costs.py file. Compare yours against this:
"""Query Azure Cost Management API for billing data."""
import pandas as pd
from azure.identity import DefaultAzureCredential
from azure.mgmt.costmanagement import CostManagementClient
from azure.mgmt.costmanagement.models import (
QueryDefinition,
QueryDataset,
QueryTimePeriod,
QueryAggregation,
QueryGrouping,
)
from datetime import datetime, timedelta
import json
def get_subscription_id():
"""Get subscription ID from Azure CLI login context."""
import subprocess
result = subprocess.run(
["az", "account", "show", "--query", "id", "-o", "tsv"],
capture_output=True,
text=True,
)
sub_id = result.stdout.strip()
if not sub_id:
raise RuntimeError("No Azure subscription found. Run 'az login' first.")
return sub_id
def query_azure_costs(subscription_id):
"""Query Azure Cost Management for month-to-date costs."""
credential = DefaultAzureCredential()
client = CostManagementClient(credential=credential)
scope = f"/subscriptions/{subscription_id}"
query_def = QueryDefinition(
type="ActualCost",
timeframe="MonthToDate",
dataset=QueryDataset(
granularity="Daily",
aggregation={
"totalCost": QueryAggregation(name="Cost", function="Sum"),
},
grouping=[
QueryGrouping(type="Dimension", name="ServiceName"),
QueryGrouping(type="Dimension", name="ResourceId"),
],
),
)
try:
result = client.query.usage(scope=scope, parameters=query_def)
if result and result.rows:
columns = [col.name for col in result.columns]
df = pd.DataFrame(result.rows, columns=columns)
print(f"[LIVE] Retrieved {len(df)} cost rows from Azure.")
return df
else:
print("[INFO] No cost data yet (new account). Using synthetic fallback.")
return None
except Exception as e:
print(f"[WARN] Could not query Azure: {e}")
print("[INFO] Using synthetic fallback data.")
return None
def generate_synthetic_costs():
"""Generate realistic synthetic Azure cost data as fallback."""
import numpy as np
np.random.seed(42)
days = 30
base_date = datetime.now() - timedelta(days=days)
services = [
("Microsoft.Compute", "Virtual Machines", 2.50),
("Microsoft.Storage", "Storage Accounts", 0.30),
("Microsoft.CognitiveServices", "Azure OpenAI", 5.00),
("Microsoft.Network", "Virtual Network", 0.10),
("Microsoft.Sql", "SQL Database", 1.80),
]
rows = []
for day_offset in range(days):
date = base_date + timedelta(days=day_offset)
for service_name, resource_name, base_cost in services:
daily_cost = base_cost * np.random.uniform(0.7, 1.3)
rows.append({
"UsageDate": date.strftime("%Y-%m-%d"),
"ServiceName": service_name,
"ResourceId": f"/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/rg-finops/providers/{service_name}/{resource_name.lower().replace(' ', '')}",
"Cost": round(daily_cost, 4),
"Currency": "USD",
})
df = pd.DataFrame(rows)
print(f"[SYNTHETIC] Generated {len(df)} cost rows.")
return df
def main():
print("=" * 60)
print("Azure Cost Management - Data Ingestion")
print("=" * 60)
try:
subscription_id = get_subscription_id()
print(f"Subscription: {subscription_id}")
df = query_azure_costs(subscription_id)
except RuntimeError as e:
print(f"[WARN] {e}")
df = None
if df is None:
df = generate_synthetic_costs()
print(f"\nDataset shape: {df.shape}")
print(f"\nFirst 5 rows:")
print(df.head())
print(f"\nCost summary by service:")
if "ServiceName" in df.columns:
print(df.groupby("ServiceName")["Cost"].sum().sort_values(ascending=False))
return df
if __name__ == "__main__":
main()Your cost ingestion script is working. Next up, you will generate realistic LLM token usage logs that simulate what an Azure API Management gateway would produce.
Generate LLM Token Usage Logs
Your pipeline can now pull Azure billing data. The next challenge is tracking GenAI costs specifically.
In production, Azure API Management acts as a gateway in front of Azure OpenAI and logs every LLM request to the ApiManagementGatewayLlmLog table, capturing prompt tokens, completion tokens, and model deployment details. In this step, you will build a script that generates simulated versions of these gateway logs so your pipeline has GenAI cost data to work with.
In this step, get ready to:
- Write a token log generator that simulates 30 days of LLM gateway traffic.
- Inject three distinct anomaly patterns that represent real-world cost spikes.
- Run the script and verify summary statistics showing cost breakdowns by model.
Create the token log generator structure
Each simulated log record mirrors the fields that Azure API Management captures: timestamp, model deployment, token counts, caller IP, and subscription ID. You will also compute an estimated cost per request using token pricing.
- In VS Code, create a new file called token_logs.py in your project folder (the same folder that contains azure_costs.py).
- Add the imports and pricing constants by pasting this code:
"""Generate simulated LLM gateway token usage logs."""
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
# Pricing per 1K tokens (simulated Azure OpenAI rates)
MODEL_PRICING = {
"gpt-4o": {"prompt": 0.005, "completion": 0.015},
"gpt-4o-mini": {"prompt": 0.00015, "completion": 0.0006},
}
BUSINESS_UNITS = ["Engineering", "Marketing", "Data Science", "Product"]What does this code do?
MODEL_PRICING holds simulated per-1K-token rates for two model tiers. The expensive model (gpt-4o) costs 100x more per completion token than the budget model (gpt-4o-mini). This price gap is what makes model-choice anomalies so impactful.
BUSINESS_UNITS tags each request with a team label. In production, this would come from a subscription key or Microsoft Entra ID identity passed through the gateway.
- Below the constants, add the generate_token_logs() function that builds 30 days of simulated traffic:
def generate_token_logs(days=30, records_per_day=70):
"""Generate realistic API Management gateway LLM token logs."""
np.random.seed(123)
base_date = datetime.now() - timedelta(days=days)
logs = []
for day_offset in range(days):
date = base_date + timedelta(days=day_offset)
is_weekday = date.weekday() < 5
daily_records = records_per_day if is_weekday else int(records_per_day * 0.3)
for _ in range(daily_records):
hour = np.random.choice(
range(24),
p=_business_hour_distribution(),
)
timestamp = date.replace(
hour=hour,
minute=np.random.randint(0, 60),
second=np.random.randint(0, 60),
)
model = np.random.choice(
["gpt-4o", "gpt-4o-mini"],
p=[0.3, 0.7],
)
prompt_tokens = int(np.random.lognormal(mean=6.5, sigma=0.8))
completion_tokens = int(np.random.lognormal(mean=5.5, sigma=1.0))
total_tokens = prompt_tokens + completion_tokens
pricing = MODEL_PRICING[model]
estimated_cost = (
prompt_tokens / 1000 * pricing["prompt"]
+ completion_tokens / 1000 * pricing["completion"]
)
logs.append({
"timestamp": timestamp,
"model_deployment_name": f"{model}-deployment",
"model_name": model,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": total_tokens,
"estimated_cost": round(estimated_cost, 6),
"caller_ip": f"10.0.{np.random.randint(1, 10)}.{np.random.randint(1, 255)}",
"subscription_id": "00000000-0000-0000-0000-000000000000",
"business_unit": np.random.choice(BUSINESS_UNITS, p=[0.4, 0.2, 0.3, 0.1]),
})
# Inject anomalies
logs.extend(_inject_anomalies(base_date, days))
df = pd.DataFrame(logs)
df = df.sort_values("timestamp").reset_index(drop=True)
return dfWhat does this code do?
The function generates ~70 records per weekday and ~21 on weekends, mimicking real business usage patterns. Each record uses a lognormal distribution for token counts, which produces the long-tail pattern you see in real LLM traffic (most requests are small, a few are very large).
The 70/30 model split (gpt-4o-mini dominant, gpt-4o for complex tasks) reflects a typical enterprise routing strategy where cheaper models handle bulk traffic.
- You won't see a change yet. The next chunk adds the helper functions that generate_token_logs() calls.
Inject realistic usage patterns and anomalies
Real GenAI cost problems come in distinct flavors. A runaway automation loop burns tokens at 10x normal rate. A model upgrade silently doubles per-request cost. An overnight batch job spikes usage when no one is watching.
- Below generate_token_logs(), add the business-hour probability distribution helper:
def _business_hour_distribution():
"""Probability distribution favoring business hours."""
probs = np.array([
0.01, 0.01, 0.01, 0.01, 0.01, 0.02, # 0-5
0.03, 0.05, 0.08, 0.10, 0.10, 0.09, # 6-11
0.07, 0.09, 0.10, 0.08, 0.06, 0.04, # 12-17
0.03, 0.02, 0.01, 0.01, 0.01, 0.01, # 18-23
])
return probs / probs.sum()What does this code do?
This creates a 24-element probability array where business hours (9AM-5PM) have the highest weights. The distribution ensures most LLM requests happen during the workday, with a small trickle overnight. Normalizing with probs.sum() guarantees the probabilities add to 1.0.
- Below _business_hour_distribution(), add the anomaly injection function:
def _inject_anomalies(base_date, days):
"""Inject 3 distinct anomaly patterns."""
anomalies = []
# Anomaly 1: Runaway loop (day 10, 500 rapid requests in 2 hours)
loop_date = base_date + timedelta(days=10)
for i in range(500):
timestamp = loop_date.replace(hour=14, minute=i // 9, second=(i * 7) % 60)
anomalies.append({
"timestamp": timestamp,
"model_deployment_name": "gpt-4o-deployment",
"model_name": "gpt-4o",
"prompt_tokens": 3000,
"completion_tokens": 2000,
"total_tokens": 5000,
"estimated_cost": round(3000 / 1000 * 0.005 + 2000 / 1000 * 0.015, 6),
"caller_ip": "10.0.5.100",
"subscription_id": "00000000-0000-0000-0000-000000000000",
"business_unit": "Engineering",
})
# Anomaly 2: Model upgrade cost spike (day 20, all traffic to expensive model)
spike_date = base_date + timedelta(days=20)
for i in range(200):
hour = 9 + (i % 8)
timestamp = spike_date.replace(hour=hour, minute=np.random.randint(0, 60))
anomalies.append({
"timestamp": timestamp,
"model_deployment_name": "gpt-4o-deployment",
"model_name": "gpt-4o",
"prompt_tokens": int(np.random.lognormal(7.0, 0.5)),
"completion_tokens": int(np.random.lognormal(6.5, 0.5)),
"total_tokens": 0, # Will be computed
"estimated_cost": 0, # Will be computed
"caller_ip": f"10.0.3.{np.random.randint(1, 50)}",
"subscription_id": "00000000-0000-0000-0000-000000000000",
"business_unit": "Data Science",
})
# Anomaly 3: Overnight batch job (day 25, 3AM burst)
batch_date = base_date + timedelta(days=25)
for i in range(150):
timestamp = batch_date.replace(hour=3, minute=i // 3, second=(i * 4) % 60)
anomalies.append({
"timestamp": timestamp,
"model_deployment_name": "gpt-4o-mini-deployment",
"model_name": "gpt-4o-mini",
"prompt_tokens": 8000,
"completion_tokens": 4000,
"total_tokens": 12000,
"estimated_cost": round(8000 / 1000 * 0.00015 + 4000 / 1000 * 0.0006, 6),
"caller_ip": "10.0.9.200",
"subscription_id": "00000000-0000-0000-0000-000000000000",
"business_unit": "Engineering",
})
# Fix computed fields for anomaly 2
for record in anomalies:
if record["total_tokens"] == 0:
record["total_tokens"] = record["prompt_tokens"] + record["completion_tokens"]
if record["estimated_cost"] == 0:
pricing = MODEL_PRICING[record["model_name"]]
record["estimated_cost"] = round(
record["prompt_tokens"] / 1000 * pricing["prompt"]
+ record["completion_tokens"] / 1000 * pricing["completion"],
6,
)
return anomaliesWhat do these anomalies simulate?
- Anomaly 1 (day 10): A runaway automation loop fires 500 requests to the expensive gpt-4o model in 2 hours. Each request uses 5000 tokens.
- Anomaly 2 (day 20): Someone switches all Data Science traffic to gpt-4o with larger prompts, doubling per-request cost without increasing volume.
- Anomaly 3 (day 25): An overnight batch job at 3AM generates 150 high-token requests on gpt-4o-mini. Cheaper per token but massive volume.
The fix loop at the end computes total_tokens and estimated_cost for anomaly 2 records where the random token counts are generated dynamically.
- Save token_logs.py.
- You won't run the file yet. The next chunk adds the main() function that ties everything together.
Run the generator and verify output
- Below _inject_anomalies(), add the main() function that runs the generator and prints summary statistics:
def main():
print("=" * 60)
print("LLM Token Usage Log Generator")
print("=" * 60)
df = generate_token_logs()
print(f"\nDataset shape: {df.shape}")
print(f"Date range: {df['timestamp'].min()} to {df['timestamp'].max()}")
print(f"\nTotal tokens consumed: {df['total_tokens'].sum():,.0f}")
print(f"Total estimated cost: ${df['estimated_cost'].sum():,.2f}")
print(f"\nCost by model:")
print(df.groupby("model_name")["estimated_cost"].sum().sort_values(ascending=False))
print(f"\nCost by business unit:")
print(df.groupby("business_unit")["estimated_cost"].sum().sort_values(ascending=False))
return df
if __name__ == "__main__":
main()What does this code do?
The main() function calls generate_token_logs() and prints a summary dashboard: total token consumption, total estimated cost, cost breakdown by model, and cost breakdown by business unit. The model breakdown reveals how much the expensive model contributes to total spend.
- Save token_logs.py.
- Run the generator by executing this command in your terminal:
python token_logs.py- You should see output showing a dataset shape of 2000+ rows, total tokens consumed, total estimated cost, and breakdowns by model and business unit.
- Look at the cost-by-model summary. The gpt-4o model should dominate total cost despite handling only 30% of requests, because of the anomaly injections on days 10 and 20.
Seeing a NameError or ImportError?
- If you see NameError: name '_business_hour_distribution' is not defined, make sure the function is defined above the if __name__ block. Python needs all functions defined before main() calls them at runtime.
- If you see ModuleNotFoundError: No module named 'numpy', confirm your virtual environment is activated. You should see (.venv) at the start of your terminal prompt.
Still stuck? Help me debug my token_logs.py script. I'm getting an error when I run it.
✔️ Awesome, I've got everything!
Great. Double check you've saved token_logs.py and confirm the output shows 2000+ rows with cost breakdowns.
ⓧ I'd like to double check the full code
Here is the complete token_logs.py file for comparison:
"""Generate simulated LLM gateway token usage logs."""
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
# Pricing per 1K tokens (simulated Azure OpenAI rates)
MODEL_PRICING = {
"gpt-4o": {"prompt": 0.005, "completion": 0.015},
"gpt-4o-mini": {"prompt": 0.00015, "completion": 0.0006},
}
BUSINESS_UNITS = ["Engineering", "Marketing", "Data Science", "Product"]
def generate_token_logs(days=30, records_per_day=70):
"""Generate realistic API Management gateway LLM token logs."""
np.random.seed(123)
base_date = datetime.now() - timedelta(days=days)
logs = []
for day_offset in range(days):
date = base_date + timedelta(days=day_offset)
is_weekday = date.weekday() < 5
daily_records = records_per_day if is_weekday else int(records_per_day * 0.3)
for _ in range(daily_records):
hour = np.random.choice(
range(24),
p=_business_hour_distribution(),
)
timestamp = date.replace(
hour=hour,
minute=np.random.randint(0, 60),
second=np.random.randint(0, 60),
)
model = np.random.choice(
["gpt-4o", "gpt-4o-mini"],
p=[0.3, 0.7],
)
prompt_tokens = int(np.random.lognormal(mean=6.5, sigma=0.8))
completion_tokens = int(np.random.lognormal(mean=5.5, sigma=1.0))
total_tokens = prompt_tokens + completion_tokens
pricing = MODEL_PRICING[model]
estimated_cost = (
prompt_tokens / 1000 * pricing["prompt"]
+ completion_tokens / 1000 * pricing["completion"]
)
logs.append({
"timestamp": timestamp,
"model_deployment_name": f"{model}-deployment",
"model_name": model,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": total_tokens,
"estimated_cost": round(estimated_cost, 6),
"caller_ip": f"10.0.{np.random.randint(1, 10)}.{np.random.randint(1, 255)}",
"subscription_id": "00000000-0000-0000-0000-000000000000",
"business_unit": np.random.choice(BUSINESS_UNITS, p=[0.4, 0.2, 0.3, 0.1]),
})
# Inject anomalies
logs.extend(_inject_anomalies(base_date, days))
df = pd.DataFrame(logs)
df = df.sort_values("timestamp").reset_index(drop=True)
return df
def _business_hour_distribution():
"""Probability distribution favoring business hours."""
probs = np.array([
0.01, 0.01, 0.01, 0.01, 0.01, 0.02, # 0-5
0.03, 0.05, 0.08, 0.10, 0.10, 0.09, # 6-11
0.07, 0.09, 0.10, 0.08, 0.06, 0.04, # 12-17
0.03, 0.02, 0.01, 0.01, 0.01, 0.01, # 18-23
])
return probs / probs.sum()
def _inject_anomalies(base_date, days):
"""Inject 3 distinct anomaly patterns."""
anomalies = []
# Anomaly 1: Runaway loop (day 10, 500 rapid requests in 2 hours)
loop_date = base_date + timedelta(days=10)
for i in range(500):
timestamp = loop_date.replace(hour=14, minute=i // 9, second=(i * 7) % 60)
anomalies.append({
"timestamp": timestamp,
"model_deployment_name": "gpt-4o-deployment",
"model_name": "gpt-4o",
"prompt_tokens": 3000,
"completion_tokens": 2000,
"total_tokens": 5000,
"estimated_cost": round(3000 / 1000 * 0.005 + 2000 / 1000 * 0.015, 6),
"caller_ip": "10.0.5.100",
"subscription_id": "00000000-0000-0000-0000-000000000000",
"business_unit": "Engineering",
})
# Anomaly 2: Model upgrade cost spike (day 20, all traffic to expensive model)
spike_date = base_date + timedelta(days=20)
for i in range(200):
hour = 9 + (i % 8)
timestamp = spike_date.replace(hour=hour, minute=np.random.randint(0, 60))
anomalies.append({
"timestamp": timestamp,
"model_deployment_name": "gpt-4o-deployment",
"model_name": "gpt-4o",
"prompt_tokens": int(np.random.lognormal(7.0, 0.5)),
"completion_tokens": int(np.random.lognormal(6.5, 0.5)),
"total_tokens": 0, # Will be computed
"estimated_cost": 0, # Will be computed
"caller_ip": f"10.0.3.{np.random.randint(1, 50)}",
"subscription_id": "00000000-0000-0000-0000-000000000000",
"business_unit": "Data Science",
})
# Anomaly 3: Overnight batch job (day 25, 3AM burst)
batch_date = base_date + timedelta(days=25)
for i in range(150):
timestamp = batch_date.replace(hour=3, minute=i // 3, second=(i * 4) % 60)
anomalies.append({
"timestamp": timestamp,
"model_deployment_name": "gpt-4o-mini-deployment",
"model_name": "gpt-4o-mini",
"prompt_tokens": 8000,
"completion_tokens": 4000,
"total_tokens": 12000,
"estimated_cost": round(8000 / 1000 * 0.00015 + 4000 / 1000 * 0.0006, 6),
"caller_ip": "10.0.9.200",
"subscription_id": "00000000-0000-0000-0000-000000000000",
"business_unit": "Engineering",
})
# Fix computed fields for anomaly 2
for record in anomalies:
if record["total_tokens"] == 0:
record["total_tokens"] = record["prompt_tokens"] + record["completion_tokens"]
if record["estimated_cost"] == 0:
pricing = MODEL_PRICING[record["model_name"]]
record["estimated_cost"] = round(
record["prompt_tokens"] / 1000 * pricing["prompt"]
+ record["completion_tokens"] / 1000 * pricing["completion"],
6,
)
return anomalies
def main():
print("=" * 60)
print("LLM Token Usage Log Generator")
print("=" * 60)
df = generate_token_logs()
print(f"\nDataset shape: {df.shape}")
print(f"Date range: {df['timestamp'].min()} to {df['timestamp'].max()}")
print(f"\nTotal tokens consumed: {df['total_tokens'].sum():,.0f}")
print(f"Total estimated cost: ${df['estimated_cost'].sum():,.2f}")
print(f"\nCost by model:")
print(df.groupby("model_name")["estimated_cost"].sum().sort_values(ascending=False))
print(f"\nCost by business unit:")
print(df.groupby("business_unit")["estimated_cost"].sum().sort_values(ascending=False))
return df
if __name__ == "__main__":
main()You now have a realistic GenAI cost dataset with clearly planted anomalies. Next up, you will merge this with the Azure billing data into a single unified dataset and build an anomaly detection engine to automatically catch these cost spikes.
Merge into FOCUS Schema and Detect Anomalies
You now have two independent data sources: Azure billing costs and LLM token usage logs. In production, these would live in separate systems with different schemas, making cross-analysis impossible without a common format.
In this step, you will merge both sources into a single dataset using the industry-standard FOCUS 1.2 schema, then run statistical anomaly detection to automatically flag cost spikes.
In this step, get ready to:
- Transform both data sources into FOCUS 1.2 schema columns.
- Implement rolling Z-score anomaly detection on the unified dataset.
- Run the full pipeline and export results to CSV.
Write the FOCUS schema transform functions
The FOCUS (FinOps Open Cost and Usage Specification) schema gives you a single set of column names that work across Azure, AWS, and GCP. By transforming both your Azure costs and LLM logs into this format, you can query them as one dataset.
- Create a new file called pipeline.py in your project folder.
- Add the imports and the first transform function by pasting this code:
"""Merge cost sources into FOCUS schema and detect anomalies."""
import pandas as pd
import numpy as np
from datetime import datetime
from azure_costs import main as get_azure_costs
from token_logs import generate_token_logs
def transform_azure_to_focus(df):
"""Transform Azure cost data to FOCUS 1.2 schema."""
focus_df = pd.DataFrame({
"BilledCost": df["Cost"].astype(float),
"BillingCurrency": df.get("Currency", "USD"),
"ChargePeriodStart": pd.to_datetime(df["UsageDate"]),
"ChargePeriodEnd": pd.to_datetime(df["UsageDate"]) + pd.Timedelta(days=1),
"ServiceName": df["ServiceName"],
"ServiceCategory": df["ServiceName"].apply(_map_service_category),
"ResourceId": df.get("ResourceId", ""),
"ResourceName": df.get("ResourceId", "").apply(lambda x: x.split("/")[-1] if "/" in str(x) else str(x)),
"ChargeCategory": "Usage",
"Tags": "{}",
"SubAccountId": "00000000-0000-0000-0000-000000000000",
"x_TokensConsumed": 0,
"x_Source": "AzureCostManagement",
})
return focus_dfWhat does this code do?
The imports pull in your two data source modules (azure_costs and token_logs) plus pandas and numpy for data manipulation.
The transform_azure_to_focus() function maps Azure billing columns to FOCUS 1.2 standard names like BilledCost, ChargePeriodStart, and ServiceCategory. The x_TokensConsumed column uses the FOCUS convention of prefixing custom extensions with x_.
Why use the FOCUS schema?
FOCUS 1.2 is the industry standard for cloud cost data. Azure, AWS, and GCP all support it natively as data generators. By transforming your data into FOCUS columns now, your pipeline can ingest costs from any cloud provider without rewriting downstream logic.
- Add the LLM token log transform function below transform_azure_to_focus():
def transform_tokens_to_focus(df):
"""Transform LLM token logs to FOCUS 1.2 schema."""
df["date"] = df["timestamp"].dt.date
daily = df.groupby(["date", "model_name", "business_unit"]).agg(
BilledCost=("estimated_cost", "sum"),
x_TokensConsumed=("total_tokens", "sum"),
).reset_index()
focus_df = pd.DataFrame({
"BilledCost": daily["BilledCost"],
"BillingCurrency": "USD",
"ChargePeriodStart": pd.to_datetime(daily["date"]),
"ChargePeriodEnd": pd.to_datetime(daily["date"]) + pd.Timedelta(days=1),
"ServiceName": "Microsoft.CognitiveServices",
"ServiceCategory": "AI and Machine Learning",
"ResourceId": daily["model_name"].apply(
lambda m: f"/subscriptions/00000000/resourceGroups/rg-ai/providers/Microsoft.CognitiveServices/accounts/{m}"
),
"ResourceName": daily["model_name"],
"ChargeCategory": "Usage",
"Tags": daily["business_unit"].apply(lambda bu: f'{{"business_unit": "{bu}"}}'),
"SubAccountId": "00000000-0000-0000-0000-000000000000",
"x_TokensConsumed": daily["x_TokensConsumed"],
"x_Source": "LLMGateway",
})
return focus_dfWhat does this function do?
This function aggregates the per-request token logs into daily summaries grouped by model and business unit. It then maps these into the same FOCUS columns as the Azure costs, setting x_Source to "LLMGateway" so you can distinguish GenAI rows from infrastructure rows in the merged dataset.
- Add the service category mapper below transform_tokens_to_focus():
def _map_service_category(service_name):
"""Map Azure service names to FOCUS ServiceCategory."""
mapping = {
"Microsoft.Compute": "Compute",
"Microsoft.Storage": "Storage",
"Microsoft.CognitiveServices": "AI and Machine Learning",
"Microsoft.Network": "Networking",
"Microsoft.Sql": "Databases",
}
return mapping.get(service_name, "Other")What does this helper do?
The _map_service_category() function converts Azure resource provider names (like Microsoft.Compute) into human-readable FOCUS categories (like Compute). The underscore prefix marks it as an internal helper.
- Save pipeline.py.
You won't see a change yet. The next chunk adds the anomaly detection logic that will let you run the full pipeline.
Seeing a syntax error on save?
Check that each function starts with def at column 0 (no extra indentation). Make sure the dictionary inside each pd.DataFrame({}) has matching opening and closing braces.
Help me fix a syntax error in my pipeline.py FOCUS transform functions.
Implement Z-score anomaly detection
A Z-score measures how far a data point is from the average, expressed in standard deviations. By computing a rolling Z-score over a 7-day window, you can flag days where spend jumps unexpectedly relative to recent history.
- Add the anomaly detection function below _map_service_category() in pipeline.py:
def detect_anomalies(df, threshold=2.5, window=7):
"""Detect cost anomalies using rolling Z-score."""
df = df.copy()
df["date"] = df["ChargePeriodStart"].dt.date
daily_costs = df.groupby(["date", "ServiceCategory"])["BilledCost"].sum().reset_index()
daily_costs = daily_costs.sort_values(["ServiceCategory", "date"])
anomalies = []
for category in daily_costs["ServiceCategory"].unique():
cat_data = daily_costs[daily_costs["ServiceCategory"] == category].copy()
cat_data["rolling_mean"] = cat_data["BilledCost"].rolling(window=window, min_periods=3).mean()
cat_data["rolling_std"] = cat_data["BilledCost"].rolling(window=window, min_periods=3).std()
cat_data["z_score"] = (
(cat_data["BilledCost"] - cat_data["rolling_mean"]) / cat_data["rolling_std"]
)
anomaly_rows = cat_data[cat_data["z_score"] > threshold]
for _, row in anomaly_rows.iterrows():
anomalies.append({
"date": row["date"],
"service_category": category,
"daily_cost": round(row["BilledCost"], 4),
"rolling_mean": round(row["rolling_mean"], 4),
"z_score": round(row["z_score"], 2),
"severity": "HIGH" if row["z_score"] > 4 else "MEDIUM",
})
return pd.DataFrame(anomalies)How does the Z-score detection work?
- For each ServiceCategory, the function computes a 7-day rolling mean and standard deviation of daily costs.
- Any day where spend exceeds the rolling mean by more than 2.5 standard deviations gets flagged as an anomaly.
- A Z-score above 4 is labeled HIGH severity. Everything else flagged is MEDIUM.
- Save pipeline.py.
You won't see a change yet. The next chunk adds the main function that ties everything together and runs the pipeline.
Worried about the rolling window logic?
The min_periods=3 parameter means the rolling calculation needs at least 3 days of data before it starts computing. This prevents false positives in the first few days of the dataset where there is not enough history.
Help me understand the rolling Z-score anomaly detection logic in my pipeline.
Build the main function and run the pipeline
The main function orchestrates the full pipeline: ingest both data sources, transform them, merge, detect anomalies, and export results.
- Add the main function and entry point below detect_anomalies() in pipeline.py:
def main():
print("=" * 60)
print("FinOps Data Pipeline - Merge and Anomaly Detection")
print("=" * 60)
# Step 1: Get Azure costs
print("\n[1/4] Ingesting Azure cost data...")
azure_df = get_azure_costs()
focus_azure = transform_azure_to_focus(azure_df)
# Step 2: Get LLM token logs
print("\n[2/4] Generating LLM token usage logs...")
token_df = generate_token_logs()
focus_tokens = transform_tokens_to_focus(token_df)
# Step 3: Merge into unified FOCUS dataset
print("\n[3/4] Merging into FOCUS 1.2 schema...")
unified = pd.concat([focus_azure, focus_tokens], ignore_index=True)
unified = unified.sort_values("ChargePeriodStart").reset_index(drop=True)
print(f"\nUnified dataset: {unified.shape[0]} rows")
print(f"\nCost by ServiceCategory:")
cost_summary = unified.groupby("ServiceCategory")["BilledCost"].sum().sort_values(ascending=False)
print(cost_summary)
total_cost = unified["BilledCost"].sum()
genai_cost = unified[unified["ServiceCategory"] == "AI and Machine Learning"]["BilledCost"].sum()
print(f"\nTotal spend: ${total_cost:,.2f}")
print(f"GenAI spend: ${genai_cost:,.2f} ({genai_cost / total_cost * 100:.1f}% of total)")
# Step 4: Anomaly detection
print("\n[4/4] Running anomaly detection (Z-score, window=7, threshold=2.5)...")
anomalies = detect_anomalies(unified)
if len(anomalies) > 0:
print(f"\nDetected {len(anomalies)} anomalies:")
print(anomalies.to_string(index=False))
else:
print("\nNo anomalies detected.")
# Export
output_file = "finops_output.csv"
unified.to_csv(output_file, index=False)
print(f"\nExported merged dataset to: {output_file}")
return unified, anomalies
if __name__ == "__main__":
main()What does the main function do?
- It calls both data source modules and transforms each into FOCUS format.
- It concatenates both DataFrames into a single unified dataset and prints cost breakdowns.
- It runs anomaly detection and prints every flagged day with its Z-score and severity.
- It exports the merged dataset to finops_output.csv for downstream analysis.
- Save pipeline.py.
- Run the full pipeline by running this command:
python pipeline.py- You should see output showing the unified dataset size, cost breakdown by ServiceCategory, total and GenAI spend percentages, a list of 3 or more detected anomalies with dates and severity levels, and a confirmation that finops_output.csv was exported.
What should I see?
The pipeline prints a step-by-step progress report. Look for the anomaly detection section at the end. You should see at least 3 detected anomalies corresponding to the patterns you injected in the token logs: the runaway loop, the model upgrade spike, and the overnight batch job.
Each anomaly row shows the date, service category, daily cost, rolling mean, Z-score, and severity (HIGH or MEDIUM).
- Confirm finops_output.csv was created by running:
ls- You should see finops_output.csv listed alongside your Python files.
Seeing a ModuleNotFoundError?
If you see an error about azure_costs or token_logs not being found, make sure all three files (azure_costs.py, token_logs.py, pipeline.py) are in the same folder. Also confirm your virtual environment is still active (you should see (.venv) in your terminal prompt).
Help me debug a ModuleNotFoundError when running pipeline.py.
Seeing zero anomalies detected?
If the pipeline prints 0 anomalies, check that token_logs.py still has the _inject_anomalies() function and that generate_token_logs() calls it. The threshold of 2.5 should catch the injected spikes easily.
Help me figure out why my anomaly detection finds zero anomalies.
✔️ Awesome, I've got everything!
Great. Double check you have saved pipeline.py and that your terminal shows detected anomalies.
ⓧ I'd like to double check the full code
Here is the complete pipeline.py file. Compare it with yours:
"""Merge cost sources into FOCUS schema and detect anomalies."""
import pandas as pd
import numpy as np
from datetime import datetime
from azure_costs import main as get_azure_costs
from token_logs import generate_token_logs
def transform_azure_to_focus(df):
"""Transform Azure cost data to FOCUS 1.2 schema."""
focus_df = pd.DataFrame({
"BilledCost": df["Cost"].astype(float),
"BillingCurrency": df.get("Currency", "USD"),
"ChargePeriodStart": pd.to_datetime(df["UsageDate"]),
"ChargePeriodEnd": pd.to_datetime(df["UsageDate"]) + pd.Timedelta(days=1),
"ServiceName": df["ServiceName"],
"ServiceCategory": df["ServiceName"].apply(_map_service_category),
"ResourceId": df.get("ResourceId", ""),
"ResourceName": df.get("ResourceId", "").apply(lambda x: x.split("/")[-1] if "/" in str(x) else str(x)),
"ChargeCategory": "Usage",
"Tags": "{}",
"SubAccountId": "00000000-0000-0000-0000-000000000000",
"x_TokensConsumed": 0,
"x_Source": "AzureCostManagement",
})
return focus_df
def transform_tokens_to_focus(df):
"""Transform LLM token logs to FOCUS 1.2 schema."""
df["date"] = df["timestamp"].dt.date
daily = df.groupby(["date", "model_name", "business_unit"]).agg(
BilledCost=("estimated_cost", "sum"),
x_TokensConsumed=("total_tokens", "sum"),
).reset_index()
focus_df = pd.DataFrame({
"BilledCost": daily["BilledCost"],
"BillingCurrency": "USD",
"ChargePeriodStart": pd.to_datetime(daily["date"]),
"ChargePeriodEnd": pd.to_datetime(daily["date"]) + pd.Timedelta(days=1),
"ServiceName": "Microsoft.CognitiveServices",
"ServiceCategory": "AI and Machine Learning",
"ResourceId": daily["model_name"].apply(
lambda m: f"/subscriptions/00000000/resourceGroups/rg-ai/providers/Microsoft.CognitiveServices/accounts/{m}"
),
"ResourceName": daily["model_name"],
"ChargeCategory": "Usage",
"Tags": daily["business_unit"].apply(lambda bu: f'{{"business_unit": "{bu}"}}'),
"SubAccountId": "00000000-0000-0000-0000-000000000000",
"x_TokensConsumed": daily["x_TokensConsumed"],
"x_Source": "LLMGateway",
})
return focus_df
def _map_service_category(service_name):
"""Map Azure service names to FOCUS ServiceCategory."""
mapping = {
"Microsoft.Compute": "Compute",
"Microsoft.Storage": "Storage",
"Microsoft.CognitiveServices": "AI and Machine Learning",
"Microsoft.Network": "Networking",
"Microsoft.Sql": "Databases",
}
return mapping.get(service_name, "Other")
def detect_anomalies(df, threshold=2.5, window=7):
"""Detect cost anomalies using rolling Z-score."""
df = df.copy()
df["date"] = df["ChargePeriodStart"].dt.date
daily_costs = df.groupby(["date", "ServiceCategory"])["BilledCost"].sum().reset_index()
daily_costs = daily_costs.sort_values(["ServiceCategory", "date"])
anomalies = []
for category in daily_costs["ServiceCategory"].unique():
cat_data = daily_costs[daily_costs["ServiceCategory"] == category].copy()
cat_data["rolling_mean"] = cat_data["BilledCost"].rolling(window=window, min_periods=3).mean()
cat_data["rolling_std"] = cat_data["BilledCost"].rolling(window=window, min_periods=3).std()
cat_data["z_score"] = (
(cat_data["BilledCost"] - cat_data["rolling_mean"]) / cat_data["rolling_std"]
)
anomaly_rows = cat_data[cat_data["z_score"] > threshold]
for _, row in anomaly_rows.iterrows():
anomalies.append({
"date": row["date"],
"service_category": category,
"daily_cost": round(row["BilledCost"], 4),
"rolling_mean": round(row["rolling_mean"], 4),
"z_score": round(row["z_score"], 2),
"severity": "HIGH" if row["z_score"] > 4 else "MEDIUM",
})
return pd.DataFrame(anomalies)
def main():
print("=" * 60)
print("FinOps Data Pipeline - Merge and Anomaly Detection")
print("=" * 60)
# Step 1: Get Azure costs
print("\n[1/4] Ingesting Azure cost data...")
azure_df = get_azure_costs()
focus_azure = transform_azure_to_focus(azure_df)
# Step 2: Get LLM token logs
print("\n[2/4] Generating LLM token usage logs...")
token_df = generate_token_logs()
focus_tokens = transform_tokens_to_focus(token_df)
# Step 3: Merge into unified FOCUS dataset
print("\n[3/4] Merging into FOCUS 1.2 schema...")
unified = pd.concat([focus_azure, focus_tokens], ignore_index=True)
unified = unified.sort_values("ChargePeriodStart").reset_index(drop=True)
print(f"\nUnified dataset: {unified.shape[0]} rows")
print(f"\nCost by ServiceCategory:")
cost_summary = unified.groupby("ServiceCategory")["BilledCost"].sum().sort_values(ascending=False)
print(cost_summary)
total_cost = unified["BilledCost"].sum()
genai_cost = unified[unified["ServiceCategory"] == "AI and Machine Learning"]["BilledCost"].sum()
print(f"\nTotal spend: ${total_cost:,.2f}")
print(f"GenAI spend: ${genai_cost:,.2f} ({genai_cost / total_cost * 100:.1f}% of total)")
# Step 4: Anomaly detection
print("\n[4/4] Running anomaly detection (Z-score, window=7, threshold=2.5)...")
anomalies = detect_anomalies(unified)
if len(anomalies) > 0:
print(f"\nDetected {len(anomalies)} anomalies:")
print(anomalies.to_string(index=False))
else:
print("\nNo anomalies detected.")
# Export
output_file = "finops_output.csv"
unified.to_csv(output_file, index=False)
print(f"\nExported merged dataset to: {output_file}")
return unified, anomalies
if __name__ == "__main__":
main()Your FinOps data pipeline is complete. You have a unified FOCUS dataset that merges Azure infrastructure costs with GenAI token spend, and an anomaly detection engine that automatically surfaces cost spikes. In the Secret Mission, you will add tag-based chargeback allocation to split shared GenAI costs across business units.
Secret mission
Add Tag-Based Chargeback Allocation
Your pipeline detects anomalies, but who pays for the GenAI spend? In this extension, you'll build a tag-based cost allocation engine that splits shared costs across business units and outputs a chargeback summary.
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally and uses an Azure free account. There are no ongoing costs unless you deploy paid services to your Azure subscription.
Decide whether to keep your resources for future projects, pause them, or delete them entirely.
Resources you used:
- Azure free account ($200 credit for 30 days, no charges while idle).
- Local project folder with Python files, virtual environment, and CSV output.
Keep everything running
No action needed. Choose this if you plan to extend this pipeline or explore more Azure Cost Management features.
- Your Azure free account stays active with $200 credit for 30 days. No charges occur unless you deploy paid services.
- Your local Python files and virtual environment remain on your Desktop ready to run again.
- The Azure CLI stays installed on your machine for future Azure projects.
Pause - I'll come back to this later
Free up local memory by deactivating your virtual environment. Your Azure free account incurs no charges when idle.
- Deactivate your virtual environment by running:
deactivate- Your project files stay on disk. When you return, reactivate with:
.venv\Scripts\activate- Your Azure free account remains active. No charges are generated while you are not deploying services.
Delete - I don't want to use this again
Remove all project resources and cancel your Azure subscription if you no longer need it.
- Deactivate your virtual environment first by running:
deactivate- Delete your project folder by running:
macOS
cd ~/Desktop
rm -rf azure-finops-pipelineWindows
cd ~\Desktop
rmdir /s /q azure-finops-pipeline- To cancel your Azure subscription, sign in to the Azure portal.
- Navigate to Cost Management + Billing.
- In the left menu, select Subscriptions.
- Select the subscription you want to cancel.
- At the top of the page, select Cancel.
- Follow the prompts to confirm the cancellation.
What happens after cancellation?
Your subscription enters a disabled state. After 3 days, you can permanently delete it from the same Subscriptions page. If you do nothing, Azure automatically deletes it after 90 days.
Nice Work!
Nice Work!
Nice work! You just built a production-pattern FinOps data pipeline that ingests Azure billing data, correlates it with GenAI token usage, and automatically surfaces cost anomalies.
You've learned how to:
- Build a Python data pipeline that authenticates to Azure and queries the Cost Management API using the Azure SDK.
- Merge heterogeneous cost sources into the industry-standard FOCUS 1.2 schema with custom extension columns for GenAI token tracking.
- Implement rolling Z-score anomaly detection that flags runaway GenAI spend patterns with severity levels.
- Secret Mission: Implemented tag-based cost allocation and chargeback reporting that splits shared GenAI costs across business units.
Ready to quiz yourself?