prices.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/benchmarks/prices.py
 
"""
Dated, provenance-stamped price table + cost computation.
 
Purpose
-------
Tokens are the source of truth; ``$`` is a best-effort convenience
layer on top (plan 136 ADR-136-3). This module loads a ``prices.json``
that carries an ``as_of`` capture date and a ``source`` URL, and turns
a token split into dollars — costing cache **hit** and **miss** at
their distinct rates (never summed). When a model has no entry, callers
fall back to tokens-only.
 
The ``source`` URL lives in the data file, not in code, so a moved
pricing page is a one-field edit; ``--prices <path>`` (the CLI) points
at an alternate table entirely.
 
Responsibilities
----------------
- :class:`PriceTable` — loaded rates + provenance, with
  :meth:`cost_for` returning a per-category USD breakdown or ``None``.
- :func:`load_prices` — read + validate a ``prices.json``.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- ``load_prices`` never raises on a missing/garbage file — it returns
  an empty table (``models == {}``) so the run degrades to tokens-only.
- ``cost_for`` returns ``None`` when the model is absent (the report
  then shows "no price table"); rates are per-million-tokens USD.
- Hit and miss are costed independently; the breakdown never merges
  them.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
import json
from dataclasses import dataclass, field
from pathlib import Path
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "module", "prices loaded", 4)
 
# Bundled default table shipped with the suite.
DEFAULT_PRICES_PATH = Path(__file__).with_name("prices.json")
 
 
@dataclass(frozen=True)
class PriceTable:
    """Per-(provider,model) rates with provenance.
 
    ``models`` maps ``"<provider>/<model>"`` (and bare ``"<model>"`` as
    a fallback key) to a rate dict with keys ``input_miss`` /
    ``input_hit`` / ``output`` (and optional ``reasoning``), all in USD
    per million tokens.
    """
 
    as_of: str = ""
    source: str = ""
    path: str = ""
    models: dict = field(default_factory=dict)
 
    def _lookup(self, provider: str, model: str) -> dict | None:
        return self.models.get(f"{provider}/{model}") or self.models.get(model)
 
    def has(self, provider: str, model: str) -> bool:
        return self._lookup(provider, model) is not None
 
    def cost_for(
        self,
        provider: str,
        model: str,
        *,
        cache_hit_tokens: int = 0,
        cache_miss_tokens: int = 0,
        output_tokens: int = 0,
        reasoning_tokens: int = 0,
    ) -> dict | None:
        """Return a USD breakdown for the given token split, or ``None``.
 
        Reasoning tokens are billed at the explicit ``reasoning`` rate
        when present, else at the ``output`` rate (DeepSeek bills
        reasoning as output). Hit and miss are costed at distinct rates.
        """
        rates = self._lookup(provider, model)
        if rates is None:
            return None
 
        def _r(key: str, default: float = 0.0) -> float:
            try:
                return float(rates.get(key, default))
            except (TypeError, ValueError):
                return default
 
        per_mtok = 1_000_000.0
        hit_usd = cache_hit_tokens / per_mtok * _r("input_hit")
        miss_usd = cache_miss_tokens / per_mtok * _r("input_miss")
        reasoning_rate = _r("reasoning", _r("output"))
        reasoning_usd = reasoning_tokens / per_mtok * reasoning_rate
        # Output tokens excluding reasoning (which is billed above).
        plain_output = max(0, output_tokens - reasoning_tokens)
        output_usd = plain_output / per_mtok * _r("output")
        return {
            "input_hit_usd": round(hit_usd, 6),
            "input_miss_usd": round(miss_usd, 6),
            "output_usd": round(output_usd, 6),
            "reasoning_usd": round(reasoning_usd, 6),
            "total_usd": round(hit_usd + miss_usd + output_usd + reasoning_usd, 6),
        }
 
    def miss_rate(self, provider: str, model: str) -> float:
        """The input cache-miss rate ($/Mtok) used for the cold-turn cost."""
        rates = self._lookup(provider, model)
        if rates is None:
            return 0.0
        try:
            return float(rates.get("input_miss", 0.0))
        except (TypeError, ValueError):
            return 0.0
 
 
def load_prices(path: str | Path | None = None) -> PriceTable:
    """Load a ``prices.json`` into a :class:`PriceTable`.
 
    Falls back to an empty table (tokens-only) on any read/parse error.
    """
    func = "load_prices"
    p = Path(path) if path is not None else DEFAULT_PRICES_PATH
    try:
        data = json.loads(p.read_text(encoding="utf-8"))
    except (OSError, ValueError) as exc:
        _dbg("OC_CHAT.MAIN", func, f"no usable price table at {p}: {exc}", 2)
        return PriceTable(path=str(p))
    models = data.get("models")
    if not isinstance(models, dict):
        models = {}
    return PriceTable(
        as_of=str(data.get("as_of", "")),
        source=str(data.get("source", "")),
        path=str(p),
        models=models,
    )
 
 
__all__ = ["PriceTable", "load_prices", "DEFAULT_PRICES_PATH"]