from __future__ import annotations as _annotations import dataclasses import re import warnings from collections.abc import Iterator, Mapping, Sequence from copy import copy from dataclasses import InitVar, dataclass, field from datetime import date, datetime, time, timezone from decimal import Decimal from numbers import Integral from typing import TYPE_CHECKING, Annotated, Any, Literal, TypeGuard, TypeVar, cast, overload import pydantic from typing_extensions import Self, TypedDict from genai_prices._usage import UsageValue, add_usage_values, usage_value_as_decimal, validate_usage_value from genai_prices.units import UnitRegistry if TYPE_CHECKING: from genai_prices.units import UnitDef __all__ = ( 'ProviderID', 'PriceCalculation', 'AbstractUsage', 'Usage', 'Provider', 'UsageExtractorMapping', 'UsageExtractor', 'ModelInfo', 'ModelPrice', 'TieredPrices', 'Tier', 'ConditionalPrice', 'StartDateConstraint', 'TimeOfDateConstraint', 'ClauseStartsWith', 'ClauseEndsWith', 'ClauseContains', 'ClauseRegex', 'ClauseEquals', 'ClauseOr', 'ClauseAnd', 'MatchLogic', 'ArrayMatch', ) # Define MatchLogic after __all__ to avoid forward reference issues def clause_discriminator(v: Any) -> str | None: assert isinstance(v, dict), f'Expected dict, got {type(v)}' return next(iter(v)) # pyright: ignore[reportUnknownArgumentType, reportUnknownVariableType] MatchLogic = Annotated[ Annotated['ClauseStartsWith', pydantic.Tag('starts_with')] | Annotated['ClauseEndsWith', pydantic.Tag('ends_with')] | Annotated['ClauseContains', pydantic.Tag('contains')] | Annotated['ClauseRegex', pydantic.Tag('regex')] | Annotated['ClauseEquals', pydantic.Tag('equals')] | Annotated['ClauseOr', pydantic.Tag('or')] | Annotated['ClauseAnd', pydantic.Tag('and')], pydantic.Discriminator(clause_discriminator), ] ProviderID = Literal[ 'avian', 'groq', 'openai', 'novita', 'fireworks', 'deepseek', 'mistral', 'x-ai', 'google', 'perplexity', 'aws', 'together', 'anthropic', 'azure', 'cohere', 'openrouter', ] @dataclass class ArrayMatch: type: Literal['array-match'] field: str match: MatchLogic def extract(self, items: Sequence[Any]) -> Mapping[str, Any] | None: for item in items: if _is_mapping(item) and (item_field := item.get(self.field)): if self.match.is_match(item_field): return item ExtractPath = str | Sequence[str | ArrayMatch] @dataclass(repr=False) class PriceCalculation: input_price: Decimal output_price: Decimal total_price: Decimal model: ModelInfo = dataclasses.field(repr=False) provider: Provider = dataclasses.field(repr=False) model_price: ModelPrice auto_update_timestamp: datetime | None def __repr__(self) -> str: return ( 'PriceCalculation(' f'input_price={self.input_price!r}, ' f'output_price={self.output_price!r}, ' f'total_price={self.total_price!r}, ' f'model={self.model.summary()}, ' f'provider={self.provider.summary()}, ' f'model_price=ModelPrice({self.model_price}), ' f'auto_update_timestamp={self.auto_update_timestamp!r})' ) @dataclass(repr=False) class ExtractedUsage: usage: Usage model: ModelInfo | None = dataclasses.field(repr=False) provider: Provider = dataclasses.field(repr=False) auto_update_timestamp: datetime | None def calc_price( self, *, genai_request_timestamp: datetime | None = None, model: ModelInfo | None = None ) -> PriceCalculation: """Calculate the price for the given usage. Args: genai_request_timestamp: The timestamp of the request to the GenAI service, use `None` to use the current time. model: The model to calculate the price for, if `None` the model from the response data is used. """ model = model or self.model if model is None: raise ValueError('No model reference found in response data and model not provided') return model.calc_price( self.usage, self.provider, genai_request_timestamp=genai_request_timestamp, auto_update_timestamp=self.auto_update_timestamp, ) def __repr__(self) -> str: return ( 'ExtractedUsage(' f'usage={self.usage!r}, ' f'model={self.model.summary() if self.model else None}, ' f'provider={self.provider.summary()}, ' f'auto_update_timestamp={self.auto_update_timestamp!r})' ) def __add__(self, other: ExtractedUsage | Any) -> ExtractedUsage: """Accumulate inner Usage, handling nullable usage fields. Accumulating usage is useful for common streaming situations where user wants to save and compute costs for all the response chunks in a stream Args: other: The usage to accumulate with this usage extraction instance. """ if not isinstance(other, ExtractedUsage): return NotImplemented # will raise a TypeError models_match = self.model and other.model and other.model.id == self.model.id if not models_match: raise ValueError(f'Cannot add {other} to {self}, models do not match {other.model} != {self.model}') providers_match = self.provider and other.provider and other.provider.id == self.provider.id if not providers_match: raise ValueError( f'Cannot add {other} to {self}, providers do not match {other.provider} != {self.provider}' ) return ExtractedUsage( model=self.model, provider=self.provider, auto_update_timestamp=self.auto_update_timestamp, usage=self.usage + other.usage, ) def __radd__(self, other: ExtractedUsage | Any) -> ExtractedUsage: return self + other AbstractUsage = object class Usage: """Simple token usage container.""" def __init__(self, **kwargs: UsageValue | None) -> None: reported_usage_keys = _reported_usage_keys() unknown_keys = kwargs.keys() - reported_usage_keys if unknown_keys: bad_keys = ', '.join(sorted(unknown_keys)) warnings.warn( f'Unsupported usage key for standard pricing: {bad_keys}', UserWarning, stacklevel=2, ) self._store_values(kwargs) @classmethod def from_raw(cls, obj: object) -> Usage: if isinstance(obj, Usage): obj._reported_values() return obj values: dict[str, UsageValue] = {} for key in _reported_usage_keys(): value = _raw_usage_value(obj, key) if value is not None: values[key] = value return cls(**values) def __setattr__(self, name: str, value: UsageValue | None) -> None: if name in _reported_usage_keys(): self._store_values({name: value}) else: object.__setattr__(self, name, value) def __getattr__(self, name: str) -> UsageValue: if name in _reported_usage_keys(): return self._infer_missing_value(name) raise AttributeError(f'{type(self).__name__!r} object has no attribute {name!r}') def _store_values(self, values: Mapping[str, UsageValue | None]) -> None: reported_usage_keys = _reported_usage_keys() for key, value in values.items(): if value is None: self.__dict__.pop(key, None) elif key in reported_usage_keys: self.__dict__[key] = validate_usage_value(key, value) else: self.__dict__[key] = value def _reported_values(self) -> dict[str, UsageValue]: reported_usage_keys = _reported_usage_keys() return { key: validate_usage_value(key, value) for key, value in self.__dict__.items() if key in reported_usage_keys } def reported_value(self, usage_key: str) -> UsageValue: return self._reported_values().get(usage_key, 0) def __add__(self, other: Usage | Any) -> Self: if not isinstance(other, Usage): return NotImplemented self_values = self._reported_values() other_values = other._reported_values() return type(self)( **{ key: add_usage_values(self_values.get(key, 0), other_values.get(key, 0)) for key in self_values.keys() | other_values.keys() } ) def __radd__(self, other: Usage | int) -> Usage: if other == 0: # Allow this to work with sum() return self if isinstance(other, Usage): return other + self return NotImplemented def __eq__(self, other: object) -> Any: if not isinstance(other, Usage): return NotImplemented return self._reported_values() == other._reported_values() def __repr__(self) -> str: values = ', '.join(f'{key}={value!r}' for key, value in self._ordered_values()) return f'{type(self).__name__}({values})' def _ordered_values(self) -> list[tuple[str, UsageValue]]: values = self._reported_values() return [(key, values[key]) for key in _reported_usage_key_order() if key in values] def _infer_missing_value(self, usage_key: str) -> UsageValue: from genai_prices.decompose import is_descendant_or_self from genai_prices.units import _get_registry # pyright: ignore[reportPrivateUsage] registry = _get_registry() requested_unit = registry.units[usage_key] reported_values = self._reported_values() descendant_keys = [ unit.usage_key for reported_key, value in reported_values.items() if value > 0 and (unit := registry.units.get(reported_key)) is not None and unit is not requested_unit and is_descendant_or_self(requested_unit, unit) ] if not descendant_keys: overlapping_keys = _reported_overlap_keys_for_join( requested_unit, [ unit for reported_key, value in reported_values.items() if value > 0 and (unit := registry.units.get(reported_key)) is not None ], ) if not overlapping_keys: return 0 reported_keys = ', '.join(overlapping_keys) raise ValueError( f'Missing usage for {usage_key}: reported overlapping usage keys {reported_keys} ' f'require explicit {usage_key}' ) reported_keys = ', '.join(sorted(descendant_keys)) raise ValueError( f'Missing usage for {usage_key}: reported descendant usage keys {reported_keys} ' f'require explicit {usage_key}' ) def _reported_overlap_keys_for_join( requested_unit: UnitDef, reported_units: Sequence[UnitDef] ) -> tuple[str, str] | None: from genai_prices.decompose import is_descendant_or_self sorted_units = sorted(reported_units, key=lambda unit: unit.usage_key) for index, left in enumerate(sorted_units): for right in sorted_units[index + 1 :]: if not left.is_compatible_with(right): continue if is_descendant_or_self(left, right) or is_descendant_or_self(right, left): continue if requested_unit.dimensions == {**left.dimensions, **right.dimensions}: return left.usage_key, right.usage_key return None _COMPACT_DATE_RE = re.compile(r'(-)(20\d{2})(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])(?=-|:|$)') def _normalize_compact_dated_ref(model_ref: str) -> str: """Rewrite a compact date suffix like ``-20251211`` to the canonical ``-2025-12-11`` form. LiteLLM and OpenRouter emit compact dated model refs (e.g. ``gpt-5.2-20251211``) that don't match the dashed aliases used in the price data. This is only applied as a fallback when a ref doesn't otherwise match, so models that match on the compact date form are left untouched. """ def replace(match: re.Match[str]) -> str: year, month, day = (int(match.group(index)) for index in range(2, 5)) try: date(year, month, day) except ValueError: return match.group(0) return f'{match.group(1)}{year:04d}-{month:02d}-{day:02d}' return _COMPACT_DATE_RE.sub(replace, model_ref) @dataclass class Provider: """Information about an LLM inference provider""" id: str """Unique identifier for the provider""" name: str """Link to pricing page for the provider""" api_pattern: str """Common name of the organization""" pricing_urls: list[str] | None = None """Pattern to identify provider via HTTP API URL.""" description: str | None = None """Description of the provider""" price_comments: str | None = None """Comments about the pricing of this provider's models, especially challenges in representing the provider's pricing model.""" model_match: MatchLogic | None = None """Logic to find a provider based on the model reference.""" provider_match: MatchLogic | None = None """Logic to find a provider based on the provider identifier.""" extractors: list[UsageExtractor] | None = None """Logic to extract usage information from the provider's API responses.""" fallback_model_providers: list[str] | None = None """List of provider identifiers to fallback to to get prices if this provider doesn't have a price. This is used when one provider offers another provider's models, e.g. Google and AWS offer Anthropic models, Azure offers OpenAI models, etc. """ models: list[ModelInfo] = dataclasses.field(default_factory=list) """List of models supported by this provider""" def find_model(self, model_ref: str, *, all_providers: list[Provider] | None = None) -> ModelInfo | None: model_ref = model_ref.lower() if model := self._match_model(model_ref, all_providers=all_providers): return model # LiteLLM/OpenRouter emit compact dated refs like `gpt-5.2-20251211`; retry with the # canonical dashed form if the ref didn't match as-is. normalized = _normalize_compact_dated_ref(model_ref) if normalized != model_ref: return self._match_model(normalized, all_providers=all_providers) return None def _match_model(self, model_ref: str, *, all_providers: list[Provider] | None = None) -> ModelInfo | None: for model in self.models: if model.is_match(model_ref): return model if self.fallback_model_providers and all_providers: for provider_id in self.fallback_model_providers: provider = next((p for p in all_providers if p.id == provider_id), None) if provider: # don't pass all_providers when falling back, so we can only have one step of fallback if model := provider._match_model(model_ref): return model return None def extract_usage(self, response_data: Any, *, api_flavor: str = 'default') -> tuple[str | None, Usage]: """Extract model name and usage information from a response. Args: response_data: The response data from the provider's API. api_flavor: The flavor of API used for this request. Raises: ValueError: If the response data is invalid or the API flavor is not found. Returns: tuple[str, Usage]: The extracted model name and usage information. """ if self.extractors is None: raise ValueError('No extraction logic defined for this provider') try: extractor = next(e for e in self.extractors if e.api_flavor == api_flavor) except StopIteration as e: fs = ', '.join(e.api_flavor for e in self.extractors) raise ValueError(f'Unknown api_flavor {api_flavor!r}, allowed values: {fs}') from e return extractor.extract(response_data) def summary(self) -> str: return f'Provider(id={self.id!r}, name={self.name!r}, ...)' @dataclass class UsageExtractorMapping: """Mappings from used to build usage.""" path: ExtractPath """Path to the value to extract""" dest: str """Destination field to store the extracted value. If multiple mappings point to the same destination, the values are summed. """ required: bool = True """Whether the value is required to be present in the response""" @dataclass class UsageExtractor: """Logic for extracting usage information from a response.""" root: ExtractPath """Path to the root of the usage information in the response, generally `usage`.""" mappings: list[UsageExtractorMapping] """Mappings from used to build usage.""" api_flavor: str = 'default' """Name of the API flavor, only needed when a provider has multiple flavors, e.g. OpenAI has `chat` and `responses`.""" model_path: ExtractPath = 'model' """Path to the model name in the response.""" _registry: InitVar[UnitRegistry | None] = None _reported_usage_keys: frozenset[str] = field(init=False, repr=False, compare=False) def __post_init__(self, _registry: UnitRegistry | None) -> None: reported_usage_keys = ( _registry._reported_usage_keys if _registry is not None else _reported_usage_keys() # pyright: ignore[reportPrivateUsage] ) object.__setattr__(self, '_reported_usage_keys', reported_usage_keys) invalid_destinations = {mapping.dest for mapping in self.mappings} - reported_usage_keys if invalid_destinations: bad_keys = ', '.join(sorted(invalid_destinations)) warnings.warn( f'Unsupported extractor destination for standard extraction: {bad_keys}', UserWarning, stacklevel=2, ) def extract(self, response_data: Any) -> tuple[str | None, Usage]: """Extract model name and usage information from a response. Args: response_data: The response data to extract usage information from, generally the decoded JSON response. Raises: ValueError: If no usage information is found at the root. Returns: tuple[str, Usage]: The extracted model name and usage information. """ model_name = _extract_path(self.model_path, response_data, str, False, []) root = self.root if isinstance(root, str): root = [root] usage_obj = cast(dict[str, Any], _extract_path(root, response_data, Mapping, True, [])) values: dict[str, UsageValue] = {} values_set = False supported_mappings = 0 for mapping in self.mappings: if mapping.dest not in self._reported_usage_keys: continue supported_mappings += 1 value = _extract_path(mapping.path, usage_obj, (Integral, float, Decimal), mapping.required, root) if value is not None: value = validate_usage_value(mapping.dest, value) if mapping.dest not in values: values[mapping.dest] = value else: values[mapping.dest] = add_usage_values(values[mapping.dest], value) values_set = True if supported_mappings and not values_set: raise ValueError(f'No usage information found at {self.root}') return model_name, Usage(**values) E = TypeVar('E') @overload def _extract_path( path: ExtractPath, data: Any, extract_type: type[E] | tuple[type[E], ...], required: Literal[True], data_path: Sequence[str | ArrayMatch], ) -> E: ... @overload def _extract_path( path: ExtractPath, data: Any, extract_type: type[E] | tuple[type[E], ...], required: Literal[False], data_path: Sequence[str | ArrayMatch], ) -> E | None: ... def _extract_path( path: ExtractPath, data: Any, extract_type: type[E] | tuple[type[E], ...], required: bool, data_path: Sequence[str | ArrayMatch], ) -> E | None: if isinstance(path, str): path = [path] *steps, last = path last = cast(str, last) error_path: list[str | ArrayMatch] = [] for step in steps: error_path.append(step) if isinstance(step, ArrayMatch): if not _is_sequence(data): if required: raise ValueError( f'Expected `{_dot_path(data_path, error_path)}` value to be a sequence, got {_type_name(data)}' ) else: return None if extracted_data := step.extract(data): data = extracted_data elif required: raise ValueError(f'Unable to find item at `{_dot_path(data_path, error_path)}`') else: return None else: if not _expect_mapping(data, required, data_path, error_path): return None try: data = data[step] except KeyError as e: if required: raise ValueError(f'Missing value at `{_dot_path(data_path, error_path)}`') from e else: return None if data is None and not required: return None if not _expect_mapping(data, required, data_path, error_path): return None try: value = data[last] except KeyError as e: if required: error_path.append(last) raise ValueError(f'Missing value at `{_dot_path(data_path, error_path)}`') from e else: return None else: if isinstance(value, extract_type): return value elif required: error_path.append(last) raise ValueError( f'Expected `{_dot_path(data_path, error_path)}` value to be a {_extract_type_name(extract_type)}, ' f'got {_type_name(value)}' ) def _expect_mapping( data: Any, required: bool, data_path: Sequence[str | ArrayMatch], error_path: Sequence[str | ArrayMatch] ) -> TypeGuard[Mapping[str, Any]]: if _is_mapping(data): return True if required: raise ValueError(f'Expected `{_dot_path(data_path, error_path)}` value to be a dict, got {_type_name(data)}') return False def _is_mapping(item: Any) -> TypeGuard[Mapping[str, Any]]: return isinstance(item, Mapping) def _is_sequence(item: Any) -> TypeGuard[Sequence[Any]]: return isinstance(item, Sequence) def _dot_path(data_path: Sequence[str | ArrayMatch], error_path: Sequence[str | ArrayMatch]) -> str: return '.'.join([str(p) for p in data_path] + [str(p) for p in error_path]) def _type_name(v: Any) -> str: return 'None' if v is None else type(v).__name__ def _extract_type_name(extract_type: type[object] | tuple[type[object], ...]) -> str: if isinstance(extract_type, tuple): return ' or '.join('int' if item is Integral else item.__name__ for item in extract_type) return extract_type.__name__ def _reported_usage_keys() -> frozenset[str]: from genai_prices.units import _get_registry # pyright: ignore[reportPrivateUsage] return _get_registry()._reported_usage_keys # pyright: ignore[reportPrivateUsage] def _reported_usage_key_order() -> tuple[str, ...]: from genai_prices.units import _get_registry # pyright: ignore[reportPrivateUsage] return _get_registry()._reported_usage_keys_in_order # pyright: ignore[reportPrivateUsage] def _raw_usage_value(obj: object, key: str) -> UsageValue | None: if _is_mapping(obj): value = obj.get(key) else: value = getattr(obj, key, None) if value is None: return None return validate_usage_value(key, value) @dataclass class ModelInfo: """Information about an LLM model""" id: str """Primary unique identifier for the model""" match: MatchLogic """Boolean logic for matching this model to any identifier which could be used to reference the model in API requests""" name: str | None = None """Name of the model""" description: str | None = None """Description of the model""" context_window: int | None = None """Maximum number of input tokens allowed for this model""" price_comments: str | None = None """Comments about the pricing of the model, especially challenges in representing the provider's pricing model.""" deprecated: bool | None = None """Flag indicating this model is deprecated by the provider but still functional.""" prices: ModelPrice | list[ConditionalPrice] = dataclasses.field(default_factory=list) """Set of prices for using this model. When multiple `ConditionalPrice`s are used, they are tried last to first to find a pricing model to use. E.g. later conditional prices take precedence over earlier ones. If no conditional models match the conditions, the first one is used. """ def is_match(self, model_ref: str) -> bool: return self.match.is_match(model_ref.lower()) def get_prices(self, request_timestamp: datetime) -> ModelPrice: if isinstance(self.prices, ModelPrice): return self.prices else: # reversed because the last price takes precedence for conditional_price in reversed(self.prices): if conditional_price.constraint is None or conditional_price.constraint.active(request_timestamp): return conditional_price.prices return self.prices[0].prices def calc_price( self, usage: AbstractUsage, provider: Provider, *, genai_request_timestamp: datetime | None = None, auto_update_timestamp: datetime | None = None, ) -> PriceCalculation: """Calculate the price for the given usage.""" genai_request_timestamp = genai_request_timestamp or datetime.now(tz=timezone.utc) model_price = self.get_prices(genai_request_timestamp) if provider.id == 'groq' and self.id in ('whisper-large-v3', 'whisper-large-v3-turbo'): usage = copy(Usage.from_raw(usage)) reported_seconds = usage.__dict__.get('audio_seconds') or usage.__dict__.get('input_audio_seconds') if reported_seconds is not None: billed_seconds = max(reported_seconds, 10) if reported_seconds > 0 else 0 usage.audio_seconds = billed_seconds usage.input_audio_seconds = billed_seconds price = model_price.calc_price(usage) return PriceCalculation( input_price=price['input_price'], output_price=price['output_price'], total_price=price['total_price'], model=self, provider=provider, model_price=model_price, auto_update_timestamp=auto_update_timestamp, ) def summary(self) -> str: return f'Model(id={self.id!r}, name={self.name!r}, ...)' class CalcPrice(TypedDict): input_price: Decimal output_price: Decimal total_price: Decimal class ModelPrice: """Set of prices for using a model""" def __init__( self, **price_kwargs: Decimal | TieredPrices | None, ) -> None: for key, value in price_kwargs.items(): object.__setattr__(self, key, value) def __repr__(self) -> str: parts = [f'{key}={value!r}' for key, value in self.__dict__.items() if value is not None] return f'{type(self).__name__}({", ".join(parts)})' def __eq__(self, other: object) -> Any: if type(other) is not type(self): return NotImplemented assert isinstance(other, ModelPrice) return self._comparable_values() == other._comparable_values() def _comparable_values(self) -> dict[str, object]: return {key: value for key, value in self.__dict__.items() if not key.startswith('_') and value is not None} def calc_price(self, usage: AbstractUsage) -> CalcPrice: """Calculate the price of usage in USD with this model price.""" from genai_prices.units import _get_registry # pyright: ignore[reportPrivateUsage] from genai_prices.validation import validate_priced_units registry = _get_registry() resolved_prices = _collect_resolved_model_prices(self, registry) validate_priced_units(tuple(unit for unit, _ in resolved_prices), registry) usage_data = Usage.from_raw(usage) priced_counts = _compute_registry_priced_counts(resolved_prices, usage_data) input_price = Decimal(0) output_price = Decimal(0) total_price = Decimal(0) # Reading input_tokens can trigger lazy inference errors; only do it when # tiered pricing actually needs the threshold. total_input_tokens = ( usage_data.input_tokens if any(isinstance(price, TieredPrices) for _, price in resolved_prices) else 0 ) for unit, price in resolved_prices: unit_price = calc_unit_price( price, priced_counts[unit.usage_key], total_input_tokens, unit.per, ) total_price += unit_price direction = unit.dimensions.get('direction') if direction == 'input': input_price += unit_price elif direction == 'output': output_price += unit_price return {'input_price': input_price, 'output_price': output_price, 'total_price': total_price} def __str__(self) -> str: from genai_prices.units import _get_registry # pyright: ignore[reportPrivateUsage] registry = _get_registry() parts: list[str] = [] for price_key in _iter_effective_model_price_keys(self, registry): value = getattr(self, price_key) if value is not None: # pragma: no branch try: unit = registry.unit_for_price_key(price_key) except KeyError: name = price_key.replace('_mtok', '').replace('_', ' ') if isinstance(value, TieredPrices): parts.append(f'${value.base}/{name} MTok (+tiers)') else: parts.append(f'${value}/{name} MTok') else: parts.append(_format_model_price_line(value, unit)) return ', '.join(parts) def __getattr__(self, name: str) -> Decimal | TieredPrices | None: if _is_registered_price_key(name): return None raise AttributeError(f'{type(self).__name__!r} object has no attribute {name!r}') def _is_registered_price_key(name: str) -> bool: from genai_prices.units import _get_registry # pyright: ignore[reportPrivateUsage] try: _get_registry().unit_for_price_key(name) except KeyError: return False else: return True def calc_unit_price( price: Decimal | TieredPrices | None, count: UsageValue | None, total_input_tokens: UsageValue, per: int ) -> Decimal: """Calculate the price for a unit count normalized by the unit's ``per`` value.""" if price is None or count is None: return Decimal(0) decimal_count = usage_value_as_decimal(count) if isinstance(price, TieredPrices): # Threshold-based pricing: tier is determined by total_input_tokens # Find the highest tier that applies based on total input tokens # When total_input_tokens is 0, no tier condition is met, so base rate is used applicable_price = price.base for tier in reversed(price.tiers): if total_input_tokens > tier.start: applicable_price = tier.price break unit_price = applicable_price * decimal_count else: unit_price = price * decimal_count return unit_price / per def _collect_resolved_model_prices( model_price: ModelPrice, registry: UnitRegistry ) -> tuple[tuple[UnitDef, Decimal | TieredPrices], ...]: stored_prices = [ (price_key, value) for price_key, value in _iter_model_price_attr_items(model_price, registry) if value is not None ] unknown_price_keys = { price_key for price_key, _ in stored_prices if price_key not in registry._all_price_keys # pyright: ignore[reportPrivateUsage] } if unknown_price_keys: bad_keys = ', '.join(sorted(unknown_price_keys)) warnings.warn( f'Unsupported price key for standard pricing: {bad_keys}', UserWarning, stacklevel=3, ) return tuple( (registry.unit_for_price_key(price_key), _validate_model_price_value(price_key, value)) for price_key, value in stored_prices if price_key not in unknown_price_keys ) def _validate_model_price_value(price_key: str, value: object) -> Decimal | TieredPrices: if _is_valid_price_decimal(value): return value if isinstance(value, TieredPrices): previous_start = -1 for tier in value.tiers: if ( type(tier.start) is not int or tier.start < 0 or tier.start < previous_start or not _is_valid_price_decimal(tier.price) ): break previous_start = tier.start else: if _is_valid_price_decimal(value.base): return value raise ValueError( f'Invalid price value for {price_key}: expected a finite non-negative Decimal or valid tiered prices' ) def _is_valid_price_decimal(value: object) -> TypeGuard[Decimal]: return isinstance(value, Decimal) and value.is_finite() and value >= 0 def _compute_registry_priced_counts( resolved_prices: Sequence[tuple[UnitDef, Decimal | TieredPrices]], usage: Usage ) -> dict[str, UsageValue]: from genai_prices.decompose import compute_leaf_values counts: dict[str, UsageValue] = {} priced_units_by_usage_key = {unit.usage_key: unit for unit, _ in resolved_prices if unit.usage_key != 'requests'} if priced_units_by_usage_key: counts.update(compute_leaf_values(set(priced_units_by_usage_key), usage, priced_units_by_usage_key)) if any(unit.usage_key == 'requests' for unit, _ in resolved_prices): counts['requests'] = 1 return counts def _iter_effective_model_price_keys(model_price: ModelPrice, registry: UnitRegistry) -> Iterator[str]: yielded_price_keys: set[str] = set() for unit in _iter_priced_registered_units(model_price, registry): yielded_price_keys.add(unit.price_key) yield unit.price_key for price_key, value in _iter_model_price_attr_items(model_price, registry): if value is not None and price_key not in yielded_price_keys: yield price_key def _iter_priced_registered_units(model_price: ModelPrice, registry: UnitRegistry) -> Iterator[UnitDef]: yield from (unit for unit in registry.units.values() if getattr(model_price, unit.price_key) is not None) def _iter_model_price_attr_items(model_price: ModelPrice, registry: UnitRegistry) -> Iterator[tuple[str, object]]: for key, value in model_price.__dict__.items(): if key.startswith('_'): continue if type(model_price) is not ModelPrice and key not in registry._all_price_keys: # pyright: ignore[reportPrivateUsage] continue yield key, value @dataclass class TieredPrices: """Pricing model when the amount paid varies by number of tokens. Uses threshold-based pricing where crossing a tier applies that rate to ALL tokens. This is the industry standard "cliff" model used by most providers (Anthropic, Google, OpenAI, etc.). Example: For a tier starting at 200K tokens: - Using 199,999 tokens: all tokens pay base rate - Using 200,001 tokens: all tokens pay tier rate (not just the tokens above 200K) """ base: Decimal """Base price in USD per million tokens, e.g. price until the first tier.""" tiers: list[Tier] """Extra price tiers.""" def __post_init__(self) -> None: """Ensure tiers are sorted in ascending order by start threshold.""" self.tiers.sort(key=lambda tier: tier.start) def _format_model_price_line(value: object, unit: UnitDef) -> str: from genai_prices.units import ( _unit_display_name, # pyright: ignore[reportPrivateUsage] _unit_per_label, # pyright: ignore[reportPrivateUsage] ) base_value = value.base if isinstance(value, TieredPrices) else value suffix = ' (+tiers)' if isinstance(value, TieredPrices) else '' unit_name = _unit_display_name(unit).lower() per_label = _unit_per_label(unit) if unit.dimensions.get('family') == 'requests': return f'${base_value} / {per_label} {unit_name}{suffix}' return f'${base_value}/{unit_name} {per_label}{suffix}' @dataclass class Tier: """Price tier""" start: int """Start of the tier""" price: Decimal """Price for this tier""" @dataclass class ConditionalPrice: """Pricing together with constraints that define when those prices should be used. The last active price (price where the constraints are met) is used. """ constraint: StartDateConstraint | TimeOfDateConstraint | None = None """Condition that determines when this price applies; when omitted, this price is always valid.""" _: dataclasses.KW_ONLY prices: ModelPrice """Prices for this condition.""" @dataclass class StartDateConstraint: """Constraint that defines when this price starts, e.g. when a new price is introduced.""" start_date: date """Date when this price starts""" def active(self, request_timestamp: datetime) -> bool: # UTC date, matching the JS package (instant vs UTC midnight). Naive timestamps are UTC. if request_timestamp.tzinfo is not None: request_timestamp = request_timestamp.astimezone(timezone.utc) return request_timestamp.date() >= self.start_date def _utc_timetz(value: time) -> time: """`value` as a UTC-aware time of day. Naive times are UTC.""" if value.tzinfo is None: return value.replace(tzinfo=timezone.utc) return datetime.combine(date(1970, 1, 1), value).astimezone(timezone.utc).timetz() @dataclass class TimeOfDateConstraint: """Constraint that defines a daily interval when a price applies, useful for off-peak pricing like deepseek.""" start_time: time """Start time of the interval.""" end_time: time """End time of the interval.""" def active(self, request_timestamp: datetime) -> bool: # Convert both sides to UTC before comparing so an offset that crosses midnight stays # inside the day. Naive timestamps and naive constraint times are UTC. if request_timestamp.tzinfo is None: request_timestamp = request_timestamp.replace(tzinfo=timezone.utc) request_time = request_timestamp.astimezone(timezone.utc).timetz() start_time = _utc_timetz(self.start_time) end_time = _utc_timetz(self.end_time) if end_time < start_time: return request_time >= start_time or request_time < end_time return start_time <= request_time < end_time @dataclass class ClauseStartsWith: starts_with: str def is_match(self, text: str) -> bool: return text.lower().startswith(self.starts_with.lower()) @dataclass class ClauseEndsWith: ends_with: str def is_match(self, text: str) -> bool: return text.lower().endswith(self.ends_with.lower()) @dataclass class ClauseContains: contains: str def is_match(self, text: str) -> bool: return self.contains.lower() in text.lower() @dataclass class ClauseRegex: regex: str def is_match(self, text: str) -> bool: return bool(re.search(self.regex, text)) @dataclass class ClauseEquals: equals: str def is_match(self, text: str) -> bool: return text.lower() == self.equals.lower() @dataclass class ClauseOr: or_: Annotated[list[MatchLogic], pydantic.Field(validation_alias='or')] def is_match(self, text: str) -> bool: return any(clause.is_match(text) for clause in self.or_) @dataclass class ClauseAnd: and_: Annotated[list[MatchLogic], pydantic.Field(validation_alias='and')] def is_match(self, text: str) -> bool: return all(clause.is_match(text) for clause in self.and_) _model_price_mapping_schema = pydantic.TypeAdapter(dict[str, Decimal | TieredPrices | None]) _providers_schema = pydantic.TypeAdapter( list[Provider], config=pydantic.ConfigDict(defer_build=True, arbitrary_types_allowed=True) ) def _providers_from_raw(raw_providers: Any, registry: UnitRegistry | None = None) -> list[Provider]: # pyright: ignore[reportUnusedFunction] normalized = _normalize_model_prices(raw_providers) if registry is not None: normalized = _inject_extractor_registry(normalized, registry) return _providers_schema.validate_python(normalized) def _inject_extractor_registry(raw_providers: Any, registry: UnitRegistry) -> Any: if not isinstance(raw_providers, list): return raw_providers providers: list[Any] = [] for raw_provider in cast(list[Any], raw_providers): if not isinstance(raw_provider, Mapping): providers.append(raw_provider) continue provider = dict(cast(Mapping[str, Any], raw_provider)) raw_extractors = provider.get('extractors') if isinstance(raw_extractors, list): extractors: list[Any] = [] for raw_extractor in cast(list[Any], raw_extractors): if isinstance(raw_extractor, Mapping): extractor = dict(cast(Mapping[str, Any], raw_extractor)) extractor['_registry'] = registry extractors.append(extractor) else: extractors.append(raw_extractor) provider['extractors'] = extractors providers.append(provider) return providers def _normalize_model_prices(value: Any) -> Any: if isinstance(value, list): return [_normalize_model_prices(item) for item in cast(list[Any], value)] if not isinstance(value, Mapping): return value normalized: dict[str, Any] = {} raw_mapping = cast(Mapping[str, Any], value) for key, raw_value in raw_mapping.items(): if key == 'prices': normalized[key] = _normalize_prices_field(raw_value) else: normalized[key] = _normalize_model_prices(raw_value) return normalized def _normalize_prices_field(value: Any) -> ModelPrice | list[Any]: if isinstance(value, list): return [_normalize_model_prices(item) for item in cast(list[Any], value)] if isinstance(value, ModelPrice): return value prices = _model_price_mapping_schema.validate_python(value) return ModelPrice(**cast(Any, prices))