Gemini 4 Argon Doubles In Price On 1 December: AI Model Routing On Cost Per Correct Answer, With The Eval Harness, Router And Budget Table In Code

"""Cost per correct answer on your own finance questions, across models and settings.

pip install "google-genai==2.30.1" "anthropic==1.13.0" "openai==3.31.0"   # Python 3.11+
Gemini runs through the Interactions API (google-genai 2.3.0 or later). Credentials come from
GEMINI_API_KEY, ANTHROPIC_API_KEY and, for a self-hosted model, OPEN_WEIGHT_BASE_URL/_API_KEY.
"""
from __future__ import annotations

import json
import math
import os
import re
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import asdict, dataclass
from datetime import date
from functools import cache
from pathlib import Path

import anthropic
from google import genai
from openai import OpenAI


@dataclass(frozen=True)
class Band:
    starts: date
    input: float    # USD per 1M fresh input tokens
    output: float   # USD per 1M generated tokens; Google bills thinking at this rate
    cached: float   # USD per 1M cached input tokens


# Standard-tier list prices, checked against the provider pricing pages on 14 October 2026.
PRICES: dict[str, list[Band]] = {
    "gemini-4-argon": [Band(date.min, 2.00, 10.00, 0.10), Band(date(2026, 12, 1), 4.00, 20.00, 0.20)],
    "gemini-3.8-flash": [Band(date.min, 0.75, 3.75, 0.075), Band(date(2027, 1, 1), 1.50, 7.50, 0.15)],
    "claude-opus-5-5": [Band(date.min, 4.00, 20.00, 0.20)],
}
TIER = {"standard": 1.0, "flex": 0.5}   # Gemini flex is half the standard rate

SYSTEM = (
    "You answer questions about company filings for a trading firm's research desk. "
    "Use only the filing text supplied. If the question has several parts, separate the answers with |. "
    "Reply with one JSON object and nothing else: "
    '{"answer": "<final answer in the unit asked for>", '
    '"evidence": ["<sentence or table row copied exactly from the filing>"], '
    '"confidence": "high" or "low"}'
)


@dataclass(frozen=True)
class Question:
    qid: str
    klass: str              # difficulty class, labelled by routing.router.classify()
    context: str            # the same filing extract for every model: retrieval is held constant
    question: str
    gold: str               # multi-part answers separated by |, graded all-pass
    kind: str               # "number" or "text"
    tolerance: float = 0.005


@dataclass(frozen=True)
class Config:
    name: str
    provider: str           # "gemini" | "anthropic" | "openai_compat"
    model: str
    pricing_key: str
    thinking: str | None = None     # Gemini thinking_level or Claude effort
    service_tier: str = "standard"


@dataclass
class Result:
    qid: str
    config: str
    klass: str
    correct: bool
    fresh_in: int
    cached_in: int
    out: int
    latency_s: float
    answer: str | None
    error: str | None = None


CONFIGS = [
    Config("argon-low", "gemini", "gemini-4-argon", "gemini-4-argon", "low"),
    Config("argon-medium", "gemini", "gemini-4-argon", "gemini-4-argon", "medium"),
    Config("argon-high", "gemini", "gemini-4-argon", "gemini-4-argon", "high"),
    Config("flash-medium", "gemini", "gemini-3.8-flash", "gemini-3.8-flash", "medium"),
    Config("opus-5.5-medium", "anthropic", "claude-opus-5-5", "claude-opus-5-5", "medium"),
]
if os.environ.get("OPEN_WEIGHT_BASE_URL"):
    # Self-hosted weights have no list price: use your GPU bill divided by measured throughput.
    PRICES["self-hosted"] = [Band(date.min, float(os.environ["OPEN_WEIGHT_USD_PER_M_IN"]),
                                  float(os.environ["OPEN_WEIGHT_USD_PER_M_OUT"]), 0.0)]
    CONFIGS.append(Config("deepseek-v4.1-flash", "openai_compat",
                          os.environ.get("OPEN_WEIGHT_MODEL", "deepseek-ai/DeepSeek-V4.1-Flash"), "self-hosted"))


@cache
def _gemini() -> genai.Client:
    return genai.Client()           # reads GEMINI_API_KEY


@cache
def _claude() -> anthropic.Anthropic:
    return anthropic.Anthropic()    # reads ANTHROPIC_API_KEY


@cache
def _local() -> OpenAI:
    return OpenAI(base_url=os.environ["OPEN_WEIGHT_BASE_URL"], api_key=os.environ["OPEN_WEIGHT_API_KEY"])


def call(cfg: Config, q: Question) -> tuple[str, int, int, int]:
    """One model call. Returns (reply, fresh input tokens, cached input tokens, billed output tokens)."""
    prompt = f"FILING:\n{q.context}\n\nQUESTION: {q.question}"
    if cfg.provider == "gemini":
        gen: dict = {"max_output_tokens": 32768}
        if cfg.thinking:
            gen["thinking_level"] = cfg.thinking
        it = _gemini().interactions.create(model=cfg.model, input=prompt, system_instruction=SYSTEM,
                                           generation_config=gen, service_tier=cfg.service_tier, store=False)
        u = it.usage
        total_in, cached = u.total_input_tokens or 0, u.total_cached_tokens or 0
        # Response and thought tokens are both billed as output: derive them from the total.
        out = (u.total_tokens or 0) - total_in - (u.total_tool_use_tokens or 0)
        return it.output_text or "", total_in - cached, cached, out
    if cfg.provider == "anthropic":
        # No refusal fallback here: an evaluation must measure the model it names.
        msg = _claude().messages.create(model=cfg.model, max_tokens=16000, system=SYSTEM,
                                        output_config={"effort": cfg.thinking or "medium"},
                                        messages=[{"role": "user", "content": prompt}])
        if msg.stop_reason == "refusal":
            raise RuntimeError("refusal")
        text = "".join(b.text for b in msg.content if b.type == "text")
        return text, msg.usage.input_tokens, msg.usage.cache_read_input_tokens or 0, msg.usage.output_tokens
    if cfg.provider == "openai_compat":
        r = _local().chat.completions.create(model=cfg.model, max_tokens=16000, messages=[
            {"role": "system", "content": SYSTEM}, {"role": "user", "content": prompt}])
        return r.choices[0].message.content or "", r.usage.prompt_tokens, 0, r.usage.completion_tokens
    raise ValueError(f"unknown provider {cfg.provider}")


def parse_reply(text: str) -> tuple[str | None, list[str], str]:
    m = re.search(r"\{.*\}", text, re.S)
    try:
        obj = json.loads(m.group(0)) if m else {}
    except json.JSONDecodeError:
        obj = {}
    answer = str(obj["answer"]) if "answer" in obj else None
    evidence = [e for e in obj.get("evidence", []) if isinstance(e, str)]
    return answer, evidence, str(obj.get("confidence", "low"))


def to_number(s: str) -> float | None:
    m = re.search(r"\(?-?\d[\d,]*\.?\d*\)?", s.replace("$", "").replace("£", ""))
    if not m:
        return None
    t = m.group(0).replace(",", "")
    try:
        v = float(t.strip("()"))
    except ValueError:
        return None
    return -v if t.startswith("(") and t.endswith(")") else v


def grade(q: Question, answer: str | None) -> bool:
    """All-pass: every part of a multi-part answer must be right."""
    if answer is None:
        return False
    got, want = [p.strip() for p in answer.split("|")], [p.strip() for p in q.gold.split("|")]
    if len(got) != len(want):
        return False
    for g, w in zip(got, want):
        if q.kind == "number":
            gv, wv = to_number(g), to_number(w)
            if gv is None or wv is None or not math.isclose(gv, wv, rel_tol=q.tolerance, abs_tol=1e-9):
                return False
        elif re.sub(r"[^a-z0-9]+", " ", g.lower()).strip() != re.sub(r"[^a-z0-9]+", " ", w.lower()).strip():
            return False
    return True


def run_one(cfg: Config, q: Question) -> Result:
    t0 = time.perf_counter()
    try:
        text, fresh, cached, out = call(cfg, q)
        answer, _, _ = parse_reply(text)
        return Result(q.qid, cfg.name, q.klass, grade(q, answer), fresh, cached, out, time.perf_counter() - t0, answer)
    except Exception as exc:  # an error is a wrong answer you still waited for
        return Result(q.qid, cfg.name, q.klass, False, 0, 0, 0, time.perf_counter() - t0, None, repr(exc)[:300])


def cost_of_tokens(cfg: Config, fresh: int, cached: int, out: int, on: date) -> float:
    band = [b for b in PRICES[cfg.pricing_key] if b.starts <= on][-1]
    return TIER[cfg.service_tier] * (fresh * band.input + cached * band.cached + out * band.output) / 1e6


def p95(xs: list[float]) -> float:
    s = sorted(xs)
    return s[max(0, math.ceil(0.95 * len(s)) - 1)] if s else float("nan")


def _stats(rows: list[Result], cfg: Config, on: date) -> dict:
    ok = sum(r.correct for r in rows)
    spend = sum(cost_of_tokens(cfg, r.fresh_in, r.cached_in, r.out, on) for r in rows)
    return {"n": len(rows), "accuracy": ok / len(rows), "cost_per_question": spend / len(rows),
            "cost_per_correct": spend / ok if ok else None, "p95_latency_s": p95([r.latency_s for r in rows])}


def summarise(results: list[Result], cfg: Config, on: date) -> dict:
    mine = [r for r in results if r.config == cfg.name]
    classes = sorted({r.klass for r in mine})
    return {"config": cfg.name, "priced_on": on.isoformat(), **_stats(mine, cfg, on),
            "errors": sum(r.error is not None for r in mine),
            "by_class": {k: _stats([r for r in mine if r.klass == k], cfg, on) for k in classes}}


def main(questions_path: str, out_dir: str = "eval_out") -> None:
    lines = Path(questions_path).read_text().splitlines()
    qs = [Question(**json.loads(line)) for line in lines if line.strip()]
    results: list[Result] = []
    with ThreadPoolExecutor(max_workers=8) as pool:
        for cfg in CONFIGS:
            results += pool.map(lambda q, c=cfg: run_one(c, q), qs)
    out = Path(out_dir)
    out.mkdir(exist_ok=True)
    (out / "results.jsonl").write_text("\n".join(json.dumps(asdict(r)) for r in results))
    # Same recorded tokens, two price lists: today's, and the day Argon's introductory price ends.
    for on in sorted({date.today(), date(2026, 12, 1)}):
        summary = [summarise(results, c, on) for c in CONFIGS]
        (out / f"summary_{on.isoformat()}.json").write_text(json.dumps(summary, indent=2))
        for s in summary:
            cpc = "n/a" if s["cost_per_correct"] is None else f"${s['cost_per_correct']:.4f}"
            print(f"{on} {s['config']:<20} acc {s['accuracy']:6.1%}  per correct {cpc:>9}  p95 {s['p95_latency_s']:5.1f}s")


if __name__ == "__main__":
    main(sys.argv[1])

"""Route each question to the cheapest setting that clears its class's accuracy bar, escalate when the
answer fails validation, and record every attempt against a team. Uses evals/harness.py from step 1.
"""
from __future__ import annotations

import json
import re
import sqlite3
import time
from dataclasses import dataclass
from datetime import date, datetime, timezone
from pathlib import Path

from evals.harness import CONFIGS, Question, call, cost_of_tokens, parse_reply, to_number

BY_NAME = {c.name: c for c in CONFIGS}

# The accuracy a setting must reach on the labelled set before it may serve a class. Set by the cost of
# being wrong, not by the model: a wrong lookup is caught in seconds, a wrong precedent may not be.
BARS = {"lookup": 0.97, "calc": 0.93, "multi_doc": 0.90, "judgement": 0.85}
ALWAYS_REVIEW = {"judgement"}

CALC = re.compile(r"\b(growth|change|margin|ratio|cagr|per share|basis points|bps|difference|average|annuali[sz]ed)\b", re.I)
JUDGE = re.compile(r"\b(why|explain|assess|material|guidance|tone|precedent|comparable|reasonable)\b", re.I)


def classify(question: str, n_documents: int = 1) -> str:
    """Free, deterministic and auditable. Label the evaluation set with this too, not by hand."""
    if JUDGE.search(question):
        return "judgement"
    if n_documents > 1:
        return "multi_doc"
    if CALC.search(question) or len(re.findall(r"\b(?:19|20)\d{2}\b", question)) >= 2:
        return "calc"
    return "lookup"


def build_table(summary: list[dict]) -> dict[str, list[tuple[str, bool]]]:
    """Per class: (setting, cleared_bar) rungs, cheapest per correct answer first, most accurate last."""
    table: dict[str, list[tuple[str, bool]]] = {}
    for klass, bar in BARS.items():
        rows = [(s["config"], s["by_class"][klass]) for s in summary if klass in s["by_class"]]
        if not rows:
            continue
        cleared = sorted((r for r in rows if r[1]["accuracy"] >= bar and r[1]["cost_per_correct"]),
                         key=lambda r: r[1]["cost_per_correct"])
        ladder = [(name, True) for name, _ in cleared]
        best = max(rows, key=lambda r: r[1]["accuracy"])[0]
        if best not in [name for name, _ in ladder]:
            ladder.append((best, False))          # nothing cheaper clears the bar: best effort plus review
        table[klass] = ladder
    return table


def table_for(on: date, out_dir: str = "eval_out") -> dict[str, list[tuple[str, bool]]]:
    """Use the summary priced for the latest date on or before `on`: November and December differ."""
    files = sorted(p for p in Path(out_dir).glob("summary_*.json") if p.stem[8:] <= on.isoformat())
    return build_table(json.loads(files[-1].read_text()))


def validate(answer: str | None, evidence: list[str], confidence: str, q: Question) -> list[str]:
    if answer is None:
        return ["reply did not parse"]
    problems = []
    flat = " ".join(q.context.split())
    if not evidence:
        problems.append("no evidence quoted")
    problems += [f"evidence not in filing: {e[:60]}" for e in evidence if " ".join(e.split()) not in flat]
    if q.kind == "number" and any(to_number(p) is None for p in answer.split("|")):
        problems.append("numeric answer did not parse")
    if confidence != "high":
        problems.append("model reported low confidence")   # 'low' escalates; 'high' never skips the checks
    return problems


class Ledger:
    def __init__(self, path: str = "ledger.db") -> None:
        self.db = sqlite3.connect(path)
        self.db.execute("""CREATE TABLE IF NOT EXISTS ledger (at TEXT, team TEXT, qid TEXT, klass TEXT,
            config TEXT, attempt INTEGER, usd REAL, latency_s REAL, passed INTEGER, problems TEXT)""")

    def record(self, *row) -> None:
        self.db.execute("INSERT INTO ledger VALUES (?,?,?,?,?,?,?,?,?,?)", row)
        self.db.commit()

    def month_to_date(self, team: str) -> float:
        month = datetime.now(timezone.utc).strftime("%Y-%m")
        (usd,) = self.db.execute("SELECT COALESCE(SUM(usd), 0) FROM ledger WHERE team = ? AND substr(at, 1, 7) = ?",
                                 (team, month)).fetchone()
        return usd


@dataclass
class Routed:
    answer: str | None
    config: str
    attempts: int
    usd: float
    needs_review: bool
    problems: list[str]


def route(q: Question, team: str, table: dict[str, list[tuple[str, bool]]], ledger: Ledger,
          max_attempts: int = 3) -> Routed:
    ladder = table[q.klass][:max_attempts]
    spent, problems = 0.0, []
    for attempt, (name, cleared) in enumerate(ladder, start=1):
        cfg, t0 = BY_NAME[name], time.perf_counter()
        try:
            text, fresh, cached, out = call(cfg, q)
            answer, evidence, confidence = parse_reply(text)
            problems = validate(answer, evidence, confidence, q)
        except Exception as exc:
            fresh = cached = out = 0
            answer, problems = None, [f"call failed: {exc!r}"[:200]]
        usd = cost_of_tokens(cfg, fresh, cached, out, date.today())
        spent += usd
        ledger.record(datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S"), team, q.qid, q.klass, name,
                      attempt, usd, time.perf_counter() - t0, int(not problems), json.dumps(problems))
        if not problems:
            return Routed(answer, name, attempt, spent, not cleared or q.klass in ALWAYS_REVIEW, [])
    # Nothing passed validation: no answer is cheaper than a wrong one in a research note.
    return Routed(None, ladder[-1][0], len(ladder), spent, True, problems)

"""The table for the budget meeting: per setting and for the routed mix, accuracy, cost per correct answer
today and from 1 December, p95 latency, monthly spend at current volume and wrong answers per month.
Reads eval_out/summary_<date>.json (step 1) and ledger.db (step 2). Usage: python -m reports.budget_table 2026-10-14
"""
from __future__ import annotations

import json
import sqlite3
import sys
from pathlib import Path

from routing.router import build_table


def load(day: str) -> dict[str, dict]:
    return {s["config"]: s for s in json.loads(Path(f"eval_out/summary_{day}.json").read_text())}


def live_mix(db_path: str = "ledger.db", days: int = 30) -> tuple[dict[str, int], dict[str, float]]:
    """Questions per class over the last 30 days, and how often each class escalated past its first rung."""
    rows = sqlite3.connect(db_path).execute(
        """SELECT klass, COUNT(*), AVG(escalated) FROM (
               SELECT klass, qid, MAX(attempt) > 1 AS escalated FROM ledger
               WHERE at >= datetime('now', ?) GROUP BY klass, qid)
           GROUP BY klass""", (f"-{days} days",)).fetchall()
    return {k: round(n * 30 / days) for k, n, _ in rows}, {k: float(e or 0) for k, _, e in rows}


def all_on(s: dict, volume: dict[str, int]) -> tuple[float, float]:
    """Monthly spend and wrong answers if every class went to this one setting."""
    spend = sum(n * s["by_class"][k]["cost_per_question"] for k, n in volume.items() if k in s["by_class"])
    wrong = sum(n * (1 - s["by_class"][k]["accuracy"]) for k, n in volume.items() if k in s["by_class"])
    return spend, wrong


def routed(summary: dict[str, dict], volume: dict[str, int], esc: dict[str, float]) -> tuple[float, float]:
    """Routed mix: first rung for every question, second rung for the measured share that escalates."""
    table, spend, wrong = build_table(list(summary.values())), 0.0, 0.0
    for k, n in volume.items():
        rungs = [summary[name]["by_class"][k] for name, _ in table.get(k, [])]
        if not rungs:
            continue
        spend += n * (rungs[0]["cost_per_question"] + esc.get(k, 0) * (rungs[1]["cost_per_question"] if len(rungs) > 1 else 0))
        wrong += n * (1 - rungs[0]["accuracy"])        # upper bound: before validation catches anything
    return spend, wrong


def usd(x: float | None) -> str:
    return "n/a" if x is None else f"${x:,.0f}" if x >= 100 else f"${x:,.4f}"


def report(today: str, dec: str = "2026-12-01") -> str:
    now, later = load(today), load(dec)
    volume, esc = live_mix()
    head = ("| Setting | Accuracy | Per correct answer (now) | Per correct answer (from 1 Dec) | p95 latency "
            "| Monthly cost (now) | Monthly cost (from 1 Dec) | Wrong answers a month |\n|" + "---|" * 8)
    lines = [head]
    for name, s in now.items():
        (m_now, wrong), (m_dec, _) = all_on(s, volume), all_on(later[name], volume)
        lines.append(f"| All on {name} | {s['accuracy']:.1%} | {usd(s['cost_per_correct'])} "
                     f"| {usd(later[name]['cost_per_correct'])} | {s['p95_latency_s']:.1f}s "
                     f"| {usd(m_now)} | {usd(m_dec)} | {wrong:,.0f} |")
    (r_now, r_wrong), (r_dec, _) = routed(now, volume, esc), routed(later, volume, esc)
    lines.append(f"| Routed mix | | | | | {usd(r_now)} | {usd(r_dec)} | {r_wrong:,.0f} (max) |")
    lines.append(f"\nVolume, last 30 days: {sum(volume.values()):,} questions; by class: {json.dumps(volume)}")
    return "\n".join(lines)


if __name__ == "__main__":
    print(report(sys.argv[1]))