Cost Tracking Skill

You are an expert in LLM API cost management, implementing the monitoring and

attribution patterns used by teams running Claude, GPT-4, and Gemini at scale.

You write production-ready cost tracking systems in British English.

IMI colours: Navy #1A1A2E · Teal #0F3D66 · Gold #E2B95A

Part 1 — Model Pricing Table & Token Calculator

from dataclasses import dataclass
from decimal import Decimal

# Pricing in USD per 1M tokens — update quarterly
MODEL_PRICING = {
    # Anthropic Claude
    "claude-3-5-sonnet-20241022":   {"input": 3.00,  "output": 15.00, "cache_write": 3.75, "cache_read": 0.30},
    "claude-3-5-haiku-20241022":    {"input": 0.80,  "output": 4.00,  "cache_write": 1.00, "cache_read": 0.08},
    "claude-3-opus-20240229":       {"input": 15.00, "output": 75.00, "cache_write": 18.75, "cache_read": 1.50},
    "claude-3-haiku-20240307":      {"input": 0.25,  "output": 1.25,  "cache_write": 0.30, "cache_read": 0.03},

    # OpenAI
    "gpt-4o":                       {"input": 2.50,  "output": 10.00},
    "gpt-4o-mini":                  {"input": 0.15,  "output": 0.60},
    "gpt-4-turbo":                  {"input": 10.00, "output": 30.00},
    "text-embedding-3-small":       {"input": 0.02,  "output": 0.0},
    "text-embedding-3-large":       {"input": 0.13,  "output": 0.0},

    # Google
    "gemini-1.5-pro":               {"input": 1.25,  "output": 5.00},
    "gemini-1.5-flash":             {"input": 0.075, "output": 0.30},
    "gemini-2.0-flash":             {"input": 0.10,  "output": 0.40},

    # Groq (fast inference)
    "llama-3.3-70b-versatile":      {"input": 0.59,  "output": 0.79},
    "mixtral-8x7b-32768":           {"input": 0.24,  "output": 0.24},
}


@dataclass
class TokenUsage:
    model: str
    input_tokens: int
    output_tokens: int
    cache_read_tokens: int = 0
    cache_write_tokens: int = 0
    project: str = "default"
    user_id: str | None = None

    @property
    def cost_usd(self) -> float:
        pricing = MODEL_PRICING.get(self.model, {"input": 0, "output": 0})
        cost = (
            (self.input_tokens / 1_000_000) * pricing["input"]
            + (self.output_tokens / 1_000_000) * pricing["output"]
            + (self.cache_read_tokens / 1_000_000) * pricing.get("cache_read", 0)
            + (self.cache_write_tokens / 1_000_000) * pricing.get("cache_write", 0)
        )
        return round(cost, 6)

    @property
    def cost_gbp(self) -> float:
        # Approximate USD → GBP conversion (update as needed)
        return round(self.cost_usd * 0.79, 6)


def calculate_cost(
    model: str,
    input_tokens: int,
    output_tokens: int,
    cache_read_tokens: int = 0,
    cache_write_tokens: int = 0
) -> dict:
    usage = TokenUsage(model, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens)
    return {
        "model": model,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "total_tokens": input_tokens + output_tokens,
        "cost_usd": usage.cost_usd,
        "cost_gbp": usage.cost_gbp
    }


def estimate_monthly_cost(
    model: str,
    avg_input_tokens: int,
    avg_output_tokens: int,
    queries_per_day: int
) -> dict:
    """Estimate monthly API costs for planning/budgeting."""
    daily = calculate_cost(model, avg_input_tokens * queries_per_day,
                           avg_output_tokens * queries_per_day)
    return {
        "model": model,
        "queries_per_day": queries_per_day,
        "daily_cost_usd": daily["cost_usd"],
        "monthly_cost_usd": round(daily["cost_usd"] * 30, 2),
        "monthly_cost_gbp": round(daily["cost_gbp"] * 30, 2),
        "annual_cost_usd": round(daily["cost_usd"] * 365, 2)
    }

Part 2 — Usage Logger & Attribution

import sqlite3, datetime, json
from pathlib import Path


class UsageTracker:
    """
    Track LLM API usage with project/user attribution.
    SQLite-backed — no external dependencies.
    """

    def __init__(self, db_path: str = "./llm_usage.db"):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS usage (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    timestamp TEXT NOT NULL,
                    model TEXT NOT NULL,
                    project TEXT NOT NULL,
                    user_id TEXT,
                    input_tokens INTEGER NOT NULL,
                    output_tokens INTEGER NOT NULL,
                    cache_read_tokens INTEGER DEFAULT 0,
                    cache_write_tokens INTEGER DEFAULT 0,
                    cost_usd REAL NOT NULL,
                    metadata TEXT DEFAULT '{}'
                )
            """)
            conn.execute("CREATE INDEX IF NOT EXISTS idx_project ON usage(project)")
            conn.execute("CREATE INDEX IF NOT EXISTS idx_ts ON usage(timestamp)")

    def log(self, usage: TokenUsage, metadata: dict | None = None):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "INSERT INTO usage (timestamp,model,project,user_id,input_tokens,"
                "output_tokens,cache_read_tokens,cache_write_tokens,cost_usd,metadata) "
                "VALUES (?,?,?,?,?,?,?,?,?,?)",
                (
                    datetime.datetime.utcnow().isoformat(),
                    usage.model, usage.project, usage.user_id,
                    usage.input_tokens, usage.output_tokens,
                    usage.cache_read_tokens, usage.cache_write_tokens,
                    usage.cost_usd, json.dumps(metadata or {})
                )
            )

    def get_project_spend(self, project: str, days: int = 30) -> dict:
        cutoff = (datetime.datetime.utcnow() - datetime.timedelta(days=days)).isoformat()
        with sqlite3.connect(self.db_path) as conn:
            row = conn.execute(
                "SELECT COUNT(*), SUM(input_tokens), SUM(output_tokens), SUM(cost_usd) "
                "FROM usage WHERE project=? AND timestamp>?",
                (project, cutoff)
            ).fetchone()

        total_cost = row[3] or 0
        return {
            "project": project,
            "period_days": days,
            "total_calls": row[0] or 0,
            "total_input_tokens": row[1] or 0,
            "total_output_tokens": row[2] or 0,
            "total_cost_usd": round(total_cost, 4),
            "total_cost_gbp": round(total_cost * 0.79, 4),
            "avg_cost_per_call_usd": round(total_cost / max(row[0] or 1, 1), 6)
        }

    def get_model_breakdown(self, days: int = 30) -> list[dict]:
        cutoff = (datetime.datetime.utcnow() - datetime.timedelta(days=days)).isoformat()
        with sqlite3.connect(self.db_path) as conn:
            rows = conn.execute(
                "SELECT model, COUNT(*), SUM(cost_usd) FROM usage "
                "WHERE timestamp>? GROUP BY model ORDER BY SUM(cost_usd) DESC",
                (cutoff,)
            ).fetchall()
        return [{"model": r[0], "calls": r[1], "cost_usd": round(r[2] or 0, 4)} for r in rows]

    def top_users_by_spend(self, project: str, days: int = 30, limit: int = 10) -> list[dict]:
        cutoff = (datetime.datetime.utcnow() - datetime.timedelta(days=days)).isoformat()
        with sqlite3.connect(self.db_path) as conn:
            rows = conn.execute(
                "SELECT user_id, COUNT(*), SUM(cost_usd) FROM usage "
                "WHERE project=? AND timestamp>? AND user_id IS NOT NULL "
                "GROUP BY user_id ORDER BY SUM(cost_usd) DESC LIMIT ?",
                (project, cutoff, limit)
            ).fetchall()
        return [{"user_id": r[0], "calls": r[1], "cost_usd": round(r[2] or 0, 4)} for r in rows]


# Singleton tracker
tracker = UsageTracker("./imi_usage.db")

Part 3 — Budget Alerts & Limits

class BudgetGuard:
    """
    Budget enforcement — blocks requests when limits exceeded.
    Per-project and per-user limits.
    """

    def __init__(self, tracker: UsageTracker):
        self.tracker = tracker
        self.limits = {}    # {project: {daily_usd, monthly_usd}}

    def set_limit(self, project: str, daily_usd: float = None, monthly_usd: float = None):
        self.limits[project] = {"daily_usd": daily_usd, "monthly_usd": monthly_usd}

    def check_budget(self, project: str, estimated_cost: float = 0) -> dict:
        """Check if a project is within budget before making an API call."""
        limits = self.limits.get(project, {})

        # Daily check
        if limits.get("daily_usd"):
            daily_spend = self.tracker.get_project_spend(project, days=1)["total_cost_usd"]
            if daily_spend + estimated_cost > limits["daily_usd"]:
                return {
                    "allowed": False,
                    "reason": f"Daily budget exceeded: £{daily_spend:.2f} / £{limits['daily_usd']:.2f}"
                }

        # Monthly check
        if limits.get("monthly_usd"):
            monthly_spend = self.tracker.get_project_spend(project, days=30)["total_cost_usd"]
            if monthly_spend + estimated_cost > limits["monthly_usd"]:
                return {
                    "allowed": False,
                    "reason": f"Monthly budget exceeded: £{monthly_spend:.2f} / £{limits['monthly_usd']:.2f}"
                }

        return {"allowed": True}


# ── Cost-aware Claude wrapper ──────────────────────────────────────────────────
import anthropic

_client = anthropic.Anthropic()
budget_guard = BudgetGuard(tracker)
budget_guard.set_limit("imi-default", daily_usd=50.0, monthly_usd=500.0)


def cost_tracked_claude(
    messages: list[dict],
    model: str = "claude-3-5-sonnet-20241022",
    project: str = "imi-default",
    user_id: str | None = None,
    **kwargs
) -> str:
    """Claude call with automatic cost tracking and budget enforcement."""
    # Estimate cost before calling
    avg_input_estimate = sum(len(m["content"].split()) * 1.3 for m in messages)
    estimated = calculate_cost(model, int(avg_input_estimate), 300)

    budget_check = budget_guard.check_budget(project, estimated["cost_usd"])
    if not budget_check["allowed"]:
        raise ValueError(f"Budget limit reached: {budget_check['reason']}")

    response = _client.messages.create(model=model, messages=messages, **kwargs)

    usage = TokenUsage(
        model=model,
        input_tokens=response.usage.input_tokens,
        output_tokens=response.usage.output_tokens,
        cache_read_tokens=getattr(response.usage, "cache_read_input_tokens", 0),
        cache_write_tokens=getattr(response.usage, "cache_creation_input_tokens", 0),
        project=project,
        user_id=user_id
    )
    tracker.log(usage)

    return response.content[0].text

Part 4 — Cost Optimisation Advisor

def recommend_model_for_task(task_description: str) -> dict:
    """Recommend the most cost-effective model for a given task."""
    task_lower = task_description.lower()

    if any(k in task_lower for k in ["classify", "extract", "segment", "label", "simple"]):
        return {"recommended": "claude-3-haiku-20240307", "reason": "Simple extraction — Haiku is 10x cheaper", "cost_multiplier": 1}

    if any(k in task_lower for k in ["summarise", "rewrite", "translate", "format"]):
        return {"recommended": "claude-3-5-haiku-20241022", "reason": "Routine writing — Haiku handles well", "cost_multiplier": 1}

    if any(k in task_lower for k in ["analyse", "research", "strategy", "complex", "reason"]):
        return {"recommended": "claude-3-5-sonnet-20241022", "reason": "Complex reasoning — Sonnet best value", "cost_multiplier": 4}

    if any(k in task_lower for k in ["groundbreaking", "novel", "difficult", "long document", "expert"]):
        return {"recommended": "claude-opus-4", "reason": "Most capable — use for hardest tasks only", "cost_multiplier": 40}

    return {"recommended": "claude-3-5-sonnet-20241022", "reason": "Default — good balance", "cost_multiplier": 4}


def optimisation_report(project: str) -> str:
    """Generate cost optimisation recommendations for a project."""
    spend = tracker.get_project_spend(project, days=30)
    by_model = tracker.get_model_breakdown(days=30)

    report = f"## IMI Cost Optimisation Report — {project}\n\n"
    report += f"Monthly spend: ${spend['total_cost_usd']:.2f} (£{spend['total_cost_gbp']:.2f})\n"
    report += f"Total calls: {spend['total_calls']:,}\n\n"
    report += "### Model Breakdown\n"
    for m in by_model:
        report += f"- {m['model']}: {m['calls']} calls, ${m['cost_usd']:.2f}\n"
    report += "\n### Recommendations\n"
    report += "- Use claude-3-haiku for classification/extraction tasks (save ~90%)\n"
    report += "- Enable prompt caching for system prompts >1024 tokens (save ~85%)\n"
    report += "- Implement exact-match cache for repeated queries (save ~50%)\n"
    return report

Output Standards

• Always log: every production LLM call must log tokens + cost to UsageTracker
• Budget limits: set daily + monthly limits per project before deploying
• Model selection: use Haiku for simple tasks, Sonnet for complex — 10x cost difference
• Prompt caching: always cache system prompts >1024 tokens on Anthropic
• Currency: log in USD, report in GBP (£) for IMI UK clients
• British English in all reports and recommendations

pip install

pip install anthropic openai
# No additional dependencies — SQLite is stdlib

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)