scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-10 22:41:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 129

Merge pull request #31725 from Srivatsa03/time-based-cost-pricing

Browse source 
feat(cost): support time-based off-peak pricing in cost calculation
This commit is contained in:
Mateo Wang 2026-09-01 13:18:30 -07:00 • committed by GitHub
commit c913b09e66
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 1039 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
litellm/litellm_core_utils/litellm_logging.py
View file

@ -111,6 +111,7 @@ from litellm.types.mcp import MCPPostCallResponseObject
from litellm.types.prompts.init_prompts import PromptSpec
from litellm.types.rerank import RerankResponse
from litellm.types.utils import (
DEPLOYMENT_SCOPED_PRICING_FIELDS,
CachingDetails,
CallTypes,
CostBreakdown,
@ -255,6 +256,7 @@ _STANDARD_LOGGING_METADATA_KEYS: Final[frozenset[str]] = frozenset(StandardLoggi
# Cache custom pricing keys as frozenset for O(1) lookups instead of looping through 49 keys
_CUSTOM_PRICING_KEYS: Final[frozenset[str]] = frozenset(CustomPricingLiteLLMParams.model_fields.keys())
_MODEL_INFO_CUSTOM_PRICING_KEYS: Final[frozenset[str]] = _CUSTOM_PRICING_KEYS | DEPLOYMENT_SCOPED_PRICING_FIELDS
sentry_sdk_instance = None
capture_exception = None
@ -5033,7 +5035,9 @@ def use_custom_pricing_for_model(litellm_params: dict | None) -> bool:
"""
Check if the model uses custom pricing
Returns True if any of `SPECIAL_MODEL_INFO_PARAMS` are present in `litellm_params` or `model_info`
Returns True if any custom pricing field is present in `litellm_params`, or if
any custom pricing or deployment-scoped pricing field (such as
``off_peak_pricing``) is present in the metadata ``model_info``
"""
if litellm_params is None:
return False
@ -5051,7 +5055,7 @@ def use_custom_pricing_for_model(litellm_params: dict | None) -> bool:
model_info: dict = metadata.get("model_info", {}) or {}
if model_info:
matching_keys = _CUSTOM_PRICING_KEYS & model_info.keys()
matching_keys = _MODEL_INFO_CUSTOM_PRICING_KEYS & model_info.keys()
for key in matching_keys:
if model_info.get(key) is not None:
return True

215
litellm/litellm_core_utils/llm_cost_calc/utils.py
View file

@ -2,10 +2,12 @@
## Helper utilities for cost_per_token()
import re
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime, timezone, tzinfo
from types import MappingProxyType
from typing import Any, Final, Literal, TypedDict, cast
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
import litellm
from litellm._logging import verbose_logger
@ -290,10 +292,187 @@ def _get_tiered_base_costs(model_info: ModelInfo, usage: Usage) -> tuple[float,
)
def _is_within_off_peak_window(off_peak_hours_utc: str | Sequence[str], current_time: datetime | None = None) -> bool:
"""Return True if current_time (UTC, defaulting to now) falls inside any off-peak window.
off_peak_hours_utc is a "HH:MM-HH:MM" string in UTC, or a list of such strings for providers
with multiple daily windows (e.g. ["16:30-00:30", "04:00-06:00"]). A window may wrap past
midnight, and a window whose start equals its end covers the whole day. The start is
inclusive and the end is exclusive; malformed windows are ignored.
An aware current_time is converted to UTC. A naive one is taken to already be UTC rather
than being localised, so callers must pass datetime.now(timezone.utc), never datetime.now(),
or every window shifts by the host's offset.
"""
reference: Final = current_time if current_time is not None else datetime.now(timezone.utc)
now: Final = (reference.astimezone(timezone.utc) if reference.tzinfo is not None else reference).time()
windows: Final = (off_peak_hours_utc,) if isinstance(off_peak_hours_utc, str) else off_peak_hours_utc
for window in windows:
try:
start_str, end_str = window.split("-")
start = datetime.strptime(start_str.strip(), "%H:%M").replace(tzinfo=timezone.utc).time()
end = datetime.strptime(end_str.strip(), "%H:%M").replace(tzinfo=timezone.utc).time()
except (ValueError, AttributeError):
continue
if start < end:
if start <= now < end:
return True
elif now >= start or now < end:
return True
return False
_WEEKDAY_NUMBERS: Final = MappingProxyType(
{
"mon": 1,
"monday": 1,
"tue": 2,
"tues": 2,
"tuesday": 2,
"wed": 3,
"wednesday": 3,
"thu": 4,
"thur": 4,
"thurs": 4,
"thursday": 4,
"fri": 5,
"friday": 5,
"sat": 6,
"saturday": 6,
"sun": 7,
"sunday": 7,
}
)
def _normalize_weekday(value: object) -> int | None:
if isinstance(value, bool):
return None
if isinstance(value, int):
return value if 1 <= value <= 7 else None
if isinstance(value, str):
return _WEEKDAY_NUMBERS.get(value.strip().lower())
return None
def _weekday_calendar(weekday_timezone: object) -> tzinfo:
if isinstance(weekday_timezone, str) and weekday_timezone.strip():
try:
return ZoneInfo(weekday_timezone.strip())
except (ValueError, ZoneInfoNotFoundError):
return timezone.utc
return timezone.utc
def _matches_weekdays(reference_utc: datetime, weekdays: object, weekday_timezone: object) -> bool:
"""Return True when reference_utc falls on one of the rule's weekdays, read on the calendar
named by weekday_timezone (default UTC). An absent weekdays means every day. The calendar
matters even when UTC and vendor-local weekdays agree at every currently priced hour: a
window past 16:00 UTC is where an Asia/Shanghai weekday diverges from the UTC one.
"""
if weekdays is None:
return True
if isinstance(weekdays, str) or not isinstance(weekdays, Sequence):
return False
allowed: Final = frozenset(day for day in map(_normalize_weekday, weekdays) if day is not None)
return reference_utc.astimezone(_weekday_calendar(weekday_timezone)).isoweekday() in allowed
def _as_window_strings(value: object) -> tuple[str, ...]:
if isinstance(value, str):
return (value,)
if isinstance(value, Sequence):
return tuple(entry for entry in value if isinstance(entry, str))
return ()
def _is_off_peak(off_peak: Mapping[str, object], current_time: datetime | None = None) -> bool:
"""Return True when current_time (UTC, defaulting to now) is off-peak under the block's
rules: the flat hours_utc windows, which apply every day, or any entry in windows, whose
hours apply only on its weekdays.
"""
reference: Final = current_time if current_time is not None else datetime.now(timezone.utc)
reference_utc: Final = (
reference.astimezone(timezone.utc) if reference.tzinfo is not None else reference.replace(tzinfo=timezone.utc)
)
flat_windows: Final = _as_window_strings(off_peak.get("hours_utc"))
if flat_windows and _is_within_off_peak_window(flat_windows, reference_utc):
return True
windows: Final = off_peak.get("windows")
if isinstance(windows, str) or not isinstance(windows, Sequence):
return False
weekday_timezone: Final = off_peak.get("weekday_timezone")
for rule in windows:
if not isinstance(rule, Mapping):
continue
rule_windows = _as_window_strings(rule.get("hours_utc"))
if not rule_windows:
continue
if not _matches_weekdays(reference_utc, rule.get("weekdays"), weekday_timezone):
continue
if _is_within_off_peak_window(rule_windows, reference_utc):
return True
return False
def _coerce_off_peak_rate(value: object, default: float) -> float:
if isinstance(value, bool):
return default
if isinstance(value, (int, float)):
return float(value)
if isinstance(value, str):
try:
return float(value)
except ValueError:
return default
return default
def _apply_off_peak_pricing(
model_info: ModelInfo,
current_time: datetime | None,
prompt_base_cost: float,
completion_base_cost: float,
cache_read_cost: float,
) -> tuple[float, float, float]:
"""Swap in off-peak per-token rates when the current UTC time is inside one of the model's
off_peak_pricing rules, the every-day hours_utc windows or a day-of-week-qualified entry in
windows. An off-peak rate replaces the rate that would otherwise apply rather than
discounting it, so a model that also has tiered or above-threshold pricing bills the flat
off-peak rate for the whole request while the window is open. Any rate left unset in
off_peak_pricing falls back to the standard rate.
"""
off_peak: Final = model_info.get("off_peak_pricing")
if not isinstance(off_peak, Mapping) or not _is_off_peak(off_peak, current_time):
return prompt_base_cost, completion_base_cost, cache_read_cost
return (
_coerce_off_peak_rate(off_peak.get("input_cost_per_token"), prompt_base_cost),
_coerce_off_peak_rate(off_peak.get("output_cost_per_token"), completion_base_cost),
_coerce_off_peak_rate(off_peak.get("cache_read_input_token_cost"), cache_read_cost),
)
def _apply_off_peak_to_base_costs(
model_info: ModelInfo,
current_time: datetime | None,
base_costs: tuple[float, float, float, float, float],
) -> tuple[float, float, float, float, float]:
"""Apply off-peak rates to an already-resolved set of base costs, whichever pricing path
produced them. Cache-creation rates are passed through untouched, since off_peak_pricing
has no field for them.
"""
prompt, completion, cache_creation, cache_creation_above_1hr, cache_read = base_costs
off_peak_prompt, off_peak_completion, off_peak_cache_read = _apply_off_peak_pricing(
model_info, current_time, prompt, completion, cache_read
)
return (off_peak_prompt, off_peak_completion, cache_creation, cache_creation_above_1hr, off_peak_cache_read)
def _get_token_base_cost(
model_info: ModelInfo,
usage: Usage,
service_tier: str | None = None,
current_time: datetime | None = None,
*,
threshold_is_inclusive: bool = False,
) -> tuple[float, float, float, float, float]:
@ -311,7 +490,7 @@ def _get_token_base_cost(
"""
tiered_base_costs: Final = _get_tiered_base_costs(model_info=model_info, usage=usage)
if tiered_base_costs is not None:
return tiered_base_costs
return _apply_off_peak_to_base_costs(model_info, current_time, tiered_base_costs)
# Get service tier aware cost keys
input_cost_key: Final = _get_service_tier_cost_key("input_cost_per_token", service_tier)
@ -345,12 +524,16 @@ def _get_token_base_cost(
k for k in model_info if k.startswith("input_cost_per_token_above_") and not k.endswith(_SERVICE_TIER_SUFFIXES)
]
if not threshold_keys:
return (
prompt_base_cost,
completion_base_cost,
cache_creation_cost,
cache_creation_cost_above_1hr,
cache_read_cost,
return _apply_off_peak_to_base_costs(
model_info,
current_time,
(
prompt_base_cost,
completion_base_cost,
cache_creation_cost,
cache_creation_cost_above_1hr,
cache_read_cost,
),
)
# Only sort the threshold keys (typically 1-2 keys instead of 66+)
@ -451,12 +634,16 @@ def _get_token_base_cost(
except Exception:
continue
return (
prompt_base_cost,
completion_base_cost,
cache_creation_cost,
cache_creation_cost_above_1hr,
cache_read_cost,
return _apply_off_peak_to_base_costs(
model_info,
current_time,
(
prompt_base_cost,
completion_base_cost,
cache_creation_cost,
cache_creation_cost_above_1hr,
cache_read_cost,
),
)

61
litellm/router.py
View file

@ -8165,6 +8165,52 @@ class Router:
if backend_value is not None:
model_info[field] = backend_value
@staticmethod
def _inherit_builtin_base_rates_for_off_peak(
model_info: dict, # mutable-ok: cost-map entry filled in place
backend_model: str,
custom_llm_provider: str | None,
) -> None:
"""Fill missing pricing fields on a deployment entry that only sets
``off_peak_pricing``, from the backend model's built-in cost map entry.
Cost lookup selects the deployment-scoped entry over the shared backend
entry only when the deployment entry carries a base pricing field, and
``off_peak_pricing`` is deliberately kept off the shared entry, so a
deployment spelling out only its off-peak schedule would otherwise
never receive the discount. The backend model's entire canonical cost
map entry is copied, field by field, so threshold, tiered,
service-tier, cache, character, and per-second rates as well as
companion billing fields like ``web_search_billing_unit`` and the
regional uplift multipliers all carry over, and peak-hour billing
through the deployment entry matches the shared backend entry exactly.
The raw ``litellm.model_cost`` entry is the copy source rather than
``get_model_info``'s view of it, since that view synthesizes zero flat
token rates for backends without one and storing those would mark a
tiered-only backend explicitly priced free. Values are deep-copied to
keep the builtin entry isolated. User-specified fields always win;
no-op when any base pricing field is already set or the backend model
has no canonical entry.
"""
if not model_info.get("off_peak_pricing"):
return
if any(
model_info.get(field) is not None
for field in ("input_cost_per_token", "input_cost_per_second", "tiered_pricing")
):
return
try:
backend_info: Final = litellm.get_model_info(model=backend_model, custom_llm_provider=custom_llm_provider)
except Exception: # noqa: BLE001 # get_model_info raises plain Exception for an unmapped backend model
return
backend_entry: Final = litellm.model_cost.get(backend_info.get("key") or "")
if not isinstance(backend_entry, dict):
return
for field, backend_value in backend_entry.items():
if model_info.get(field) is not None or backend_value is None:
continue
model_info[field] = copy.deepcopy(backend_value)
@staticmethod
def _inherit_builtin_tiered_output_rate(
model_info: dict, backend_model: str, custom_llm_provider: str | None
@ -8253,6 +8299,11 @@ class Router:
if deployment.litellm_params.get(field) is not None:
_model_info[field] = deployment.litellm_params[field]
Router._inherit_builtin_base_rates_for_off_peak(
model_info=_model_info,
backend_model=deployment.litellm_params.model,
custom_llm_provider=deployment.litellm_params.custom_llm_provider,
)
if _model_info.get("input_cost_per_token") is not None:
Router._inherit_builtin_cache_pricing(
model_info=_model_info,
@ -8994,6 +9045,11 @@ class Router:
if field_value is not None:
_model_info_dict[field] = field_value
Router._inherit_builtin_base_rates_for_off_peak(
model_info=_model_info_dict,
backend_model=deployment.litellm_params.model,
custom_llm_provider=deployment.litellm_params.custom_llm_provider,
)
if _model_info_dict.get("input_cost_per_token") is not None:
Router._inherit_builtin_cache_pricing(
model_info=_model_info_dict,
@ -9249,6 +9305,11 @@ class Router:
field_value = deployment.litellm_params.get(field)
if field_value is not None:
model_info[field] = field_value
Router._inherit_builtin_base_rates_for_off_peak(
model_info=model_info,
backend_model=deployment.litellm_params.model,
custom_llm_provider=deployment.litellm_params.custom_llm_provider,
)
if model_info.get("input_cost_per_token") is not None:
Router._inherit_builtin_cache_pricing(
model_info=model_info,

50
litellm/types/utils.py
View file

@ -193,6 +193,38 @@ class AgenticLoopParams(TypedDict, total=False):
"""The LLM provider name (e.g., 'bedrock', 'anthropic')"""
class OffPeakWindow(TypedDict, total=False):
"""One off-peak rule: UTC time-of-day windows, optionally restricted to weekdays.
hours_utc is a "HH:MM-HH:MM" string in UTC, or a list of them; a window may wrap past
midnight and an equal-ended window covers the whole day. weekdays is a list of days the
rule applies on, as ISO-8601 numbers (1 = Monday .. 7 = Sunday) or English day names;
omitted means every day. The weekday is read on the calendar named by the block's
weekday_timezone.
"""
hours_utc: ReadOnly[str | Sequence[str]]
weekdays: ReadOnly[Sequence[int | str]]
class OffPeakPricing(TypedDict, total=False):
"""Time-windowed off-peak rates for providers that discount by time of day (e.g. DeepSeek).
hours_utc is a "HH:MM-HH:MM" string in UTC, or a list of them for multiple daily windows,
applying on every day of the week; a window may wrap past midnight. windows adds
day-of-week-qualified rules (e.g. weekend-only whole-day off-peak), matched as a union
with hours_utc. weekday_timezone names the IANA calendar weekdays are read on, defaulting
to UTC. Any rate left unset falls back to the standard rate.
"""
hours_utc: ReadOnly[str | Sequence[str]]
windows: ReadOnly[Sequence[OffPeakWindow]]
weekday_timezone: ReadOnly[str]
input_cost_per_token: ReadOnly[float]
output_cost_per_token: ReadOnly[float]
cache_read_input_token_cost: ReadOnly[float]
class ModelInfoBase(ProviderSpecificModelInfo, total=False):
key: Required[str] # the key in litellm.model_cost which is returned
@ -225,6 +257,7 @@ class ModelInfoBase(ProviderSpecificModelInfo, total=False):
# Smallest prefix this model will actually cache, whatever caching mechanism its provider uses.
# Absent means the provider-agnostic default applies; see MINIMUM_PROMPT_CACHE_TOKEN_COUNT.
prompt_cache_min_tokens: int | None
off_peak_pricing: ReadOnly[OffPeakPricing | None] # time-windowed off-peak rates
input_cost_per_character: float | None # only for vertex ai models
input_cost_per_audio_token: float | None
input_cost_per_token_above_128k_tokens: float | None # only for vertex ai models
@ -3486,17 +3519,22 @@ class CustomPricingLiteLLMParams(MirroredPricingParams):
return {k: v for k, v in model_info.items() if k not in cls.model_fields}
SHARED_BACKEND_MODEL_INFO_FIELDS: Final[frozenset[str]] = frozenset(
ModelInfoBase.__required_keys__ | ModelInfoBase.__optional_keys__
) - frozenset(CustomPricingLiteLLMParams.model_fields)
DEPLOYMENT_SCOPED_PRICING_FIELDS: Final[frozenset[str]] = frozenset({"off_peak_pricing"})
SHARED_BACKEND_MODEL_INFO_FIELDS: Final[frozenset[str]] = (
frozenset(ModelInfoBase.__required_keys__ | ModelInfoBase.__optional_keys__)
- frozenset(CustomPricingLiteLLMParams.model_fields)
- DEPLOYMENT_SCOPED_PRICING_FIELDS
)
def shared_backend_model_info(model_info: dict[str, Any]) -> dict[str, Any]:
"""Return only the fields safe to register under a shared ``{provider}/{model}``
key in ``litellm.model_cost``: cost-map schema fields (``ModelInfoBase``) minus
per-deployment pricing overrides. Per-deployment metadata (``id``,
``access_via_team_ids``, arbitrary custom keys) never belongs on the shared key;
it stays under the deployment's unique model id.
per-deployment pricing overrides and deployment-scoped pricing blocks such as
``off_peak_pricing``. Per-deployment metadata (``id``, ``access_via_team_ids``,
arbitrary custom keys) never belongs on the shared key; it stays under the
deployment's unique model id.
"""
return {k: v for k, v in model_info.items() if k in SHARED_BACKEND_MODEL_INFO_FIELDS}

6
litellm/utils.py
View file

@ -2869,10 +2869,9 @@ def _update_dictionary(existing_dict: dict, new_dict: dict) -> dict:
elif isinstance(v, dict):
existing_nested_dict = existing_dict.get(k)
if isinstance(existing_nested_dict, dict):
existing_nested_dict.update(v)
existing_dict[k] = existing_nested_dict
existing_dict[k] = {**existing_nested_dict, **v} # mutable-ok: copy-on-write merge
else:
existing_dict[k] = v
existing_dict[k] = dict(v) # mutable-ok: detached copy, never the caller's dict by reference
else:
existing_dict[k] = v
@ -5860,6 +5859,7 @@ def _get_model_info_helper(
cache_creation_input_token_cost_above_1hr=_model_info.get(
"cache_creation_input_token_cost_above_1hr", None
),
off_peak_pricing=_model_info.get("off_peak_pricing", None),
input_cost_per_character=_model_info.get("input_cost_per_character", None),
input_cost_per_token_above_128k_tokens=_model_info.get("input_cost_per_token_above_128k_tokens", None),
input_cost_per_token_above_200k_tokens=_model_info.get("input_cost_per_token_above_200k_tokens", None),

373
tests/test_litellm/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py
View file

@ -32,6 +32,8 @@ from litellm.litellm_core_utils.llm_cost_calc.utils import (
TokenTypeCostBreakdown,
_calculate_input_cost,
_get_token_base_cost,
_is_off_peak,
_is_within_off_peak_window,
calculate_cache_writing_cost,
generic_cost_per_token,
get_token_type_cost_breakdown,
@ -409,6 +411,377 @@ def test_get_token_base_cost_picks_highest_crossed_tier():
assert prompt_base_cost == 9e-6
def test_is_within_off_peak_window_same_day():
from datetime import datetime, timezone
window = "09:00-17:00"
assert _is_within_off_peak_window(window, datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc)) is True
assert _is_within_off_peak_window(window, datetime(2026, 1, 1, 8, 59, tzinfo=timezone.utc)) is False
assert _is_within_off_peak_window(window, datetime(2026, 1, 1, 9, 0, tzinfo=timezone.utc)) is True
assert _is_within_off_peak_window(window, datetime(2026, 1, 1, 17, 0, tzinfo=timezone.utc)) is False
def test_is_within_off_peak_window_wraps_midnight():
from datetime import datetime, timezone
window = "16:30-00:30"
assert _is_within_off_peak_window(window, datetime(2026, 1, 1, 18, 0, tzinfo=timezone.utc)) is True
assert _is_within_off_peak_window(window, datetime(2026, 1, 1, 0, 15, tzinfo=timezone.utc)) is True
assert _is_within_off_peak_window(window, datetime(2026, 1, 1, 16, 30, tzinfo=timezone.utc)) is True
assert _is_within_off_peak_window(window, datetime(2026, 1, 1, 0, 30, tzinfo=timezone.utc)) is False
assert _is_within_off_peak_window(window, datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc)) is False
def test_is_within_off_peak_window_equal_start_and_end_covers_whole_day():
"""An equal start and end is the natural way to spell off-peak all day. It used to take the
non-wrap branch, where start <= now < end can never hold, so it matched nothing and billed at
standard rates around the clock without raising or logging anything."""
from datetime import datetime, timezone
for window in ("00:00-00:00", "10:00-10:00"):
for hour in range(24):
assert (
_is_within_off_peak_window(window, datetime(2026, 1, 1, hour, 0, tzinfo=timezone.utc)) is True
), f"{window} should cover {hour:02d}:00"
def test_is_within_off_peak_window_multiple_windows():
from datetime import datetime, timezone
# Providers like DeepSeek V4 have more than one daily peak/off-peak window.
windows = ["01:00-05:00", "13:00-16:00"]
assert _is_within_off_peak_window(windows, datetime(2026, 1, 1, 3, 0, tzinfo=timezone.utc)) is True
assert _is_within_off_peak_window(windows, datetime(2026, 1, 1, 14, 30, tzinfo=timezone.utc)) is True
assert _is_within_off_peak_window(windows, datetime(2026, 1, 1, 9, 0, tzinfo=timezone.utc)) is False
# a malformed entry in the list is ignored, valid entries still match
assert _is_within_off_peak_window(["bad", "13:00-16:00"], datetime(2026, 1, 1, 14, 0, tzinfo=timezone.utc)) is True
assert _is_within_off_peak_window([], datetime(2026, 1, 1, 14, 0, tzinfo=timezone.utc)) is False
def test_is_within_off_peak_window_normalizes_timezone_aware_input():
from datetime import datetime, timedelta, timezone
# A caller may pass a non-UTC aware datetime; the window is UTC and must be
# evaluated in UTC, not against the caller's wall-clock. 09:00 at UTC+8 is
# 01:00 UTC, inside the 01:00-05:00 window.
tz_plus_8 = timezone(timedelta(hours=8))
assert _is_within_off_peak_window("01:00-05:00", datetime(2026, 1, 1, 9, 0, tzinfo=tz_plus_8)) is True
assert _is_within_off_peak_window("01:00-05:00", datetime(2026, 1, 1, 12, 0, tzinfo=tz_plus_8)) is True
# 06:00 at UTC+8 is 22:00 UTC the previous day, outside the window
assert _is_within_off_peak_window("01:00-05:00", datetime(2026, 1, 1, 6, 0, tzinfo=tz_plus_8)) is False
def test_is_within_off_peak_window_malformed_returns_false():
from datetime import datetime, timezone
now = datetime(2026, 1, 1, 18, 0, tzinfo=timezone.utc)
assert _is_within_off_peak_window("not-a-window", now) is False
assert _is_within_off_peak_window("16:30", now) is False
assert _is_within_off_peak_window("25:00-26:00", now) is False
def test_is_off_peak_weekday_qualified_windows_deepseek_schedule():
"""DeepSeek since 2026-08-23: peak is 01:00-04:00 and 06:00-10:00 UTC on weekdays only, with
weekends off-peak around the clock. The weekday axis is not a filter on one window set; on
two days of seven the off-peak window becomes the whole day, so the schedule needs two
day-qualified rules. The weekend instants inside would-be peak hours are the ones a
time-only implementation bills wrong."""
from datetime import datetime, timezone
deepseek = {
"windows": [
{"hours_utc": ["00:00-01:00", "04:00-06:00", "10:00-00:00"], "weekdays": [1, 2, 3, 4, 5]},
{"hours_utc": "00:00-00:00", "weekdays": [6, 7]},
],
}
peak_instants = [
datetime(2026, 8, 24, 1, 30, tzinfo=timezone.utc),
datetime(2026, 8, 26, 7, 0, tzinfo=timezone.utc),
datetime(2026, 8, 28, 9, 59, tzinfo=timezone.utc),
]
off_peak_instants = [
datetime(2026, 8, 23, 1, 30, tzinfo=timezone.utc),
datetime(2026, 8, 29, 2, 0, tzinfo=timezone.utc),
datetime(2026, 8, 30, 8, 0, tzinfo=timezone.utc),
datetime(2026, 8, 26, 5, 0, tzinfo=timezone.utc),
datetime(2026, 8, 28, 16, 30, tzinfo=timezone.utc),
datetime(2026, 8, 24, 0, 30, tzinfo=timezone.utc),
]
for when in peak_instants:
assert _is_off_peak(deepseek, when) is False, f"{when.isoformat()} should bill peak"
for when in off_peak_instants:
assert _is_off_peak(deepseek, when) is True, f"{when.isoformat()} should bill off-peak"
def test_is_off_peak_weekday_timezone_reads_vendor_calendar():
"""The UTC and Asia/Shanghai calendars only disagree about the date over 16:00-24:00 UTC, so
a window in that stretch is the one place a vendor-local weekday differs from a UTC one:
2026-08-28T16:30Z is Friday in UTC but already Saturday in Beijing."""
from datetime import datetime, timezone
shanghai_saturday = {
"weekday_timezone": "Asia/Shanghai",
"windows": [{"hours_utc": "16:00-17:00", "weekdays": [6]}],
}
assert _is_off_peak(shanghai_saturday, datetime(2026, 8, 28, 16, 30, tzinfo=timezone.utc)) is True
assert _is_off_peak(shanghai_saturday, datetime(2026, 8, 29, 16, 30, tzinfo=timezone.utc)) is False
def test_is_off_peak_weekdays_default_utc_calendar_and_accept_names():
from datetime import datetime, timezone
named_weekend = {"windows": [{"hours_utc": "00:00-00:00", "weekdays": ["Sat", "sunday"]}]}
assert _is_off_peak(named_weekend, datetime(2026, 8, 29, 12, 0, tzinfo=timezone.utc)) is True
assert _is_off_peak(named_weekend, datetime(2026, 8, 28, 12, 0, tzinfo=timezone.utc)) is False
utc_friday = {"windows": [{"hours_utc": "16:00-17:00", "weekdays": [5]}]}
assert _is_off_peak(utc_friday, datetime(2026, 8, 28, 16, 30, tzinfo=timezone.utc)) is True
assert _is_off_peak(utc_friday, datetime(2026, 8, 29, 16, 30, tzinfo=timezone.utc)) is False
def test_is_off_peak_naive_current_time_read_as_utc():
from datetime import datetime
block = {"windows": [{"hours_utc": "16:00-17:00", "weekdays": [5]}]}
assert _is_off_peak(block, datetime(2026, 8, 28, 16, 30)) is True
assert _is_off_peak(block, datetime(2026, 8, 29, 16, 30)) is False
def test_is_off_peak_invalid_weekday_timezone_falls_back_to_utc():
from datetime import datetime, timezone
block = {"weekday_timezone": "Not/AZone", "windows": [{"hours_utc": "16:00-17:00", "weekdays": [5]}]}
assert _is_off_peak(block, datetime(2026, 8, 28, 16, 30, tzinfo=timezone.utc)) is True
assert _is_off_peak(block, datetime(2026, 8, 29, 16, 30, tzinfo=timezone.utc)) is False
def test_is_off_peak_ignores_malformed_weekday_rules():
from datetime import datetime, timezone
when = datetime(2026, 8, 29, 12, 0, tzinfo=timezone.utc)
assert _is_off_peak({"windows": [{"hours_utc": "00:00-00:00", "weekdays": []}]}, when) is False
assert _is_off_peak({"windows": [{"hours_utc": "00:00-00:00", "weekdays": [0, 8, "noday", True]}]}, when) is False
assert _is_off_peak({"windows": [{"weekdays": [6]}]}, when) is False
assert _is_off_peak({"windows": [{"hours_utc": 1630}]}, when) is False
assert _is_off_peak({"windows": ["00:00-00:00"]}, when) is False
assert _is_off_peak({"windows": "00:00-00:00"}, when) is False
assert _is_off_peak({"hours_utc": 1630}, when) is False
assert _is_off_peak({}, when) is False
def test_is_off_peak_flat_hours_and_windows_are_a_union():
from datetime import datetime, timezone
block = {
"hours_utc": "04:00-06:00",
"windows": [{"hours_utc": "00:00-00:00", "weekdays": [7]}],
}
assert _is_off_peak(block, datetime(2026, 8, 28, 5, 0, tzinfo=timezone.utc)) is True
assert _is_off_peak(block, datetime(2026, 8, 30, 20, 0, tzinfo=timezone.utc)) is True
assert _is_off_peak(block, datetime(2026, 8, 28, 20, 0, tzinfo=timezone.utc)) is False
def test_get_token_base_cost_weekend_only_off_peak_rate():
from datetime import datetime, timezone
from typing import cast
from litellm.types.utils import ModelInfo
model_info = cast(
ModelInfo,
{
"input_cost_per_token": 1e-6,
"output_cost_per_token": 2e-6,
"off_peak_pricing": {
"windows": [
{"hours_utc": ["00:00-01:00", "04:00-06:00", "10:00-00:00"], "weekdays": [1, 2, 3, 4, 5]},
{"hours_utc": "00:00-00:00", "weekdays": [6, 7]},
],
"input_cost_per_token": 5e-7,
"output_cost_per_token": 1e-6,
},
},
)
usage = Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150)
saturday_peak_hours = _get_token_base_cost(
model_info, usage, current_time=datetime(2026, 8, 29, 2, 0, tzinfo=timezone.utc)
)
assert saturday_peak_hours[:2] == (5e-7, 1e-6)
monday_same_hours = _get_token_base_cost(
model_info, usage, current_time=datetime(2026, 8, 24, 2, 0, tzinfo=timezone.utc)
)
assert monday_same_hours[:2] == (1e-6, 2e-6)
def test_get_token_base_cost_applies_off_peak_pricing():
from datetime import datetime, timezone
from typing import cast
from litellm.types.utils import ModelInfo
model_info = cast(
ModelInfo,
{
"input_cost_per_token": 1e-6,
"output_cost_per_token": 2e-6,
"cache_read_input_token_cost": 1e-7,
"off_peak_pricing": {
"hours_utc": "16:30-00:30",
"input_cost_per_token": 5e-7,
"output_cost_per_token": 1e-6,
"cache_read_input_token_cost": 5e-8,
},
},
)
usage = Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150)
off_peak = _get_token_base_cost(model_info, usage, current_time=datetime(2026, 1, 1, 18, 0, tzinfo=timezone.utc))
assert off_peak[0] == 5e-7
assert off_peak[1] == 1e-6
assert off_peak[4] == 5e-8
peak = _get_token_base_cost(model_info, usage, current_time=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc))
assert peak[0] == 1e-6
assert peak[1] == 2e-6
assert peak[4] == 1e-7
def test_get_token_base_cost_non_mapping_off_peak_block_bills_standard_rates():
"""A truthy non-mapping off_peak_pricing value (a bare string or a list in
YAML) must bill standard rates rather than raising, matching how every
other malformed piece of the block behaves.
"""
from datetime import datetime, timezone
from typing import cast
from litellm.types.utils import ModelInfo
usage = Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150)
when = datetime(2026, 1, 1, 18, 0, tzinfo=timezone.utc)
for malformed_block in ("16:00-19:00", ["16:00-19:00"], 5e-7, True):
model_info = cast(
ModelInfo,
{
"input_cost_per_token": 1e-6,
"output_cost_per_token": 2e-6,
"off_peak_pricing": malformed_block,
},
)
result = _get_token_base_cost(model_info, usage, current_time=when)
assert result[0] == 1e-6
assert result[1] == 2e-6
def test_get_token_base_cost_off_peak_falls_back_to_standard_when_unset():
from datetime import datetime, timezone
from typing import cast
from litellm.types.utils import ModelInfo
model_info = cast(
ModelInfo,
{
"input_cost_per_token": 1e-6,
"output_cost_per_token": 2e-6,
"off_peak_pricing": {"hours_utc": "16:30-00:30", "input_cost_per_token": 5e-7},
},
)
usage = Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150)
result = _get_token_base_cost(model_info, usage, current_time=datetime(2026, 1, 1, 18, 0, tzinfo=timezone.utc))
assert result[0] == 5e-7
assert result[1] == 2e-6
def test_get_token_base_cost_off_peak_wins_over_threshold():
from datetime import datetime, timezone
from typing import cast
from litellm.types.utils import ModelInfo
model_info = cast(
ModelInfo,
{
"input_cost_per_token": 1e-6,
"output_cost_per_token": 2e-6,
"input_cost_per_token_above_200k_tokens": 3e-6,
"output_cost_per_token_above_200k_tokens": 4e-6,
"off_peak_pricing": {
"hours_utc": "16:30-00:30",
"input_cost_per_token": 5e-7,
"output_cost_per_token": 1e-6,
},
},
)
usage = Usage(prompt_tokens=250000, completion_tokens=250000, total_tokens=500000)
off_peak = _get_token_base_cost(model_info, usage, current_time=datetime(2026, 1, 1, 18, 0, tzinfo=timezone.utc))
assert off_peak[0] == 5e-7
assert off_peak[1] == 1e-6
peak = _get_token_base_cost(model_info, usage, current_time=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc))
assert peak[0] == 3e-6
assert peak[1] == 4e-6
def test_get_model_info_propagates_off_peak_fields():
model_name = "test-off-peak-model"
off_peak_pricing = {
"hours_utc": "16:30-00:30",
"input_cost_per_token": 5e-7,
"output_cost_per_token": 1e-6,
"cache_read_input_token_cost": 5e-8,
}
litellm.register_model(
{
model_name: {
"litellm_provider": "openai",
"mode": "chat",
"input_cost_per_token": 1e-6,
"output_cost_per_token": 2e-6,
"off_peak_pricing": off_peak_pricing,
}
}
)
info = litellm.get_model_info(model=model_name)
assert info["off_peak_pricing"] == off_peak_pricing
def test_get_token_base_cost_off_peak_wins_over_tiered_pricing():
"""Tiered pricing resolves base rates on its own path and returns early, so off-peak has to
be applied there too or a model carrying both would silently bill the tier rate all day."""
from datetime import datetime, timezone
model_name = "litellm-test-off-peak-tiered"
litellm.register_model(
{
model_name: {
"litellm_provider": "openai",
"mode": "chat",
"tiered_pricing": [
{"range": [0, 128000], "input_cost_per_token": 3e-6, "output_cost_per_token": 6e-6},
],
"off_peak_pricing": {
"hours_utc": "16:30-00:30",
"input_cost_per_token": 5e-7,
"output_cost_per_token": 1e-6,
},
}
}
)
info = litellm.get_model_info(model=model_name)
usage = Usage(prompt_tokens=1_000, completion_tokens=100, total_tokens=1_100)
inside = _get_token_base_cost(info, usage, current_time=datetime(2026, 1, 1, 18, 0, tzinfo=timezone.utc))
assert inside[:2] == (5e-7, 1e-6)
outside = _get_token_base_cost(info, usage, current_time=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc))
assert outside[:2] == (3e-6, 6e-6)
def test_generic_cost_per_token_gpt54_above_272k_tokens(_local_model_cost_map):
"""GPT-5.4/5.4-pro: prompts >272K input tokens priced at 2x input, 1.5x output."""
model = "gpt-5.4"

200
tests/test_litellm/test_register_model_custom_pricing.py
View file

@ -793,3 +793,203 @@ def test_embedding_direct_sdk_custom_pricing_still_registers_shared_key():
finally:
litellm.model_cost.pop(model_key, None)
_invalidate_model_cost_lowercase_map()
def test_update_dictionary_merges_nested_dicts_without_aliasing():
"""A nested dict must be merged copy-on-write: the pre-existing nested dict
object stays untouched, and the caller's incoming nested dict is never
inserted by reference into the merged result.
"""
from litellm.utils import _update_dictionary
existing_nested = {"hours_utc": "01:00-02:00"}
existing = {"off_peak_pricing": existing_nested}
incoming_nested = {"windows": [{"hours_utc": "16:00-19:00", "weekdays": [2]}]}
incoming = {"off_peak_pricing": incoming_nested}
merged = _update_dictionary(existing, incoming)
assert merged["off_peak_pricing"] == {
"hours_utc": "01:00-02:00",
"windows": [{"hours_utc": "16:00-19:00", "weekdays": [2]}],
}
assert existing_nested == {"hours_utc": "01:00-02:00"}
assert merged["off_peak_pricing"] is not incoming_nested
fresh = _update_dictionary({}, incoming)
assert fresh["off_peak_pricing"] == incoming_nested
assert fresh["off_peak_pricing"] is not incoming_nested
def test_router_deployments_sharing_backend_keep_their_own_off_peak_pricing():
"""Two deployments of the same backend model with different
``off_peak_pricing`` blocks must each keep their own schedule under their
unique model id, and neither block may leak onto the shared backend keys.
Before the fix, ``register_model`` inserted the first deployment's block by
reference into the built-in ``gpt-4o-mini`` entry, and the second
deployment's registration merged its keys into that same object, corrupting
the first deployment's schedule and polluting the built-in entry.
"""
from litellm import Router
active_block = {
"windows": [{"hours_utc": "16:00-19:00", "weekdays": [2]}],
"input_cost_per_token": 5e-07,
"output_cost_per_token": 1e-06,
}
inactive_block = {
"hours_utc": "05:00-06:00",
"input_cost_per_token": 5e-07,
"output_cost_per_token": 1e-06,
}
shared_keys = ["gpt-4o-mini", "openai/gpt-4o-mini"]
deployment_ids = ["offpeak-alias-dep-1", "offpeak-alias-dep-2"]
original_entries = _snapshot_model_cost_entries(shared_keys)
router = Router(
model_list=[
{
"model_name": "offpeak-active-weekday",
"litellm_params": {
"model": "openai/gpt-4o-mini",
"api_key": "fake-key-for-registration",
},
"model_info": {
"id": deployment_ids[0],
"input_cost_per_token": 1e-06,
"output_cost_per_token": 2e-06,
"off_peak_pricing": dict(active_block),
},
},
{
"model_name": "offpeak-inactive-hours",
"litellm_params": {
"model": "openai/gpt-4o-mini",
"api_key": "fake-key-for-registration",
},
"model_info": {
"id": deployment_ids[1],
"input_cost_per_token": 1e-06,
"output_cost_per_token": 2e-06,
"off_peak_pricing": dict(inactive_block),
},
},
]
)
try:
registered_first = litellm.model_cost[deployment_ids[0]]["off_peak_pricing"]
registered_second = litellm.model_cost[deployment_ids[1]]["off_peak_pricing"]
assert registered_first == active_block
assert registered_second == inactive_block
for shared_key in shared_keys:
shared_entry = litellm.model_cost.get(shared_key) or {}
assert not shared_entry.get("off_peak_pricing")
finally:
for deployment_id in deployment_ids:
litellm.model_cost.pop(deployment_id, None)
_restore_model_cost_entries(original_entries)
del router
def test_router_off_peak_only_deployment_inherits_builtin_base_rates():
"""A deployment that sets only ``off_peak_pricing`` on its model_info must
still be costed from its deployment-scoped entry: the base token rates are
inherited from the backend model's built-in cost map entry, since the
shared backend key deliberately never carries the off-peak block.
"""
from litellm import Router
block = {
"hours_utc": "00:00-00:00",
"input_cost_per_token": 5e-05,
"output_cost_per_token": 1e-04,
}
shared_keys = ["gpt-4o-mini", "openai/gpt-4o-mini"]
deployment_id = "offpeak-only-dep-1"
original_entries = _snapshot_model_cost_entries(shared_keys + [deployment_id])
builtin_info = litellm.get_model_info(model="openai/gpt-4o-mini")
router = Router(
model_list=[
{
"model_name": "offpeak-only",
"litellm_params": {
"model": "openai/gpt-4o-mini",
"api_key": "fake-key-for-registration",
},
"model_info": {"id": deployment_id, "off_peak_pricing": dict(block)},
}
]
)
try:
entry = litellm.model_cost[deployment_id]
assert entry["off_peak_pricing"] == block
assert entry["input_cost_per_token"] is not None
assert entry["input_cost_per_token"] == builtin_info["input_cost_per_token"]
assert entry["output_cost_per_token"] == builtin_info["output_cost_per_token"]
for shared_key in shared_keys:
shared_entry = litellm.model_cost.get(shared_key) or {}
assert not shared_entry.get("off_peak_pricing")
finally:
_restore_model_cost_entries(original_entries)
del router
def test_use_custom_pricing_for_model_sees_off_peak_only_model_info():
from litellm.litellm_core_utils.litellm_logging import use_custom_pricing_for_model
block = {"hours_utc": "00:00-00:00", "input_cost_per_token": 5e-05}
assert use_custom_pricing_for_model({"metadata": {"model_info": {"off_peak_pricing": block}}}) is True
assert use_custom_pricing_for_model({"metadata": {"model_info": {"off_peak_pricing": None}}}) is False
assert use_custom_pricing_for_model({"metadata": {"model_info": {"id": "some-id"}}}) is False
def test_completion_cost_applies_off_peak_only_deployment_pricing():
"""End to end through the cost calculator: with ``custom_pricing`` set and
a ``router_model_id`` whose entry carries only an always-on off-peak block,
the request bills at the block's rates rather than the shared backend rate.
"""
from litellm import Router
from litellm.types.utils import ModelResponse, Usage
block = {
"hours_utc": "00:00-00:00",
"input_cost_per_token": 5e-05,
"output_cost_per_token": 1e-04,
}
shared_keys = ["gpt-4o-mini", "openai/gpt-4o-mini"]
deployment_id = "offpeak-only-dep-2"
original_entries = _snapshot_model_cost_entries(shared_keys + [deployment_id])
router = Router(
model_list=[
{
"model_name": "offpeak-only",
"litellm_params": {
"model": "openai/gpt-4o-mini",
"api_key": "fake-key-for-registration",
},
"model_info": {"id": deployment_id, "off_peak_pricing": dict(block)},
}
]
)
try:
response = ModelResponse(
model="gpt-4o-mini",
usage=Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150),
)
cost = litellm.completion_cost(
completion_response=response,
model="openai/gpt-4o-mini",
custom_llm_provider="openai",
custom_pricing=True,
router_model_id=deployment_id,
)
assert cost == pytest.approx(100 * 5e-05 + 50 * 1e-04)
finally:
_restore_model_cost_entries(original_entries)
del router

151
tests/test_litellm/test_router_model_cost_isolation.py
View file

@ -538,6 +538,157 @@ def test_inherit_builtin_cache_pricing_noop_for_unknown_backend():
assert model_info == {"input_cost_per_token": 0.000003}
def test_inherit_builtin_base_rates_for_off_peak_fills_missing_rates():
"""Direct unit test of the helper: an entry carrying only an
off_peak_pricing block inherits the backend model's built-in base token
rates, so cost lookup via the deployment id can bill standard rates
outside the windows.
"""
backend_model = "gpt-4o-mini"
builtin_info = litellm.get_model_info(model=backend_model, custom_llm_provider="openai")
off_peak_block = {
"hours_utc": "00:00-00:00",
"input_cost_per_token": 5e-07,
"output_cost_per_token": 1e-06,
}
model_info = {"off_peak_pricing": off_peak_block}
Router._inherit_builtin_base_rates_for_off_peak(
model_info=model_info,
backend_model=backend_model,
custom_llm_provider="openai",
)
assert model_info["input_cost_per_token"] == builtin_info["input_cost_per_token"]
assert model_info["output_cost_per_token"] == builtin_info["output_cost_per_token"]
assert model_info["off_peak_pricing"] == off_peak_block
def test_inherit_builtin_base_rates_for_off_peak_carries_threshold_rates():
"""A backend with above-threshold pricing hands the whole rate structure to
the deployment entry, so peak-hour billing of large prompts through that
entry matches the shared backend entry instead of flattening to the base
rate.
"""
backend_model = "gemini/gemini-2.5-pro"
builtin_info = litellm.get_model_info(model=backend_model)
assert builtin_info["input_cost_per_token_above_200k_tokens"] is not None
model_info = {
"off_peak_pricing": {"hours_utc": "00:00-00:00", "input_cost_per_token": 5e-07},
}
Router._inherit_builtin_base_rates_for_off_peak(
model_info=model_info,
backend_model=backend_model,
custom_llm_provider="gemini",
)
assert model_info["input_cost_per_token"] == builtin_info["input_cost_per_token"]
assert (
model_info["input_cost_per_token_above_200k_tokens"]
== builtin_info["input_cost_per_token_above_200k_tokens"]
)
assert (
model_info["output_cost_per_token_above_200k_tokens"]
== builtin_info["output_cost_per_token_above_200k_tokens"]
)
def test_inherit_builtin_base_rates_for_off_peak_carries_companion_billing_fields():
"""Billing rules that are not literal cost rates, like the web search
billing unit, must ride along, or grounding and regional uplifts would
bill differently through the deployment entry than through the shared
backend entry.
"""
backend_model = "gemini-3-pro-image"
raw_entry = litellm.model_cost[backend_model]
assert raw_entry.get("web_search_billing_unit") is not None
model_info = {
"off_peak_pricing": {"hours_utc": "00:00-00:00", "input_cost_per_token": 5e-07},
}
Router._inherit_builtin_base_rates_for_off_peak(
model_info=model_info,
backend_model=backend_model,
custom_llm_provider=None,
)
assert model_info["web_search_billing_unit"] == raw_entry["web_search_billing_unit"]
assert model_info["input_cost_per_token"] == raw_entry["input_cost_per_token"]
def test_inherit_builtin_base_rates_for_off_peak_tiered_only_backend_stores_no_zero():
"""A tiered-only backend has no flat token rates; get_model_info synthesizes
zeros for them, and storing those would mark the deployment explicitly
priced free. The tier table itself must carry over as an isolated copy so
mutating the deployment entry never touches the shared cost map.
"""
backend_model = "dashscope/qwen-flash"
raw_tiers = litellm.model_cost[backend_model]["tiered_pricing"]
model_info = {
"off_peak_pricing": {"hours_utc": "00:00-00:00", "input_cost_per_token": 5e-07},
}
Router._inherit_builtin_base_rates_for_off_peak(
model_info=model_info,
backend_model=backend_model,
custom_llm_provider="dashscope",
)
assert model_info.get("input_cost_per_token") != 0
assert model_info.get("output_cost_per_token") != 0
assert model_info["tiered_pricing"] == raw_tiers
assert model_info["tiered_pricing"] is not raw_tiers
assert model_info["tiered_pricing"][0] is not raw_tiers[0]
original_first_tier = copy.deepcopy(raw_tiers[0])
model_info["tiered_pricing"][0]["input_cost_per_token"] = 123.0
assert raw_tiers[0] == original_first_tier
def test_inherit_builtin_base_rates_for_off_peak_leaves_explicit_rates_alone():
"""An entry that sets its own base rate beside the block already counts as
a full custom pricing entry; the helper must not mix builtin rates into it.
"""
model_info = {
"off_peak_pricing": {"hours_utc": "00:00-00:00", "input_cost_per_token": 5e-07},
"input_cost_per_token": 3e-06,
}
Router._inherit_builtin_base_rates_for_off_peak(
model_info=model_info,
backend_model="gpt-4o-mini",
custom_llm_provider="openai",
)
assert model_info["input_cost_per_token"] == 3e-06
assert "output_cost_per_token" not in model_info
def test_inherit_builtin_base_rates_for_off_peak_noop_without_block_or_backend():
"""Nothing happens without an off_peak_pricing block, and an unmapped
backend model leaves the entry unchanged rather than raising.
"""
plain_info = {"id": "dep-1"}
Router._inherit_builtin_base_rates_for_off_peak(
model_info=plain_info,
backend_model="gpt-4o-mini",
custom_llm_provider="openai",
)
assert plain_info == {"id": "dep-1"}
off_peak_info = {"off_peak_pricing": {"hours_utc": "00:00-00:00", "input_cost_per_token": 5e-07}}
Router._inherit_builtin_base_rates_for_off_peak(
model_info=off_peak_info,
backend_model="this-backend-model-does-not-exist-x9y8z7",
custom_llm_provider=None,
)
assert "input_cost_per_token" not in off_peak_info
def test_custom_pricing_field_denylist_covers_all_builtin_pricing_fields():
"""The shared-backend-key stripping in Router relies on
CustomPricingLiteLLMParams enumerating every per-deployment pricing field.