1"""Track D: hosted APIs (Anthropic, OpenAI) with observed usage/cost accounting and documented prompt caching only.2Keys are read from the environment or sibling .env files; never logged."""3from __future__ import annotations4import os, re, time, json5from pathlib import Path67# Prices in USD per million tokens, from the provider pricing pages fetched 2026-10-29 (literature/audit_group3.md) —8# estimated cost, labelled 'estimated' in records.9PRICES = {10 "claude-haiku-4-5": {"input": 1.00, "output": 5.00, "cache_write_5m": 1.25, "cache_read": 0.10, "source": "platform.claude.com pricing table, 2026-10-29"},11 "gpt-5-nano": {"input": 0.05, "output": 0.40, "cached_input": 0.005, "source": "openai.com/api/pricing, as known 2026-10 (verify)"},12}1314def _load_key(name: str) -> str | None:15 if os.environ.get(name):16 return os.environ[name]17 for p in [Path(__file__).resolve().parents[2] / ".env", Path.home() / "Desktop/Recherche-UQO/immobilier/agentic-valuation-quebec/.env"]:18 if p.exists():19 m = re.search(rf"^{name}=(.*)$", p.read_text(), re.M)20 if m:21 return m.group(1).strip().strip('"').strip("'")22 return None2324class Budget:25 def __init__(self, cap_usd: float, ledger_path: str):26 self.cap = cap_usd; self.path = Path(ledger_path); self.spent = 0.027 if self.path.exists():28 for line in self.path.read_text().splitlines():29 try: self.spent += json.loads(line)["cost_usd"]30 except Exception: pass31 def charge(self, cost: float, meta: dict):32 self.spent += cost33 self.path.parent.mkdir(parents=True, exist_ok=True)34 with open(self.path, "a") as f:35 f.write(json.dumps({"cost_usd": cost, **meta}) + "\n")36 def exhausted(self) -> bool:37 return self.spent >= self.cap3839def anthropic_answer(system: str, user_blocks: list[dict], model: str = "claude-haiku-4-5", max_tokens: int = 128, use_cache_control: bool = False) -> dict:40 import anthropic41 client = anthropic.Anthropic(api_key=_load_key("ANTHROPIC_API_KEY"))42 content = []43 for i, b in enumerate(user_blocks):44 blk = {"type": "text", "text": b["text"]}45 if use_cache_control and b.get("cache"):46 blk["cache_control"] = {"type": "ephemeral"}47 content.append(blk)48 t0 = time.perf_counter(); first = None; text = ""49 with client.messages.stream(model=model, max_tokens=max_tokens, temperature=0.0, system=system, messages=[{"role": "user", "content": content}]) as stream:50 for ev in stream.text_stream:51 if first is None and ev:52 first = time.perf_counter()53 text += ev54 final = stream.get_final_message()55 t1 = time.perf_counter()56 u = final.usage57 pr = PRICES[model]58 cost = (u.input_tokens * pr["input"] + u.output_tokens * pr["output"] + (u.cache_creation_input_tokens or 0) * pr["cache_write_5m"] + (u.cache_read_input_tokens or 0) * pr["cache_read"]) / 1e659 return {"text": text, "model": final.model, "input_tokens": u.input_tokens, "output_tokens": u.output_tokens,60 "cache_creation_input_tokens": u.cache_creation_input_tokens, "cache_read_input_tokens": u.cache_read_input_tokens,61 "ttft_ms": ((first or t1) - t0) * 1000, "total_ms": (t1 - t0) * 1000, "cost_usd_estimated": cost, "price_source": pr["source"]}6263def openai_answer(system: str, user_text: str, model: str = "gpt-5-nano", max_tokens: int = 128) -> dict:64 import urllib.request65 key = _load_key("OPENAI_API_KEY")66 body = {"model": model, "input": [{"role": "developer", "content": system}, {"role": "user", "content": user_text}], "max_output_tokens": max_tokens, "reasoning": {"effort": "minimal"}}67 req = urllib.request.Request("https://api.openai.com/v1/responses", data=json.dumps(body).encode(), headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"})68 t0 = time.perf_counter()69 with urllib.request.urlopen(req, timeout=120) as r:70 d = json.load(r)71 t1 = time.perf_counter()72 text = "".join(c.get("text", "") for o in d.get("output", []) for c in o.get("content", []) if c.get("type") == "output_text")73 u = d.get("usage", {})74 cached = (u.get("input_tokens_details") or {}).get("cached_tokens", 0)75 pr = PRICES[model]76 cost = ((u.get("input_tokens", 0) - cached) * pr["input"] + cached * pr["cached_input"] + u.get("output_tokens", 0) * pr["output"]) / 1e677 return {"text": text, "model": d.get("model"), "input_tokens": u.get("input_tokens"), "output_tokens": u.get("output_tokens"), "cached_tokens": cached,78 "ttft_ms": None, "total_ms": (t1 - t0) * 1000, "cost_usd_estimated": cost, "price_source": pr["source"]}79