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

feat(cost): honor off_peak_pricing reasoning and cache-creation rates

Browse source 
The block accepts output_cost_per_reasoning_token and cache_creation_input_token_cost. The generic
cost path and the DashScope calculator swap them in while a window is open, and unset keys keep the
standard rate. One shared TokenRates value replaces the DashScope-local copy, and
apply_off_peak_pricing takes and returns it.
This commit is contained in:
mateo-berri 2026-09-03 13:45:06 -07:00
parent b9e030ddd6
commit e297968826
5 changed files with 470 additions and 68 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -415,40 +415,69 @@ def _is_off_peak(off_peak: Mapping[str, object], current_time: datetime | None =
return False
def _coerce_off_peak_rate(value: object, default: float) -> float:
@dataclass(frozen=True, slots=True)
class TokenRates:
"""The per-token rates one request bills at. reasoning_rate is None when reasoning bills at
output_rate: the model has no dedicated reasoning rate, or the caller resolves reasoning on
its own.
"""
input_rate: float
output_rate: float
cache_read_rate: float
cache_creation_rate: float
reasoning_rate: float | None
@property
def billed_reasoning_rate(self) -> float:
return self.output_rate if self.reasoning_rate is None else self.reasoning_rate
def _parse_off_peak_rate(value: object) -> float | None:
if isinstance(value, bool):
return default
return None
if isinstance(value, (int, float)):
return float(value)
if isinstance(value, str):
try:
return float(value)
except ValueError:
return default
return default
return None
return None
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]:
def _off_peak_rate(off_peak: Mapping[str, object], key: str, standard_rate: float) -> float:
parsed: Final = _parse_off_peak_rate(off_peak.get(key))
return standard_rate if parsed is None else parsed
def _open_off_peak_block(model_info: ModelInfo, current_time: datetime | None) -> Mapping[str, object] | None:
off_peak: Final = model_info.get("off_peak_pricing")
if not isinstance(off_peak, Mapping) or not _is_off_peak(off_peak, current_time):
return None
return off_peak
def apply_off_peak_pricing(model_info: ModelInfo, current_time: datetime | None, rates: TokenRates) -> TokenRates:
"""Swap in off-peak per-token rates when the current UTC time is inside one of the model's
off_peak_pricing rules, the every-day hours_utc windows or a day-of-week-qualified entry in
windows. An off-peak rate replaces the rate that would otherwise apply rather than
discounting it, so a model that also has tiered or above-threshold pricing bills the flat
off-peak rate for the whole request while the window is open. Any rate left unset in
off_peak_pricing falls back to the standard rate.
off_peak_pricing falls back to the standard rate, so a block without
output_cost_per_reasoning_token keeps the model's own reasoning rate, or its off-peak output
rate when reasoning has no dedicated rate at all.
"""
off_peak: Final = model_info.get("off_peak_pricing")
if not isinstance(off_peak, Mapping) or not _is_off_peak(off_peak, current_time):
return prompt_base_cost, completion_base_cost, cache_read_cost
return (
_coerce_off_peak_rate(off_peak.get("input_cost_per_token"), prompt_base_cost),
_coerce_off_peak_rate(off_peak.get("output_cost_per_token"), completion_base_cost),
_coerce_off_peak_rate(off_peak.get("cache_read_input_token_cost"), cache_read_cost),
off_peak: Final = _open_off_peak_block(model_info, current_time)
if off_peak is None:
return rates
off_peak_reasoning_rate: Final = _parse_off_peak_rate(off_peak.get("output_cost_per_reasoning_token"))
return TokenRates(
input_rate=_off_peak_rate(off_peak, "input_cost_per_token", rates.input_rate),
output_rate=_off_peak_rate(off_peak, "output_cost_per_token", rates.output_rate),
cache_read_rate=_off_peak_rate(off_peak, "cache_read_input_token_cost", rates.cache_read_rate),
cache_creation_rate=_off_peak_rate(off_peak, "cache_creation_input_token_cost", rates.cache_creation_rate),
reasoning_rate=rates.reasoning_rate if off_peak_reasoning_rate is None else off_peak_reasoning_rate,
)
@ -458,14 +487,28 @@ def _apply_off_peak_to_base_costs(
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.
produced them. The one-hour cache-creation rate passes through untouched, since
off_peak_pricing has no field for it, and reasoning is left to _resolve_billed_reasoning_rate.
"""
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
rates: Final = apply_off_peak_pricing(
model_info,
current_time,
TokenRates(
input_rate=prompt,
output_rate=completion,
cache_read_rate=cache_read,
cache_creation_rate=cache_creation,
reasoning_rate=None,
),
)
return (
rates.input_rate,
rates.output_rate,
rates.cache_creation_rate,
cache_creation_above_1hr,
rates.cache_read_rate,
)
return (off_peak_prompt, off_peak_completion, cache_creation, cache_creation_above_1hr, off_peak_cache_read)
def _get_token_base_cost(
@ -1029,6 +1072,29 @@ def _resolve_reasoning_token_cost(
return standard_reasoning_cost if standard_reasoning_cost is not None else completion_base_cost
def _resolve_billed_reasoning_rate(
model_info: ModelInfo,
usage: Usage,
service_tier: str | None,
completion_base_cost: float,
current_time: datetime | None,
) -> float:
off_peak: Final = _open_off_peak_block(model_info, current_time)
off_peak_reasoning_rate: Final = (
None if off_peak is None else _parse_off_peak_rate(off_peak.get("output_cost_per_reasoning_token"))
)
if off_peak_reasoning_rate is not None:
return off_peak_reasoning_rate
tiered_reasoning_rate: Final = _get_tiered_reasoning_rate(model_info=model_info, usage=usage)
if tiered_reasoning_rate is not None:
return tiered_reasoning_rate
return _resolve_reasoning_token_cost(
model_info=model_info,
service_tier=service_tier,
completion_base_cost=completion_base_cost,
)
def generic_cost_per_token(
model: str,
usage: Usage,
@ -1037,6 +1103,7 @@ def generic_cost_per_token(
data_residency: str | None = None,
model_info: ModelInfo | None = None,
vertex_location: str | None = None,
current_time: datetime | None = None,
) -> tuple[float, float]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@ -1051,6 +1118,7 @@ def generic_cost_per_token(
- vertex_location: optional Vertex AI location the request was served from
(e.g. "us-east5", "global"), used to apply the per-model
regional-endpoint uplift multiplier when non-global.
- current_time: the moment the request is billed at, for off_peak_pricing; defaults to now, UTC
Returns:
Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
@ -1117,6 +1185,7 @@ def generic_cost_per_token(
usage.prompt_tokens - cache_hit - audio_tokens - cache_creation - image_tokens - video_tokens, 0
)
billing_time: Final = current_time if current_time is not None else datetime.now(timezone.utc)
(
prompt_base_cost,
completion_base_cost,
@ -1127,6 +1196,7 @@ def generic_cost_per_token(
model_info=model_info,
usage=usage,
service_tier=service_tier,
current_time=billing_time,
threshold_is_inclusive=_uses_inclusive_token_thresholds(custom_llm_provider),
)
@ -1185,17 +1255,13 @@ def generic_cost_per_token(
## REASONING COST
if not is_text_tokens_total and reasoning_tokens and reasoning_tokens > 0:
tiered_reasoning_rate: Final = _get_tiered_reasoning_rate(model_info=model_info, usage=usage)
_output_cost_per_reasoning_token = (
tiered_reasoning_rate
if tiered_reasoning_rate is not None
else _resolve_reasoning_token_cost(
model_info=model_info,
service_tier=service_tier,
completion_base_cost=completion_base_cost,
)
completion_cost += float(reasoning_tokens) * _resolve_billed_reasoning_rate(
model_info=model_info,
usage=usage,
service_tier=service_tier,
completion_base_cost=completion_base_cost,
current_time=billing_time,
)
completion_cost += float(reasoning_tokens) * _output_cost_per_reasoning_token
## IMAGE COST
if not is_text_tokens_total and image_tokens and image_tokens > 0:
@ -1247,6 +1313,7 @@ def get_token_type_cost_breakdown(
service_tier: str | None = None,
data_residency: str | None = None,
vertex_location: str | None = None,
current_time: datetime | None = None,
) -> TokenTypeCostBreakdown:
"""
Provider-agnostic cost of reasoning and cache tokens, derived from the usage
@ -1265,6 +1332,7 @@ def get_token_type_cost_breakdown(
except Exception:
return TokenTypeCostBreakdown(0.0, 0.0, 0.0)
billing_time: Final = current_time if current_time is not None else datetime.now(timezone.utc)
(
_prompt_base_cost,
completion_base_cost,
@ -1275,6 +1343,7 @@ def get_token_type_cost_breakdown(
model_info=model_info,
usage=usage,
service_tier=service_tier,
current_time=billing_time,
threshold_is_inclusive=_uses_inclusive_token_thresholds(custom_llm_provider),
)
@ -1284,18 +1353,12 @@ def get_token_type_cost_breakdown(
if not reasoning_tokens:
reasoning_tokens = _coerce_token_count(getattr(usage, "reasoning_tokens", 0))
# Reasoning is billed at the selected tier's reasoning rate for tiered models,
# else at the service-tier-aware per-reasoning-token rate - this mirrors how the
# total completion cost is computed, so the breakdown can never diverge from it.
tiered_reasoning_rate: Final = _get_tiered_reasoning_rate(model_info=model_info, usage=usage)
reasoning_rate: Final = (
tiered_reasoning_rate
if tiered_reasoning_rate is not None
else _resolve_reasoning_token_cost(
model_info=model_info,
service_tier=service_tier,
completion_base_cost=completion_base_cost,
)
reasoning_rate: Final = _resolve_billed_reasoning_rate(
model_info=model_info,
usage=usage,
service_tier=service_tier,
completion_base_cost=completion_base_cost,
current_time=billing_time,
)
reasoning_cost = float(reasoning_tokens) * reasoning_rate

25
litellm/llms/dashscope/cost_calculator.py
View file

@ -7,12 +7,13 @@ cached, cache-creation, output, reasoning) is billed at that one tier's rate.
See https://help.aliyun.com/zh/model-studio/billing-for-model-studio
"""
from dataclasses import dataclass, replace
from dataclasses import dataclass
from datetime import datetime
from typing import Final
from litellm.litellm_core_utils.llm_cost_calc.tiered_pricing import select_tier_for_input, tier_rate
from litellm.litellm_core_utils.llm_cost_calc.utils import (
TokenRates,
apply_off_peak_pricing,
parse_completion_tokens_details,
parse_prompt_tokens_details,
@ -34,19 +35,6 @@ class TokenBreakdown:
return self.text_tokens + self.cached_tokens + self.cache_creation_tokens
@dataclass(frozen=True, slots=True)
class TokenRates:
input_rate: float
cache_read_rate: float
cache_creation_rate: float
output_rate: float
reasoning_rate: float | None
@property
def billed_reasoning_rate(self) -> float:
return self.output_rate if self.reasoning_rate is None else self.reasoning_rate
def _extract_token_breakdown(usage: Usage) -> TokenBreakdown:
prompt_details: Final = parse_prompt_tokens_details(usage)
cached_tokens: Final = prompt_details["cache_hit_tokens"]
@ -105,13 +93,6 @@ def _tier_rates(model_info: ModelInfo, tier: dict) -> TokenRates:
)
def _off_peak_rates(model_info: ModelInfo, current_time: datetime | None, rates: TokenRates) -> TokenRates:
input_rate, output_rate, cache_read_rate = apply_off_peak_pricing(
model_info, current_time, rates.input_rate, rates.output_rate, rates.cache_read_rate
)
return replace(rates, input_rate=input_rate, output_rate=output_rate, cache_read_rate=cache_read_rate)
def _bill(breakdown: TokenBreakdown, rates: TokenRates) -> tuple[float, float]:
prompt_cost: Final = (
(breakdown.text_tokens * rates.input_rate)
@ -155,6 +136,6 @@ def cost_per_token(
else None
)
standard_rates: Final = _flat_rates(model_info) if tier is None else _tier_rates(model_info, tier)
rates: Final = _off_peak_rates(model_info, current_time, standard_rates)
rates: Final = apply_off_peak_pricing(model_info, current_time, standard_rates)
return _bill(breakdown, rates)

2
litellm/types/utils.py
View file

@ -222,7 +222,9 @@ class OffPeakPricing(TypedDict, total=False):
weekday_timezone: ReadOnly[str]
input_cost_per_token: ReadOnly[float]
output_cost_per_token: ReadOnly[float]
output_cost_per_reasoning_token: ReadOnly[float]
cache_read_input_token_cost: ReadOnly[float]
cache_creation_input_token_cost: ReadOnly[float]
class ModelInfoBase(ProviderSpecificModelInfo, total=False):

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

@ -29,11 +29,13 @@ from litellm.types.utils import (
from litellm.litellm_core_utils.llm_cost_calc.utils import (
CostCalculatorUtils,
PromptTokensDetailsResult,
TokenRates,
TokenTypeCostBreakdown,
_calculate_input_cost,
_get_token_base_cost,
_is_off_peak,
_is_within_off_peak_window,
apply_off_peak_pricing,
calculate_cache_writing_cost,
generic_cost_per_token,
get_token_type_cost_breakdown,
@ -782,6 +784,271 @@ def test_get_token_base_cost_off_peak_wins_over_tiered_pricing():
assert outside[:2] == (3e-6, 6e-6)
def _register_off_peak_reasoning_model(
model_name: str, off_peak_pricing: dict, reasoning_rate: float | None = 4e-6, **service_tier_rates: float
) -> None:
reasoning_entry = {} if reasoning_rate is None else {"output_cost_per_reasoning_token": reasoning_rate}
litellm.register_model(
{
model_name: {
"litellm_provider": "openai",
"mode": "chat",
"input_cost_per_token": 1e-6,
"output_cost_per_token": 2e-6,
"cache_read_input_token_cost": 1e-7,
"cache_creation_input_token_cost": 1.25e-6,
"off_peak_pricing": off_peak_pricing,
**reasoning_entry,
**service_tier_rates,
}
}
)
def _off_peak_reasoning_usage() -> Usage:
return Usage(
prompt_tokens=100,
completion_tokens=80,
total_tokens=180,
completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=30, text_tokens=50),
)
def test_generic_cost_per_token_off_peak_reasoning_rate():
"""Regression (LIT-6887): the block's output_cost_per_reasoning_token used to be ignored, so
reasoning tokens billed at the model's standard reasoning rate all through the window."""
from datetime import datetime, timezone
model_name = "litellm-test-off-peak-reasoning"
_register_off_peak_reasoning_model(
model_name,
{"hours_utc": "16:30-00:30", "output_cost_per_token": 1e-6, "output_cost_per_reasoning_token": 5e-7},
)
_, inside = generic_cost_per_token(
model=model_name,
usage=_off_peak_reasoning_usage(),
custom_llm_provider="openai",
current_time=datetime(2026, 1, 1, 18, 0, tzinfo=timezone.utc),
)
assert inside == pytest.approx(50 * 1e-6 + 30 * 5e-7)
_, outside = generic_cost_per_token(
model=model_name,
usage=_off_peak_reasoning_usage(),
custom_llm_provider="openai",
current_time=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc),
)
assert outside == pytest.approx(50 * 2e-6 + 30 * 4e-6)
def test_generic_cost_per_token_off_peak_block_without_reasoning_rate():
"""A block that leaves output_cost_per_reasoning_token unset keeps the model's own reasoning
rate, and a model with no reasoning rate at all follows the off-peak output rate."""
from datetime import datetime, timezone
inside_window = datetime(2026, 1, 1, 18, 0, tzinfo=timezone.utc)
block = {"hours_utc": "16:30-00:30", "output_cost_per_token": 1e-6}
_register_off_peak_reasoning_model("litellm-test-off-peak-model-reasoning-rate", block)
_, with_model_rate = generic_cost_per_token(
model="litellm-test-off-peak-model-reasoning-rate",
usage=_off_peak_reasoning_usage(),
custom_llm_provider="openai",
current_time=inside_window,
)
assert with_model_rate == pytest.approx(50 * 1e-6 + 30 * 4e-6)
_register_off_peak_reasoning_model("litellm-test-off-peak-no-reasoning-rate", block, reasoning_rate=None)
_, without_model_rate = generic_cost_per_token(
model="litellm-test-off-peak-no-reasoning-rate",
usage=_off_peak_reasoning_usage(),
custom_llm_provider="openai",
current_time=inside_window,
)
assert without_model_rate == pytest.approx(80 * 1e-6)
def test_generic_cost_per_token_off_peak_reasoning_rate_wins_over_the_tier():
"""Tiered models resolve reasoning on their own path, so the block has to win there too."""
from datetime import datetime, timezone
model_name = "litellm-test-off-peak-tiered-reasoning"
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,
"output_cost_per_reasoning_token": 8e-6,
},
],
"off_peak_pricing": {
"hours_utc": "16:30-00:30",
"output_cost_per_token": 1e-6,
"output_cost_per_reasoning_token": 5e-7,
},
}
}
)
_, inside = generic_cost_per_token(
model=model_name,
usage=_off_peak_reasoning_usage(),
custom_llm_provider="openai",
current_time=datetime(2026, 1, 1, 18, 0, tzinfo=timezone.utc),
)
assert inside == pytest.approx(50 * 1e-6 + 30 * 5e-7)
_, outside = generic_cost_per_token(
model=model_name,
usage=_off_peak_reasoning_usage(),
custom_llm_provider="openai",
current_time=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc),
)
assert outside == pytest.approx(50 * 6e-6 + 30 * 8e-6)
def test_generic_cost_per_token_off_peak_reasoning_rate_wins_over_the_service_tier():
"""A priority request bills its service-tier reasoning rate outside the window and the block's
rate inside it."""
from datetime import datetime, timezone
model_name = "litellm-test-off-peak-reasoning-service-tier"
_register_off_peak_reasoning_model(
model_name,
{"hours_utc": "16:30-00:30", "output_cost_per_token": 1e-6, "output_cost_per_reasoning_token": 5e-7},
output_cost_per_token_priority=3e-6,
output_cost_per_reasoning_token_priority=6e-6,
)
_, inside = generic_cost_per_token(
model=model_name,
usage=_off_peak_reasoning_usage(),
custom_llm_provider="openai",
service_tier="priority",
current_time=datetime(2026, 1, 1, 18, 0, tzinfo=timezone.utc),
)
assert inside == pytest.approx(50 * 1e-6 + 30 * 5e-7)
_, outside = generic_cost_per_token(
model=model_name,
usage=_off_peak_reasoning_usage(),
custom_llm_provider="openai",
service_tier="priority",
current_time=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc),
)
assert outside == pytest.approx(50 * 3e-6 + 30 * 6e-6)
def test_apply_off_peak_pricing_treats_bool_as_unset_and_parses_strings():
"""A YAML true never turns into a rate of 1.0, and a quoted number still counts."""
from datetime import datetime, timezone
model_name = "litellm-test-off-peak-odd-values"
_register_off_peak_reasoning_model(
model_name,
{
"hours_utc": "16:30-00:30",
"cache_creation_input_token_cost": True,
"output_cost_per_reasoning_token": "5e-7",
},
)
standard = TokenRates(
input_rate=1e-6, output_rate=2e-6, cache_read_rate=1e-7, cache_creation_rate=1.25e-6, reasoning_rate=4e-6
)
rates = apply_off_peak_pricing(
litellm.get_model_info(model_name, custom_llm_provider="openai"),
datetime(2026, 1, 1, 18, 0, tzinfo=timezone.utc),
standard,
)
assert rates.cache_creation_rate == 1.25e-6
assert rates.reasoning_rate == 5e-7
def test_get_token_base_cost_off_peak_cache_creation_rate():
"""Regression (LIT-6887): the block's cache_creation_input_token_cost used to be ignored. It
replaces the five-minute cache-creation rate inside the window; the one-hour rate, and a
block without the key, keep the standard 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,
"cache_creation_input_token_cost": 1.25e-6,
"cache_creation_input_token_cost_above_1hr": 2e-6,
"off_peak_pricing": {"hours_utc": "16:30-00:30", "cache_creation_input_token_cost": 5e-7},
},
)
usage = Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150)
inside_window = datetime(2026, 1, 1, 18, 0, tzinfo=timezone.utc)
inside = _get_token_base_cost(model_info, usage, current_time=inside_window)
assert inside[2] == 5e-7
assert inside[3] == 2e-6
outside = _get_token_base_cost(model_info, usage, current_time=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc))
assert outside[2] == 1.25e-6
without_key = cast(
ModelInfo,
{**model_info, "off_peak_pricing": {"hours_utc": "16:30-00:30", "input_cost_per_token": 5e-7}},
)
assert _get_token_base_cost(without_key, usage, current_time=inside_window)[2] == 1.25e-6
def test_get_token_type_cost_breakdown_reflects_off_peak_reasoning_and_cache_creation_rates():
"""The per-token-type breakdown feeds the spend logs, so it has to bill the new keys the same
way the total does."""
from datetime import datetime, timezone
model_name = "litellm-test-off-peak-breakdown"
_register_off_peak_reasoning_model(
model_name,
{
"hours_utc": "16:30-00:30",
"output_cost_per_reasoning_token": 5e-7,
"cache_creation_input_token_cost": 5e-7,
},
)
usage = Usage(
prompt_tokens=1000,
completion_tokens=80,
total_tokens=1080,
completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=30, text_tokens=50),
prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=100, cache_creation_tokens=400, text_tokens=500),
)
inside = get_token_type_cost_breakdown(
model=model_name,
custom_llm_provider="openai",
usage=usage,
current_time=datetime(2026, 1, 1, 18, 0, tzinfo=timezone.utc),
)
assert inside.reasoning_cost == pytest.approx(30 * 5e-7)
assert inside.cache_creation_cost == pytest.approx(400 * 5e-7)
assert inside.cache_read_cost == pytest.approx(100 * 1e-7)
outside = get_token_type_cost_breakdown(
model=model_name,
custom_llm_provider="openai",
usage=usage,
current_time=datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc),
)
assert outside.reasoning_cost == pytest.approx(30 * 4e-6)
assert outside.cache_creation_cost == pytest.approx(400 * 1.25e-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"

89
tests/test_litellm/llms/dashscope/test_dashscope_cost_calculator.py
View file

@ -649,6 +649,95 @@ class TestDashscopeCostCalculator:
assert math.isclose(completion_cost, 200 * 2.4e-06, rel_tol=1e-10)
def test_dashscope_off_peak_reasoning_rate_replaces_the_dedicated_reasoning_rate(self):
"""Regression (LIT-6887): a block carrying output_cost_per_reasoning_token bills reasoning
tokens at it inside the window, over the model's own reasoning rate, which returns outside."""
self._register_off_peak_flat_model(
"dashscope/qwen-reasoning-rate-off-peak-test",
{
"hours_utc": self.OFF_PEAK_WINDOW,
"output_cost_per_token": 2.4e-06,
"output_cost_per_reasoning_token": 4.5e-06,
},
)
litellm.model_cost["dashscope/qwen-reasoning-rate-off-peak-test"]["output_cost_per_reasoning_token"] = 9e-06
usage = Usage(
prompt_tokens=100,
completion_tokens=200,
completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=50),
)
_, completion_cost = dashscope_cost_per_token(
model="qwen-reasoning-rate-off-peak-test", usage=usage, current_time=self.INSIDE_WINDOW
)
assert math.isclose(completion_cost, (150 * 2.4e-06) + (50 * 4.5e-06), rel_tol=1e-10)
_, peak_completion_cost = dashscope_cost_per_token(
model="qwen-reasoning-rate-off-peak-test", usage=usage, current_time=self.OUTSIDE_WINDOW
)
assert math.isclose(peak_completion_cost, (150 * 4.8e-06) + (50 * 9e-06), rel_tol=1e-10)
def test_dashscope_off_peak_cache_creation_rate_replaces_the_standard_rate(self):
"""Regression (LIT-6887): a block carrying cache_creation_input_token_cost bills cache-creation
tokens at it inside the window, while the cache-read rate it leaves unset stays standard."""
self._register_off_peak_flat_model(
"dashscope/qwen-cache-creation-off-peak-test",
{"hours_utc": self.OFF_PEAK_WINDOW, "cache_creation_input_token_cost": 1.5e-06},
)
usage = Usage(
prompt_tokens=1000,
completion_tokens=10,
prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=300, cache_creation_tokens=100),
)
prompt_cost, _ = dashscope_cost_per_token(
model="qwen-cache-creation-off-peak-test", usage=usage, current_time=self.INSIDE_WINDOW
)
assert math.isclose(prompt_cost, (600 * 2.4e-06) + (300 * 2e-07) + (100 * 1.5e-06), rel_tol=1e-10)
peak_prompt_cost, _ = dashscope_cost_per_token(
model="qwen-cache-creation-off-peak-test", usage=usage, current_time=self.OUTSIDE_WINDOW
)
assert math.isclose(peak_prompt_cost, (600 * 2.4e-06) + (300 * 2e-07) + (100 * 3e-06), rel_tol=1e-10)
def test_dashscope_off_peak_reasoning_and_cache_creation_rates_override_the_selected_tier(self):
"""The new keys override the selected tier the way the input and output rates already do."""
self._register_tiered_model(
"dashscope/qwen-tiered-reasoning-off-peak-test",
[
{
"range": [0, 1000],
"input_cost_per_token": 4e-07,
"cache_creation_input_token_cost": 3e-07,
"output_cost_per_token": 1.6e-06,
"output_cost_per_reasoning_token": 3.2e-06,
},
],
)
litellm.model_cost["dashscope/qwen-tiered-reasoning-off-peak-test"]["off_peak_pricing"] = {
"hours_utc": self.OFF_PEAK_WINDOW,
"cache_creation_input_token_cost": 1e-07,
"output_cost_per_reasoning_token": 8e-07,
}
usage = Usage(
prompt_tokens=500,
completion_tokens=100,
prompt_tokens_details=PromptTokensDetailsWrapper(cache_creation_tokens=200),
completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=40),
)
prompt_cost, completion_cost = dashscope_cost_per_token(
model="qwen-tiered-reasoning-off-peak-test", usage=usage, current_time=self.INSIDE_WINDOW
)
assert math.isclose(prompt_cost, (300 * 4e-07) + (200 * 1e-07), rel_tol=1e-10)
assert math.isclose(completion_cost, (60 * 1.6e-06) + (40 * 8e-07), rel_tol=1e-10)
peak_prompt_cost, peak_completion_cost = dashscope_cost_per_token(
model="qwen-tiered-reasoning-off-peak-test", usage=usage, current_time=self.OUTSIDE_WINDOW
)
assert math.isclose(peak_prompt_cost, (300 * 4e-07) + (200 * 3e-07), rel_tol=1e-10)
assert math.isclose(peak_completion_cost, (60 * 1.6e-06) + (40 * 3.2e-06), rel_tol=1e-10)
def test_dashscope_off_peak_defaults_to_the_current_time(self):
"""The proxy's cost dispatch passes no clock, so an all-day window has to apply on the
default current time."""