llm-connect/llm_connect/fx.py
tegwick 09aa1f3604
Some checks failed
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Has been cancelled
Implement LLM-WP-0007: Kimi K3 default and EUR spend reporting
Add moonshotai/kimi-k3 as OpenRouter basemodel default after live smoke,
USD→EUR cost conversion, append-only usage ledger, and CLI run/cost/spend
week commands with token and euro reporting.
2026-08-03 22:00:34 +02:00

93 lines
2.9 KiB
Python

"""USD to EUR conversion helpers for display and spend reporting.
Rate tables remain USD-denominated (OpenRouter list prices). This module only
converts already-estimated USD amounts into euros.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Any
# Snapshot: euros per one US dollar. Operator can override via env.
DEFAULT_EUR_PER_USD = 0.92
DEFAULT_FX_CAPTURED_AT = "2026-08-03"
DEFAULT_FX_SOURCE = f"snapshot:{DEFAULT_FX_CAPTURED_AT}"
ENV_EUR_PER_USD = "LLM_CONNECT_EUR_PER_USD"
@dataclass(frozen=True)
class FxRate:
"""EUR conversion factor for one USD."""
eur_per_usd: float
source: str
captured_at: str = ""
def __post_init__(self) -> None:
rate = _positive_float("eur_per_usd", self.eur_per_usd)
source = str(self.source or "").strip()
if not source:
raise ValueError("source must be a non-empty string")
object.__setattr__(self, "eur_per_usd", rate)
object.__setattr__(self, "source", source)
object.__setattr__(self, "captured_at", str(self.captured_at or ""))
def usd_to_eur(self, amount_usd: float) -> float:
"""Convert a USD amount to EUR."""
return float(amount_usd) * self.eur_per_usd
def resolve_fx_rate(
explicit: FxRate | float | None = None,
*,
env: dict[str, str] | None = None,
) -> FxRate | None:
"""Resolve an FX rate from explicit value, environment, or bundled snapshot.
Resolution order:
1. *explicit* :class:`FxRate` or positive float (euros per USD)
2. Environment variable ``LLM_CONNECT_EUR_PER_USD``
3. Bundled snapshot rate
Returns ``None`` only when an explicit/env value is present but invalid
callers should pass nothing to get the snapshot. Empty env string falls
through to the snapshot.
"""
if explicit is not None:
if isinstance(explicit, FxRate):
return explicit
return FxRate(
eur_per_usd=float(explicit),
source="explicit",
captured_at="",
)
environ = env if env is not None else os.environ
raw = environ.get(ENV_EUR_PER_USD)
if raw is not None and str(raw).strip() != "":
return FxRate(
eur_per_usd=_positive_float(ENV_EUR_PER_USD, raw),
source=f"env:{ENV_EUR_PER_USD}",
captured_at="",
)
return FxRate(
eur_per_usd=DEFAULT_EUR_PER_USD,
source=DEFAULT_FX_SOURCE,
captured_at=DEFAULT_FX_CAPTURED_AT,
)
def _positive_float(name: str, value: Any) -> float:
if isinstance(value, bool):
raise ValueError(f"{name} must be a positive number")
try:
number = float(value)
except (TypeError, ValueError) as exc:
raise ValueError(f"{name} must be a positive number") from exc
if number <= 0:
raise ValueError(f"{name} must be a positive number")
return number