From 9fc77f12227e36a6d8e86336e1995931659f1c25 Mon Sep 17 00:00:00 2001 From: Srivatsa03 Date: Tue, 30 Jun 2026 12:10:48 -0500 Subject: [PATCH 01/14] feat(cost): support time-based off-peak pricing in cost calculation Some providers charge different per-token rates depending on the time of day. DeepSeek, for example, has historically discounted its chat and reasoner models during an off-peak window (16:30-00:30 UTC). LiteLLM's cost map only modeled static per-token pricing, so cost tracking could not stay accurate for these providers. This adds optional off-peak pricing to a model entry: input_cost_per_token_off_peak, output_cost_per_token_off_peak, cache_read_input_token_cost_off_peak, and an off_peak_hours_utc window expressed as "HH:MM-HH:MM" in UTC (the window may wrap past midnight). When the current UTC time falls inside the window, the cost calculator uses the off-peak rates and otherwise falls back to the standard rates, so existing models are unaffected. The fields are also accepted as custom pricing on a deployment, so they can be set from the proxy config or the SDK. The window check is a pure function that takes the current time as an argument, which keeps the regression tests deterministic without patching the clock. --- .../litellm_core_utils/llm_cost_calc/utils.py | 72 +++++++++ litellm/types/utils.py | 14 ++ litellm/utils.py | 1 + .../llm_cost_calc/test_llm_cost_calc_utils.py | 152 ++++++++++++++++++ 4 files changed, 239 insertions(+) diff --git a/litellm/litellm_core_utils/llm_cost_calc/utils.py b/litellm/litellm_core_utils/llm_cost_calc/utils.py index 19e3f624268..576efb18bb0 100644 --- a/litellm/litellm_core_utils/llm_cost_calc/utils.py +++ b/litellm/litellm_core_utils/llm_cost_calc/utils.py @@ -4,6 +4,7 @@ import re from collections.abc import Mapping from dataclasses import dataclass +from datetime import datetime, timezone from types import MappingProxyType from typing import Any, Final, Literal, TypedDict, cast @@ -290,10 +291,75 @@ def _get_tiered_base_costs(model_info: ModelInfo, usage: Usage) -> tuple[float, ) +def _is_within_off_peak_window(off_peak_hours_utc: str | list[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. The start is inclusive and the end is exclusive; malformed windows are ignored. + """ + if current_time is None: + current_time = datetime.now(timezone.utc) + now = current_time.time() + windows = [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").time() + end = datetime.strptime(end_str.strip(), "%H:%M").time() + except (ValueError, AttributeError): + continue + if start <= end: + if start <= now < end: + return True + elif now >= start or now < end: + 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 windows. Applied after threshold pricing so the discount is honored rather + than overwritten when a model combines off-peak and above-threshold rates. Any rate left + unset in off_peak_pricing falls back to the standard rate. + """ + off_peak = model_info.get("off_peak_pricing") + if not off_peak: + return prompt_base_cost, completion_base_cost, cache_read_cost + hours_utc = off_peak.get("hours_utc") + if not hours_utc or not _is_within_off_peak_window(hours_utc, 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 _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]: @@ -345,6 +411,9 @@ 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: + prompt_base_cost, completion_base_cost, cache_read_cost = _apply_off_peak_pricing( + model_info, current_time, prompt_base_cost, completion_base_cost, cache_read_cost + ) return ( prompt_base_cost, completion_base_cost, @@ -451,6 +520,9 @@ def _get_token_base_cost( except Exception: continue + prompt_base_cost, completion_base_cost, cache_read_cost = _apply_off_peak_pricing( + model_info, current_time, prompt_base_cost, completion_base_cost, cache_read_cost + ) return ( prompt_base_cost, completion_base_cost, diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 4bf8289d725..6583b125b62 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -193,6 +193,19 @@ class AgenticLoopParams(TypedDict, total=False): """The LLM provider name (e.g., 'bedrock', 'anthropic')""" +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; + a window may wrap past midnight. Any rate left unset falls back to the standard rate. + """ + + hours_utc: str | list[str] + input_cost_per_token: float + output_cost_per_token: float + cache_read_input_token_cost: float + + class ModelInfoBase(ProviderSpecificModelInfo, total=False): key: Required[str] # the key in litellm.model_cost which is returned @@ -225,6 +238,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: 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 diff --git a/litellm/utils.py b/litellm/utils.py index 5e9e115ed54..3389b8fcb78 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -5842,6 +5842,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..e226e255b05 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,7 @@ from litellm.litellm_core_utils.llm_cost_calc.utils import ( TokenTypeCostBreakdown, _calculate_input_cost, _get_token_base_cost, + _is_within_off_peak_window, calculate_cache_writing_cost, generic_cost_per_token, get_token_type_cost_breakdown, @@ -3946,3 +3947,154 @@ def test_route_image_generation_cost_falls_back_to_requested_size(monkeypatch, r ) assert cost == expected_cost + + +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_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_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_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_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 From f2c663515cd335482ab8bd29651e68cf102ab4cd Mon Sep 17 00:00:00 2001 From: Srivatsa03 Date: Sun, 2 Aug 2026 21:34:09 -0500 Subject: [PATCH 02/14] fix(cost): evaluate off-peak windows in UTC for timezone-aware inputs _is_within_off_peak_window used current_time.time(), which drops tzinfo, so a caller passing a non-UTC aware datetime had the window compared against local wall-clock instead of UTC. That silently mispriced off-peak requests. Normalize aware datetimes to UTC before comparing; naive datetimes stay as-is per the documented UTC contract. Added a regression test with a UTC+8 datetime that fails without the fix --- litellm/litellm_core_utils/llm_cost_calc/utils.py | 2 ++ .../llm_cost_calc/test_llm_cost_calc_utils.py | 13 +++++++++++++ 2 files changed, 15 insertions(+) diff --git a/litellm/litellm_core_utils/llm_cost_calc/utils.py b/litellm/litellm_core_utils/llm_cost_calc/utils.py index 576efb18bb0..ffa39850a5e 100644 --- a/litellm/litellm_core_utils/llm_cost_calc/utils.py +++ b/litellm/litellm_core_utils/llm_cost_calc/utils.py @@ -300,6 +300,8 @@ def _is_within_off_peak_window(off_peak_hours_utc: str | list[str], current_time """ if current_time is None: current_time = datetime.now(timezone.utc) + elif current_time.tzinfo is not None: + current_time = current_time.astimezone(timezone.utc) now = current_time.time() windows = [off_peak_hours_utc] if isinstance(off_peak_hours_utc, str) else off_peak_hours_utc for window in windows: 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 e226e255b05..463a871c6cc 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 @@ -3983,6 +3983,19 @@ def test_is_within_off_peak_window_multiple_windows(): 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 From c813386bb2a482a9fb4e5ece08267bfd5d1eb2b3 Mon Sep 17 00:00:00 2001 From: Srivatsa03 Date: Mon, 10 Aug 2026 21:47:08 -0500 Subject: [PATCH 03/14] refactor(cost): conform off-peak pricing to current lint budgets Rebasing onto litellm_internal_staging picked up stricter ceilings than this branch was written against. Bind the off-peak results to fresh names instead of reassigning the base costs, mark the new locals Final, avoid rebinding the current_time parameter, and make the window parse explicit about UTC so DTZ007, LIT010 and LIT011 all stay within budget --- .../litellm_core_utils/llm_cost_calc/utils.py | 37 +++++++++---------- litellm/types/utils.py | 10 ++--- 2 files changed, 22 insertions(+), 25 deletions(-) diff --git a/litellm/litellm_core_utils/llm_cost_calc/utils.py b/litellm/litellm_core_utils/llm_cost_calc/utils.py index ffa39850a5e..66f47affb4d 100644 --- a/litellm/litellm_core_utils/llm_cost_calc/utils.py +++ b/litellm/litellm_core_utils/llm_cost_calc/utils.py @@ -2,7 +2,7 @@ ## 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 from types import MappingProxyType @@ -291,24 +291,21 @@ def _get_tiered_base_costs(model_info: ModelInfo, usage: Usage) -> tuple[float, ) -def _is_within_off_peak_window(off_peak_hours_utc: str | list[str], current_time: datetime | None = None) -> bool: +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. The start is inclusive and the end is exclusive; malformed windows are ignored. """ - if current_time is None: - current_time = datetime.now(timezone.utc) - elif current_time.tzinfo is not None: - current_time = current_time.astimezone(timezone.utc) - now = current_time.time() - windows = [off_peak_hours_utc] if isinstance(off_peak_hours_utc, str) else off_peak_hours_utc + 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").time() - end = datetime.strptime(end_str.strip(), "%H:%M").time() + 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: @@ -344,10 +341,10 @@ def _apply_off_peak_pricing( than overwritten when a model combines off-peak and above-threshold rates. Any rate left unset in off_peak_pricing falls back to the standard rate. """ - off_peak = model_info.get("off_peak_pricing") + off_peak: Final = model_info.get("off_peak_pricing") if not off_peak: return prompt_base_cost, completion_base_cost, cache_read_cost - hours_utc = off_peak.get("hours_utc") + hours_utc: Final = off_peak.get("hours_utc") if not hours_utc or not _is_within_off_peak_window(hours_utc, current_time): return prompt_base_cost, completion_base_cost, cache_read_cost return ( @@ -413,15 +410,15 @@ 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: - prompt_base_cost, completion_base_cost, cache_read_cost = _apply_off_peak_pricing( + off_peak_prompt_cost, off_peak_completion_cost, off_peak_cache_read_cost = _apply_off_peak_pricing( model_info, current_time, prompt_base_cost, completion_base_cost, cache_read_cost ) return ( - prompt_base_cost, - completion_base_cost, + off_peak_prompt_cost, + off_peak_completion_cost, cache_creation_cost, cache_creation_cost_above_1hr, - cache_read_cost, + off_peak_cache_read_cost, ) # Only sort the threshold keys (typically 1-2 keys instead of 66+) @@ -522,15 +519,15 @@ def _get_token_base_cost( except Exception: continue - prompt_base_cost, completion_base_cost, cache_read_cost = _apply_off_peak_pricing( + discounted_prompt_cost, discounted_completion_cost, discounted_cache_read_cost = _apply_off_peak_pricing( model_info, current_time, prompt_base_cost, completion_base_cost, cache_read_cost ) return ( - prompt_base_cost, - completion_base_cost, + discounted_prompt_cost, + discounted_completion_cost, cache_creation_cost, cache_creation_cost_above_1hr, - cache_read_cost, + discounted_cache_read_cost, ) diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 6583b125b62..3ab3a2382dc 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -200,10 +200,10 @@ class OffPeakPricing(TypedDict, total=False): a window may wrap past midnight. Any rate left unset falls back to the standard rate. """ - hours_utc: str | list[str] - input_cost_per_token: float - output_cost_per_token: float - cache_read_input_token_cost: float + hours_utc: ReadOnly[str | Sequence[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): @@ -238,7 +238,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: OffPeakPricing | None # time-windowed off-peak rates + 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 From d302301a4e6a250170ef96a621026af78760be4a Mon Sep 17 00:00:00 2001 From: Srivatsa03 Date: Thu, 13 Aug 2026 17:28:00 -0500 Subject: [PATCH 04/14] test(cost): move off-peak tests beside the related cost tests They sat at the end of the file, which is where everyone else appends too, so this branch picked up a conflict there on nearly every rebase. Grouping them with the other _get_token_base_cost test keeps them clear of that churn and next to the code they cover. Pure move, no test changes --- .../llm_cost_calc/test_llm_cost_calc_utils.py | 328 +++++++++--------- 1 file changed, 164 insertions(+), 164 deletions(-) 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 463a871c6cc..f1340df8b69 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 @@ -410,6 +410,170 @@ 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_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_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_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_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" @@ -3947,167 +4111,3 @@ def test_route_image_generation_cost_falls_back_to_requested_size(monkeypatch, r ) assert cost == expected_cost - - -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_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_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_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 From 4f174ffdd1022a0ffa21c2ae6bd2154011a9368e Mon Sep 17 00:00:00 2001 From: Srivatsa03 Date: Sat, 15 Aug 2026 10:49:51 -0500 Subject: [PATCH 05/14] fix(cost): apply off-peak rates on the tiered-pricing path Tiered pricing resolves its own base rates and returns early, before the off-peak swap ran, so a model carrying both tiered_pricing and off_peak_pricing billed the tier rate around the clock. Route every base-cost path through one helper so the window applies wherever the rates came from, and say plainly in the docstring that an off-peak rate replaces the rate it lands on rather than discounting it --- .../litellm_core_utils/llm_cost_calc/utils.py | 63 ++++++++++++------- .../llm_cost_calc/test_llm_cost_calc_utils.py | 32 ++++++++++ 2 files changed, 73 insertions(+), 22 deletions(-) diff --git a/litellm/litellm_core_utils/llm_cost_calc/utils.py b/litellm/litellm_core_utils/llm_cost_calc/utils.py index 66f47affb4d..be289f620cf 100644 --- a/litellm/litellm_core_utils/llm_cost_calc/utils.py +++ b/litellm/litellm_core_utils/llm_cost_calc/utils.py @@ -337,9 +337,10 @@ def _apply_off_peak_pricing( 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 windows. Applied after threshold pricing so the discount is honored rather - than overwritten when a model combines off-peak and above-threshold rates. Any rate left - unset in off_peak_pricing falls back to the standard rate. + off_peak_pricing 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 off_peak: @@ -354,6 +355,22 @@ def _apply_off_peak_pricing( ) +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, @@ -376,7 +393,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) @@ -410,15 +427,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: - off_peak_prompt_cost, off_peak_completion_cost, off_peak_cache_read_cost = _apply_off_peak_pricing( - model_info, current_time, prompt_base_cost, completion_base_cost, cache_read_cost - ) - return ( - off_peak_prompt_cost, - off_peak_completion_cost, - cache_creation_cost, - cache_creation_cost_above_1hr, - off_peak_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+) @@ -519,15 +537,16 @@ def _get_token_base_cost( except Exception: continue - discounted_prompt_cost, discounted_completion_cost, discounted_cache_read_cost = _apply_off_peak_pricing( - model_info, current_time, prompt_base_cost, completion_base_cost, cache_read_cost - ) - return ( - discounted_prompt_cost, - discounted_completion_cost, - cache_creation_cost, - cache_creation_cost_above_1hr, - discounted_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/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 f1340df8b69..8f348b36d35 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 @@ -574,6 +574,38 @@ def test_get_model_info_propagates_off_peak_fields(): 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" From 27aefade5fc9c9276cf752f949376a32ddedf3a7 Mon Sep 17 00:00:00 2001 From: Srivatsa03 Date: Sat, 22 Aug 2026 23:11:57 -0500 Subject: [PATCH 06/14] fix(cost): treat an equal-ended off-peak window as the whole day A window whose start equals its end is the natural way to spell off-peak all day, and the docstring's promise that a window may wrap past midnight invites it. It took the non-wrap branch instead, where start <= now < end can never hold, so it matched nothing. It parses cleanly, so it never reached the branch that ignores malformed windows: no exception, no log, and the model billed at standard rates around the clock while the config said otherwise. Let equality fall through to the wrap branch, which covers every instant, and say so in the docstring. Reported by @xyzs996 in review. --- litellm/litellm_core_utils/llm_cost_calc/utils.py | 5 +++-- .../llm_cost_calc/test_llm_cost_calc_utils.py | 13 +++++++++++++ 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/litellm/litellm_core_utils/llm_cost_calc/utils.py b/litellm/litellm_core_utils/llm_cost_calc/utils.py index be289f620cf..30c25253eea 100644 --- a/litellm/litellm_core_utils/llm_cost_calc/utils.py +++ b/litellm/litellm_core_utils/llm_cost_calc/utils.py @@ -296,7 +296,8 @@ def _is_within_off_peak_window(off_peak_hours_utc: str | Sequence[str], current_ 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. The start is inclusive and the end is exclusive; malformed windows are ignored. + 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. """ 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() @@ -308,7 +309,7 @@ def _is_within_off_peak_window(off_peak_hours_utc: str | Sequence[str], current_ end = datetime.strptime(end_str.strip(), "%H:%M").replace(tzinfo=timezone.utc).time() except (ValueError, AttributeError): continue - if start <= end: + if start < end: if start <= now < end: return True elif now >= start or now < end: 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 8f348b36d35..c9ed5936f03 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 @@ -431,6 +431,19 @@ def test_is_within_off_peak_window_wraps_midnight(): 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 From cc3ea1fb08387a511b7cc243613df29d41b97062 Mon Sep 17 00:00:00 2001 From: Srivatsa03 Date: Sat, 22 Aug 2026 23:40:03 -0500 Subject: [PATCH 07/14] docs(cost): state that a naive off-peak current_time is read as UTC An aware value is converted, a naive one is taken to already be UTC rather than localised. Nothing signals the difference, so a caller passing datetime.now() instead of datetime.now(timezone.utc) shifts every window by the host's offset and bills silently wrong. Say so where a caller will read it. Reported by @xyzs996 in review. --- litellm/litellm_core_utils/llm_cost_calc/utils.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/litellm/litellm_core_utils/llm_cost_calc/utils.py b/litellm/litellm_core_utils/llm_cost_calc/utils.py index 30c25253eea..21680129ed4 100644 --- a/litellm/litellm_core_utils/llm_cost_calc/utils.py +++ b/litellm/litellm_core_utils/llm_cost_calc/utils.py @@ -298,6 +298,10 @@ def _is_within_off_peak_window(off_peak_hours_utc: str | Sequence[str], current_ 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() From 7abed91523f8a1bfb79697949f070e367a1069c0 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Tue, 1 Sep 2026 10:11:25 -0700 Subject: [PATCH 08/14] feat(cost): support day-of-week qualified off-peak windows --- .../litellm_core_utils/llm_cost_calc/utils.py | 108 ++++++++++++-- litellm/types/utils.py | 27 +++- .../llm_cost_calc/test_llm_cost_calc_utils.py | 136 ++++++++++++++++++ 3 files changed, 259 insertions(+), 12 deletions(-) diff --git a/litellm/litellm_core_utils/llm_cost_calc/utils.py b/litellm/litellm_core_utils/llm_cost_calc/utils.py index 21680129ed4..c968fac254e 100644 --- a/litellm/litellm_core_utils/llm_cost_calc/utils.py +++ b/litellm/litellm_core_utils/llm_cost_calc/utils.py @@ -4,9 +4,10 @@ import re from collections.abc import Mapping, Sequence from dataclasses import dataclass -from datetime import datetime, timezone +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 @@ -321,6 +322,99 @@ def _is_within_off_peak_window(off_peak_hours_utc: str | Sequence[str], current_ 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 @@ -342,16 +436,14 @@ def _apply_off_peak_pricing( 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 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 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 off_peak: - return prompt_base_cost, completion_base_cost, cache_read_cost - hours_utc: Final = off_peak.get("hours_utc") - if not hours_utc or not _is_within_off_peak_window(hours_utc, current_time): + if not off_peak 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), diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 3ab3a2382dc..17051714c25 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -193,14 +193,33 @@ class AgenticLoopParams(TypedDict, total=False): """The LLM provider name (e.g., 'bedrock', 'anthropic')""" -class OffPeakPricing(TypedDict, total=False): - """Time-windowed off-peak rates for providers that discount by time of day (e.g. DeepSeek). +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 for multiple daily windows; - a window may wrap past midnight. Any rate left unset falls back to the standard rate. + 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] 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 c9ed5936f03..501a9314c90 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,7 @@ 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, @@ -479,6 +480,141 @@ def test_is_within_off_peak_window_malformed_returns_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 1ba13fcc25ff2401d6ed87541d97ab275e3b1430 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Tue, 1 Sep 2026 11:17:55 -0700 Subject: [PATCH 09/14] fix(cost): keep off_peak_pricing scoped to its deployment register_model inserted the first deployment's off_peak_pricing dict by reference into the shared backend cost-map entry, and later deployments sharing that backend merged their schedules into the same object, corrupting the first deployment's schedule and polluting the built-in entry. Nested dicts now merge copy-on-write, and off_peak_pricing stays off the shared backend keys. --- litellm/types/utils.py | 17 ++-- litellm/utils.py | 5 +- .../test_register_model_custom_pricing.py | 98 +++++++++++++++++++ 3 files changed, 111 insertions(+), 9 deletions(-) diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 17051714c25..4cba3193e43 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -3506,17 +3506,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 3389b8fcb78..2014df6b17c 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -2851,10 +2851,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 diff --git a/tests/test_litellm/test_register_model_custom_pricing.py b/tests/test_litellm/test_register_model_custom_pricing.py index 39f498b4e58..87e9895806f 100644 --- a/tests/test_litellm/test_register_model_custom_pricing.py +++ b/tests/test_litellm/test_register_model_custom_pricing.py @@ -793,3 +793,101 @@ 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 From b0751169ebdb4df24a9ddafacb4932995159ab3b Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Tue, 1 Sep 2026 12:14:46 -0700 Subject: [PATCH 10/14] fix(cost): bill off-peak rates for deployments that set only off_peak_pricing Cost lookup selects the deployment-scoped cost map entry only when custom pricing is detected and the entry carries a base pricing field. A deployment whose model_info held nothing but off_peak_pricing failed both conditions, so its schedule was silently ignored and every request billed at the shared backend rate. use_custom_pricing_for_model now also treats deployment-scoped pricing fields in the metadata model_info as custom pricing, and the router inherits the backend model's built-in base token rates onto such an entry at registration, which also lets cache pricing inheritance apply. Regression tests cover the registration, the detection, and the costed request end to end. --- litellm/litellm_core_utils/litellm_logging.py | 8 +- litellm/router.py | 49 +++++++++ .../test_register_model_custom_pricing.py | 102 ++++++++++++++++++ 3 files changed, 157 insertions(+), 2 deletions(-) diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 97c4d038734..2312eb6130f 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 @@ -5030,7 +5032,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 @@ -5048,7 +5052,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/router.py b/litellm/router.py index c93c1753f0e..f3563d9c387 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -8163,6 +8163,40 @@ 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 base token rates 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. User-specified rates 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 + for field in ("input_cost_per_token", "output_cost_per_token"): + if model_info.get(field) is None: + backend_value = backend_info.get(field) + if backend_value is not None: + model_info[field] = backend_value + @staticmethod def _inherit_builtin_tiered_output_rate( model_info: dict, backend_model: str, custom_llm_provider: str | None @@ -8251,6 +8285,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, @@ -8992,6 +9031,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, @@ -9246,6 +9290,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/tests/test_litellm/test_register_model_custom_pricing.py b/tests/test_litellm/test_register_model_custom_pricing.py index 87e9895806f..452a15334ef 100644 --- a/tests/test_litellm/test_register_model_custom_pricing.py +++ b/tests/test_litellm/test_register_model_custom_pricing.py @@ -891,3 +891,105 @@ def test_router_deployments_sharing_backend_keep_their_own_off_peak_pricing(): 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 From 8d0e7aee2ff61e0100137feff993cb1a34a8ad52 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Tue, 1 Sep 2026 12:20:09 -0700 Subject: [PATCH 11/14] test(router): cover _inherit_builtin_base_rates_for_off_peak directly The router_code_coverage gate only counts calls made from test files with router in the filename, so the helper needs direct unit tests beside the other inheritance helpers in test_router_model_cost_isolation.py: fills missing base rates from the builtin entry, leaves explicit rates alone, and no-ops without a block or for an unmapped backend model. --- .../test_router_model_cost_isolation.py | 66 +++++++++++++++++++ 1 file changed, 66 insertions(+) diff --git a/tests/test_litellm/test_router_model_cost_isolation.py b/tests/test_litellm/test_router_model_cost_isolation.py index b580b03574e..c8adc0af431 100644 --- a/tests/test_litellm/test_router_model_cost_isolation.py +++ b/tests/test_litellm/test_router_model_cost_isolation.py @@ -538,6 +538,72 @@ 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_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. From 4875872fe559fb9d30a9efe3766106459dedd914 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Tue, 1 Sep 2026 12:25:39 -0700 Subject: [PATCH 12/14] fix(cost): treat a non-mapping off_peak_pricing value as never off-peak A bare string or list under off_peak_pricing in YAML passed the truthy guard and crashed _is_off_peak with AttributeError, breaking cost calculation for that deployment. Malformed pieces of the block are documented to not match rather than error, so guard the block itself the same way and bill standard rates. --- .../litellm_core_utils/llm_cost_calc/utils.py | 2 +- .../llm_cost_calc/test_llm_cost_calc_utils.py | 27 +++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/litellm/litellm_core_utils/llm_cost_calc/utils.py b/litellm/litellm_core_utils/llm_cost_calc/utils.py index c968fac254e..b34c416cd40 100644 --- a/litellm/litellm_core_utils/llm_cost_calc/utils.py +++ b/litellm/litellm_core_utils/llm_cost_calc/utils.py @@ -443,7 +443,7 @@ def _apply_off_peak_pricing( off_peak_pricing falls back to the standard rate. """ off_peak: Final = model_info.get("off_peak_pricing") - if not off_peak or not _is_off_peak(off_peak, current_time): + 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), 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 501a9314c90..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 @@ -648,6 +648,33 @@ def test_get_token_base_cost_applies_off_peak_pricing(): 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 0ec3e936b73a9ebc4445c025a1e4d15a940e29de Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Tue, 1 Sep 2026 12:47:26 -0700 Subject: [PATCH 13/14] fix(cost): inherit the backend's full price structure for off-peak-only deployments Copying only the flat token rates dropped threshold, tiered, service-tier, cache, character, and per-second rates from peak-hour billing once cost lookup switched to the deployment entry, and get_model_info synthesizes zero flat rates for backends without one, which would have marked tiered-only backends explicitly priced free. Copy every price-bearing field instead, deep-copied, rejecting the synthesized zeros the way _inherit_builtin_tiered_output_rate already does. --- litellm/router.py | 28 ++++++--- .../test_router_model_cost_isolation.py | 61 +++++++++++++++++++ 2 files changed, 80 insertions(+), 9 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index f3563d9c387..dd1cb51d9af 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -8169,16 +8169,23 @@ class Router: backend_model: str, custom_llm_provider: str | None, ) -> None: - """Fill missing base token rates on a deployment entry that only sets + """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. User-specified rates always win; no-op when - any base pricing field is already set or the backend model has no - canonical entry. + never receive the discount. Every price-bearing backend field is + copied, not just the flat token rates: threshold, tiered, service-tier, + cache, character, and per-second rates all carry over, so peak-hour + billing through the deployment entry matches the shared backend entry + exactly. Values are deep-copied to keep the builtin entry isolated, and + a flat token rate ``get_model_info`` synthesized as zero for a backend + without one is rejected, like ``_inherit_builtin_tiered_output_rate`` + does, so a tiered-only backend is never marked explicitly priced free. + User-specified rates 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 @@ -8191,11 +8198,14 @@ class Router: 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 - for field in ("input_cost_per_token", "output_cost_per_token"): - if model_info.get(field) is None: - backend_value = backend_info.get(field) - if backend_value is not None: - model_info[field] = backend_value + for field, backend_value in backend_info.items(): + if "cost" not in field and field != "tiered_pricing": + continue + if model_info.get(field) is not None or backend_value is None: + continue + if field in ("input_cost_per_token", "output_cost_per_token") and not backend_value: + continue + model_info[field] = copy.deepcopy(backend_value) @staticmethod def _inherit_builtin_tiered_output_rate( diff --git a/tests/test_litellm/test_router_model_cost_isolation.py b/tests/test_litellm/test_router_model_cost_isolation.py index c8adc0af431..a1040e972f2 100644 --- a/tests/test_litellm/test_router_model_cost_isolation.py +++ b/tests/test_litellm/test_router_model_cost_isolation.py @@ -564,6 +564,67 @@ def test_inherit_builtin_base_rates_for_off_peak_fills_missing_rates(): 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_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. From 60de5468d748c576b49455391dce897d1e5e800f Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Tue, 1 Sep 2026 13:05:25 -0700 Subject: [PATCH 14/14] fix(cost): inherit the backend's raw cost map entry for off-peak-only deployments Filtering copied fields by name dropped companion billing rules like web_search_billing_unit and the regional uplift multipliers, so grounding and uplifts billed differently through the deployment entry. Copy the backend's raw litellm.model_cost entry wholesale instead, which also removes the synthesized-zero special case since the raw entry only holds real values. --- litellm/router.py | 32 ++++++++++--------- .../test_router_model_cost_isolation.py | 24 ++++++++++++++ 2 files changed, 41 insertions(+), 15 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index dd1cb51d9af..c50fab5fe0f 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -8176,16 +8176,19 @@ class Router: 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. Every price-bearing backend field is - copied, not just the flat token rates: threshold, tiered, service-tier, - cache, character, and per-second rates all carry over, so peak-hour - billing through the deployment entry matches the shared backend entry - exactly. Values are deep-copied to keep the builtin entry isolated, and - a flat token rate ``get_model_info`` synthesized as zero for a backend - without one is rejected, like ``_inherit_builtin_tiered_output_rate`` - does, so a tiered-only backend is never marked explicitly priced free. - User-specified rates always win; no-op when any base pricing field is - already set or the backend model has no canonical entry. + 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 @@ -8198,13 +8201,12 @@ class Router: 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 - for field, backend_value in backend_info.items(): - if "cost" not in field and field != "tiered_pricing": - continue + 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 - if field in ("input_cost_per_token", "output_cost_per_token") and not backend_value: - continue model_info[field] = copy.deepcopy(backend_value) @staticmethod diff --git a/tests/test_litellm/test_router_model_cost_isolation.py b/tests/test_litellm/test_router_model_cost_isolation.py index a1040e972f2..7b7a962bf00 100644 --- a/tests/test_litellm/test_router_model_cost_isolation.py +++ b/tests/test_litellm/test_router_model_cost_isolation.py @@ -595,6 +595,30 @@ def test_inherit_builtin_base_rates_for_off_peak_carries_threshold_rates(): ) +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