feat: add pricing table and cost calculation in telemetry emitter

src/polygateway/pricing.py

@@ -0,0 +1,73 @@
+"""模型单价表与成本换算(M2 设计 §6;参考仓零先例,从头设计)。
+
+**零内置单价**: 实验室走中转网关,计费非官方牌价;库内硬编码单价表
+必然过时并掩盖真实成本(P5 严禁默认值掩盖错误)。价格一律由使用方
+提供——JSON 文件(`PGW_PRICING_PATH`)或 dict 注入;币种由使用方全表
+统一口径,库不设币种字段(18 字段冻结)。查不到的 model → cost=None
+且每 model 仅首次 warning(防日志风暴),不阻塞调用。
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from loguru import logger
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+
+@dataclass(frozen=True)
+class ModelPrice:
+    """每百万 token 的输入/输出单价(币种由使用方口径统一)。"""
+
+    input_per_1m: float
+    output_per_1m: float
+
+    def __post_init__(self) -> None:
+        if self.input_per_1m < 0 or self.output_per_1m < 0:
+            raise ValueError("单价不能为负")
+
+
+class PricingTable:
+    """model → 单价 的只读表;cost() 是全库唯一换算点(经 TelemetryEmitter)。"""
+
+    def __init__(self, prices: Mapping[str, ModelPrice]) -> None:
+        self._prices = dict(prices)
+        self._warned: set[str] = set()
+
+    @classmethod
+    def from_file(cls, path: Path | str) -> PricingTable:
+        """装配 fail-loud: 文件缺失/JSON 坏/条目缺字段/负单价一律 ValueError。"""
+        p = Path(path)
+        try:
+            data = json.loads(p.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError) as exc:
+            raise ValueError(f"价格表 {p} 读取失败: {exc}") from exc
+        if not isinstance(data, dict):
+            raise ValueError(f"价格表 {p} 顶层必须是对象: model -> 单价")
+        prices: dict[str, ModelPrice] = {}
+        for model, entry in data.items():
+            if not isinstance(entry, dict) or not {"input_per_1m", "output_per_1m"} <= set(entry):
+                raise ValueError(f"价格表 {p} 条目 {model!r} 须含 input_per_1m 与 output_per_1m")
+            prices[model] = ModelPrice(
+                input_per_1m=float(entry["input_per_1m"]),
+                output_per_1m=float(entry["output_per_1m"]),
+            )
+        return cls(prices)
+
+    def cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> float | None:
+        """换算一次调用成本;未知 model 记 None 并仅首次 warning。"""
+        price = self._prices.get(model)
+        if price is None:
+            if model not in self._warned:
+                self._warned.add(model)
+                logger.warning("pricing 表无 model {!r} 的单价,cost 记 None", model)
+            return None
+        return (
+            prompt_tokens / 1_000_000 * price.input_per_1m
+            + completion_tokens / 1_000_000 * price.output_per_1m
+        )