diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 350e5403e4c..eb223c2f988 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -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 diff --git a/litellm/litellm_core_utils/llm_cost_calc/utils.py b/litellm/litellm_core_utils/llm_cost_calc/utils.py index 19e3f624268..b34c416cd40 100644 --- a/litellm/litellm_core_utils/llm_cost_calc/utils.py +++ b/litellm/litellm_core_utils/llm_cost_calc/utils.py @@ -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, + ), ) diff --git a/litellm/router.py b/litellm/router.py index edff8294c3e..471a1116f44 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -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, diff --git a/litellm/types/utils.py b/litellm/types/utils.py index addb7b730de..3a0883b6607 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -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} diff --git a/litellm/utils.py b/litellm/utils.py index 028682661f7..2fdc4a45148 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -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), diff --git a/tests/test_litellm/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py b/tests/test_litellm/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py index 3c4121977de..0e1c832ebf5 100644 --- a/tests/test_litellm/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py +++ b/tests/test_litellm/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py @@ -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" diff --git a/tests/test_litellm/test_register_model_custom_pricing.py b/tests/test_litellm/test_register_model_custom_pricing.py index 39f498b4e58..452a15334ef 100644 --- a/tests/test_litellm/test_register_model_custom_pricing.py +++ b/tests/test_litellm/test_register_model_custom_pricing.py @@ -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 diff --git a/tests/test_litellm/test_router_model_cost_isolation.py b/tests/test_litellm/test_router_model_cost_isolation.py index b580b03574e..7b7a962bf00 100644 --- a/tests/test_litellm/test_router_model_cost_isolation.py +++ b/tests/test_litellm/test_router_model_cost_isolation.py @@ -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.