f841dfaeed
Add TOPLOC honest-noise calibration storage/dispatch and validator divergence reporting for AH-021. Add opt-in HuggingFace marketplace pricing refresh, price-change history, CLI flags, and AH-023 tracking docs. Verification: .venv/bin/python -m pytest tests/ -q -k 'not integration' => 346 passed, 2 skipped, 1 deselected; compileall packages tests passed; focused AH-021/AH-023 tests 32 passed.
315 lines
11 KiB
Python
315 lines
11 KiB
Python
"""Dynamic per-model pricing benchmarked against HuggingFace inference rates (issue 23).
|
|
|
|
Client-facing price per model tracks the market: 80% of the cheapest
|
|
comparable provider rate on HuggingFace's inference marketplace
|
|
(https://huggingface.co/inference/models), refreshed daily. Nodes are
|
|
unaffected — this only ever calls ``BillingLedger.set_price`` (the ledger's
|
|
existing write path), never touches node payouts (ADR-0015's 90/10 split
|
|
still applies to whatever price is charged).
|
|
|
|
Confirmed 2026-07-06: the pricing table is server-rendered into the initial
|
|
HTML response (SvelteKit SSR) — a plain stdlib ``urllib.request`` GET plus
|
|
HTML parsing is sufficient. No headless-browser fetch is required. Each
|
|
table row carries an anchor whose href is
|
|
``/<org>/<repo>/?inference_api=true&inference_provider=<provider>``, which is
|
|
a cheaper and more stable extraction anchor than the display text (which
|
|
duplicates the repo id at two responsive breakpoints).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
from dataclasses import dataclass
|
|
from html.parser import HTMLParser
|
|
from typing import Callable
|
|
|
|
HF_INFERENCE_MODELS_URL = "https://huggingface.co/inference/models"
|
|
DEFAULT_HF_PRICING_LOG_DB_PATH = "hf_pricing_log.sqlite"
|
|
DEFAULT_CLIENT_PRICE_FRACTION = 0.80 # charge 80% of the cheapest comparable rate
|
|
|
|
_ROW_HREF_RE = re.compile(
|
|
r"^/(?P<repo>[^/]+/[^/?]+)/\?inference_api=true&inference_provider=(?P<provider>[^&\"]+)"
|
|
)
|
|
_PRICE_RE = re.compile(r"^\$[\d,]*\.?\d+$")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HfPriceQuote:
|
|
"""One (model, provider) row from the HF inference pricing table."""
|
|
|
|
repo_id: str
|
|
provider: str
|
|
input_per_1m: float
|
|
output_per_1m: float
|
|
|
|
def blended_price_per_1k_tokens(self) -> float:
|
|
"""Average of input/output $-per-1M-token rates, converted to $/1k.
|
|
|
|
The tracker bills a single per-1k-token rate (``BillingLedger``
|
|
doesn't distinguish prompt vs. completion tokens), so this is the
|
|
simplest fair proxy for "this provider's rate" in that unit.
|
|
"""
|
|
return (self.input_per_1m + self.output_per_1m) / 2.0 / 1000.0
|
|
|
|
def alias_keys(self) -> tuple[str, str]:
|
|
"""Both the bare-repo and repo::provider forms an ``hf_aliases`` entry may use."""
|
|
return (self.repo_id.lower(), f"{self.repo_id.lower()}::{self.provider.lower()}")
|
|
|
|
|
|
class _HfPricingTableParser(HTMLParser):
|
|
"""Extracts (repo_id, provider, input$/1M, output$/1M) rows from the raw HTML."""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self._in_tr = False
|
|
self._row_match: tuple[str, str] | None = None
|
|
self._row_prices: list[float] = []
|
|
self._in_td = False
|
|
self._td_text: list[str] = []
|
|
self.quotes: list[HfPriceQuote] = []
|
|
|
|
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
|
if tag == "tr":
|
|
self._in_tr = True
|
|
self._row_match = None
|
|
self._row_prices = []
|
|
elif tag == "a" and self._in_tr and self._row_match is None:
|
|
href = dict(attrs).get("href") or ""
|
|
m = _ROW_HREF_RE.match(href)
|
|
if m:
|
|
self._row_match = (
|
|
urllib.parse.unquote(m.group("repo")),
|
|
urllib.parse.unquote(m.group("provider")),
|
|
)
|
|
elif tag == "td":
|
|
self._in_td = True
|
|
self._td_text = []
|
|
|
|
def handle_data(self, data: str) -> None:
|
|
if self._in_td:
|
|
self._td_text.append(data)
|
|
|
|
def handle_endtag(self, tag: str) -> None:
|
|
if tag == "td":
|
|
self._in_td = False
|
|
text = "".join(self._td_text).strip()
|
|
if _PRICE_RE.match(text):
|
|
self._row_prices.append(float(text.replace("$", "").replace(",", "")))
|
|
elif tag == "tr":
|
|
self._in_tr = False
|
|
if self._row_match and len(self._row_prices) >= 2:
|
|
repo_id, provider = self._row_match
|
|
self.quotes.append(
|
|
HfPriceQuote(
|
|
repo_id=repo_id,
|
|
provider=provider,
|
|
input_per_1m=self._row_prices[0],
|
|
output_per_1m=self._row_prices[1],
|
|
)
|
|
)
|
|
self._row_match = None
|
|
self._row_prices = []
|
|
|
|
|
|
def parse_hf_pricing_table(html: str) -> list[HfPriceQuote]:
|
|
"""Pure parsing function — no network I/O, so it's directly unit-testable."""
|
|
parser = _HfPricingTableParser()
|
|
parser.feed(html)
|
|
return parser.quotes
|
|
|
|
|
|
def _default_fetch_html(url: str, *, timeout: float) -> str:
|
|
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
return resp.read().decode("utf-8", errors="replace")
|
|
|
|
|
|
def fetch_hf_price_quotes(
|
|
search_term: str,
|
|
*,
|
|
fetch_html: Callable[[str], str] | None = None,
|
|
timeout: float = 15.0,
|
|
) -> list[HfPriceQuote]:
|
|
"""Fetch and parse the HF inference pricing table filtered by ``search_term``.
|
|
|
|
``fetch_html`` is the test injection point (mirrors the ``backend=``
|
|
convention used elsewhere in this package) — it takes the full URL and
|
|
returns the raw HTML text, so tests never hit the network.
|
|
"""
|
|
url = f"{HF_INFERENCE_MODELS_URL}?{urllib.parse.urlencode({'search': search_term})}"
|
|
if fetch_html is not None:
|
|
html = fetch_html(url)
|
|
else:
|
|
html = _default_fetch_html(url, timeout=timeout)
|
|
return parse_hf_pricing_table(html)
|
|
|
|
|
|
def cheapest_matching_quote(
|
|
quotes: list[HfPriceQuote], aliases: list[str]
|
|
) -> HfPriceQuote | None:
|
|
"""Cheapest quote whose repo (optionally ``repo::provider``) is in ``aliases``.
|
|
|
|
An alias of ``"org/repo"`` matches that repo under any provider; an
|
|
alias of ``"org/repo::provider"`` matches only that specific provider —
|
|
useful when only one provider's deployment has been human-verified as a
|
|
fair comparable (matching quantization/params).
|
|
"""
|
|
alias_set = {a.strip().lower() for a in aliases if isinstance(a, str) and a.strip()}
|
|
if not alias_set:
|
|
return None
|
|
matches = [q for q in quotes if alias_set & set(q.alias_keys())]
|
|
if not matches:
|
|
return None
|
|
return min(matches, key=lambda q: q.blended_price_per_1k_tokens())
|
|
|
|
|
|
class HfPricingLog:
|
|
"""Thread-safe SQLite-backed audit log of dynamic price changes (issue 23).
|
|
|
|
Every price change (old, new, source alias/provider, timestamp) is
|
|
recorded here so a client dispute over a charge can be reconciled
|
|
against exactly what the market-tracking job did and when — mirrors
|
|
``calibration.py``'s persistence shape.
|
|
"""
|
|
|
|
def __init__(self, db_path: str | None = None) -> None:
|
|
self._db_path = db_path
|
|
self._lock = threading.Lock()
|
|
self._changes: list[dict] = []
|
|
if self._db_path:
|
|
self._init_db()
|
|
self._load_from_db()
|
|
|
|
def record_change(
|
|
self,
|
|
*,
|
|
model: str,
|
|
old_price_per_1k: float,
|
|
new_price_per_1k: float,
|
|
source_repo_id: str,
|
|
source_provider: str,
|
|
ts: float | None = None,
|
|
) -> dict:
|
|
change = {
|
|
"model": model,
|
|
"old_price_per_1k": old_price_per_1k,
|
|
"new_price_per_1k": new_price_per_1k,
|
|
"source_repo_id": source_repo_id,
|
|
"source_provider": source_provider,
|
|
"ts": ts if ts is not None else time.time(),
|
|
}
|
|
with self._lock:
|
|
self._changes.append(change)
|
|
self._save_change(change)
|
|
return change
|
|
|
|
def history(self, model: str | None = None, *, limit: int = 200) -> list[dict]:
|
|
with self._lock:
|
|
changes = list(self._changes)
|
|
if model is not None:
|
|
changes = [c for c in changes if c["model"] == model]
|
|
return changes[-limit:]
|
|
|
|
# ---- persistence (billing.py / calibration.py pattern) ----
|
|
|
|
def _init_db(self) -> None:
|
|
con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
|
|
con.execute(
|
|
"CREATE TABLE IF NOT EXISTS hf_price_changes "
|
|
"(id INTEGER PRIMARY KEY AUTOINCREMENT, model TEXT NOT NULL, "
|
|
"payload TEXT NOT NULL, ts REAL NOT NULL)"
|
|
)
|
|
con.commit()
|
|
con.close()
|
|
|
|
def _load_from_db(self) -> None:
|
|
con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
|
|
rows = con.execute(
|
|
"SELECT payload FROM hf_price_changes ORDER BY ts, id"
|
|
).fetchall()
|
|
con.close()
|
|
for (payload,) in rows:
|
|
try:
|
|
self._changes.append(json.loads(payload))
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
def _save_change(self, change: dict) -> None:
|
|
if not self._db_path:
|
|
return
|
|
con = sqlite3.connect(self._db_path) # type: ignore[arg-type]
|
|
con.execute(
|
|
"INSERT INTO hf_price_changes (model, payload, ts) VALUES (?, ?, ?)",
|
|
(change["model"], json.dumps(change), float(change["ts"])),
|
|
)
|
|
con.commit()
|
|
con.close()
|
|
|
|
|
|
def hf_search_term(preset: dict, model_name: str) -> str:
|
|
"""Best-effort search term for the HF pricing page's ``?search=`` filter."""
|
|
hf_repo = preset.get("hf_repo")
|
|
if isinstance(hf_repo, str) and hf_repo:
|
|
return hf_repo.rsplit("/", 1)[-1]
|
|
return model_name
|
|
|
|
|
|
def refresh_preset_price(
|
|
*,
|
|
model_name: str,
|
|
preset: dict,
|
|
current_price: float,
|
|
fetch_html: Callable[[str], str] | None = None,
|
|
price_fraction: float = DEFAULT_CLIENT_PRICE_FRACTION,
|
|
) -> dict | None:
|
|
"""Compute the new price for one preset, or None if nothing should change.
|
|
|
|
Never raises — any fetch/parse failure or absence of a verified match is
|
|
treated identically: keep the static default (deliverable's fallback
|
|
requirement). Callers are responsible for actually applying the result
|
|
(``BillingLedger.set_price`` + logging), so this function stays a pure
|
|
"what should the new price be" computation and is trivially unit-testable.
|
|
"""
|
|
aliases = preset.get("hf_aliases")
|
|
if not aliases:
|
|
return None
|
|
try:
|
|
quotes = fetch_hf_price_quotes(
|
|
hf_search_term(preset, model_name), fetch_html=fetch_html
|
|
)
|
|
quote = cheapest_matching_quote(quotes, aliases)
|
|
except Exception:
|
|
return None
|
|
if quote is None:
|
|
return None
|
|
new_price = round(quote.blended_price_per_1k_tokens() * price_fraction, 6)
|
|
if new_price <= 0:
|
|
return None
|
|
return {
|
|
"model": model_name,
|
|
"old_price_per_1k": current_price,
|
|
"new_price_per_1k": new_price,
|
|
"source_repo_id": quote.repo_id,
|
|
"source_provider": quote.provider,
|
|
}
|
|
|
|
|
|
__all__ = [
|
|
"HF_INFERENCE_MODELS_URL",
|
|
"DEFAULT_HF_PRICING_LOG_DB_PATH",
|
|
"DEFAULT_CLIENT_PRICE_FRACTION",
|
|
"HfPriceQuote",
|
|
"HfPricingLog",
|
|
"parse_hf_pricing_table",
|
|
"fetch_hf_price_quotes",
|
|
"cheapest_matching_quote",
|
|
"hf_search_term",
|
|
"refresh_preset_price",
|
|
]
|