scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-06 08:16:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 79

fix(cost): apply off_peak_pricing in the dashscope cost calculator

Browse source
This commit is contained in:
mateo-berri 2026-09-03 10:38:25 -07:00
parent 2c30fe16b0
commit b9e030ddd6
3 changed files with 209 additions and 53 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -428,7 +428,7 @@ def _coerce_off_peak_rate(value: object, default: float) -> float:
return default
def _apply_off_peak_pricing(
def apply_off_peak_pricing(
model_info: ModelInfo,
current_time: datetime | None,
prompt_base_cost: float,
@ -462,7 +462,7 @@ def _apply_off_peak_to_base_costs(
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(
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)

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

@ -7,11 +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
from dataclasses import dataclass, replace
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 (
apply_off_peak_pricing,
parse_completion_tokens_details,
parse_prompt_tokens_details,
)
@ -32,6 +34,19 @@ 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"]
@ -57,69 +72,75 @@ def _flat_rate(model_info: ModelInfo, cost_key: str, fallback_cost_key: str) ->
return float(value)
def _calculate_prompt_cost(
breakdown: TokenBreakdown,
model_info: ModelInfo,
tier: dict | None,
) -> float:
if tier is not None:
return (
(breakdown.text_tokens * tier_rate(tier, "input_cost_per_token"))
+ (breakdown.cached_tokens * tier_rate(tier, "cache_read_input_token_cost", "input_cost_per_token"))
+ (
breakdown.cache_creation_tokens
* tier_rate(tier, "cache_creation_input_token_cost", "input_cost_per_token")
)
)
input_cost: Final = float(model_info.get("input_cost_per_token") or 0.0)
cache_read_cost: Final = _flat_rate(model_info, "cache_read_input_token_cost", "input_cost_per_token")
cache_creation_cost: Final = _flat_rate(model_info, "cache_creation_input_token_cost", "input_cost_per_token")
return (
(breakdown.text_tokens * input_cost)
+ (breakdown.cached_tokens * cache_read_cost)
+ (breakdown.cache_creation_tokens * cache_creation_cost)
def _flat_rates(model_info: ModelInfo) -> TokenRates:
reasoning_rate: Final = model_info.get("output_cost_per_reasoning_token")
return TokenRates(
input_rate=float(model_info.get("input_cost_per_token") or 0.0),
cache_read_rate=_flat_rate(model_info, "cache_read_input_token_cost", "input_cost_per_token"),
cache_creation_rate=_flat_rate(model_info, "cache_creation_input_token_cost", "input_cost_per_token"),
output_rate=float(model_info.get("output_cost_per_token") or 0.0),
reasoning_rate=None if reasoning_rate is None else float(reasoning_rate),
)
def _calculate_completion_cost(
breakdown: TokenBreakdown,
model_info: ModelInfo,
tier: dict | None,
) -> float:
def _tier_rates(model_info: ModelInfo, tier: dict) -> TokenRates:
# A tier that declares output rates keeps the request on them, all-or-nothing. A tier table
# spelling out only input rates would serve every completion for free, so there the model's
# own output rates stand in
tier_declares_output: Final = tier is not None and "output_cost_per_token" in tier
output_cost: Final = (
tier_rate(tier, "output_cost_per_token")
if tier_declares_output
else float(model_info.get("output_cost_per_token") or 0.0)
)
tier_declares_reasoning: Final = tier is not None and "output_cost_per_reasoning_token" in tier
model_reasoning_rate: Final = None if tier_declares_output else model_info.get("output_cost_per_reasoning_token")
reasoning_cost: Final = (
tier_rate(tier, "output_cost_per_reasoning_token", "output_cost_per_token")
if tier_declares_reasoning
else float(model_reasoning_rate)
if model_reasoning_rate is not None
else output_cost
flat_rates: Final = _flat_rates(model_info)
tier_declares_output: Final = "output_cost_per_token" in tier
tier_declares_reasoning: Final = "output_cost_per_reasoning_token" in tier
return TokenRates(
input_rate=tier_rate(tier, "input_cost_per_token"),
cache_read_rate=tier_rate(tier, "cache_read_input_token_cost", "input_cost_per_token"),
cache_creation_rate=tier_rate(tier, "cache_creation_input_token_cost", "input_cost_per_token"),
output_rate=tier_rate(tier, "output_cost_per_token") if tier_declares_output else flat_rates.output_rate,
reasoning_rate=(
tier_rate(tier, "output_cost_per_reasoning_token")
if tier_declares_reasoning
else None
if tier_declares_output
else flat_rates.reasoning_rate
),
)
return (breakdown.completion_tokens * output_cost) + (breakdown.reasoning_tokens * reasoning_cost)
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 cost_per_token(model: str, usage: Usage, custom_llm_provider: str = "dashscope") -> tuple[float, float]:
def _bill(breakdown: TokenBreakdown, rates: TokenRates) -> tuple[float, float]:
prompt_cost: Final = (
(breakdown.text_tokens * rates.input_rate)
+ (breakdown.cached_tokens * rates.cache_read_rate)
+ (breakdown.cache_creation_tokens * rates.cache_creation_rate)
)
completion_cost: Final = (breakdown.completion_tokens * rates.output_rate) + (
breakdown.reasoning_tokens * rates.billed_reasoning_rate
)
return prompt_cost, completion_cost
def cost_per_token(
model: str,
usage: Usage,
custom_llm_provider: str = "dashscope",
current_time: datetime | None = None,
) -> tuple[float, float]:
"""
Calculate cost per token for Dashscope models.
Supports both tiered and flat pricing with cached and reasoning tokens.
Supports both tiered and flat pricing with cached and reasoning tokens, and swaps in the
model's off_peak_pricing rates while one of its windows is open.
Args:
model: Model name without provider prefix
usage: LiteLLM Usage block
custom_llm_provider: The provider id the request resolved to; dashscope or one of its brand aliases
current_time: The moment the request is billed at; defaults to now, UTC
Returns:
Tuple[float, float] - (prompt_cost_in_usd, completion_cost_in_usd)
@ -133,8 +154,7 @@ def cost_per_token(model: str, usage: Usage, custom_llm_provider: str = "dashsco
if tiered_pricing
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)
prompt_cost: Final = _calculate_prompt_cost(breakdown=breakdown, model_info=model_info, tier=tier)
completion_cost: Final = _calculate_completion_cost(breakdown=breakdown, model_info=model_info, tier=tier)
return prompt_cost, completion_cost
return _bill(breakdown, rates)

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

@ -10,11 +10,11 @@ Tests the cost calculation for Dashscope models including:
import math
import os
from datetime import datetime, timezone
import pytest
# Add the project root to Python path
import litellm
from litellm.llms.dashscope.cost_calculator import (
cost_per_token as dashscope_cost_per_token,
@ -526,3 +526,139 @@ class TestDashscopeCostCalculator:
assert prompt_cost == 0.0
assert math.isclose(completion_cost, 500 * 1.6e-06, rel_tol=1e-10)
OFF_PEAK_WINDOW = "14:00-00:00"
INSIDE_WINDOW = datetime(2026, 9, 3, 17, 25, tzinfo=timezone.utc)
OUTSIDE_WINDOW = datetime(2026, 9, 3, 9, 0, tzinfo=timezone.utc)
def _register_off_peak_flat_model(self, model_key: str, off_peak_pricing: dict) -> None:
litellm.model_cost[model_key] = {
"litellm_provider": "dashscope",
"mode": "chat",
"input_cost_per_token": 2.4e-06,
"output_cost_per_token": 4.8e-06,
"cache_read_input_token_cost": 2e-07,
"cache_creation_input_token_cost": 3e-06,
"off_peak_pricing": off_peak_pricing,
}
def test_dashscope_off_peak_window_swaps_in_the_off_peak_rates(self):
"""
Regression (LIT-6782): a deployment configured with off_peak_pricing kept billing the
standard dashscope rates inside its window, while the same block on a deepseek
deployment billed the off-peak rates.
"""
self._register_off_peak_flat_model(
"dashscope/deepseek-off-peak-test",
{
"hours_utc": self.OFF_PEAK_WINDOW,
"input_cost_per_token": 1.2e-06,
"output_cost_per_token": 2.4e-06,
"cache_read_input_token_cost": 1e-07,
},
)
usage = Usage(
prompt_tokens=1000,
completion_tokens=200,
prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=300, cache_creation_tokens=100),
)
prompt_cost, completion_cost = dashscope_cost_per_token(
model="deepseek-off-peak-test", usage=usage, current_time=self.INSIDE_WINDOW
)
assert math.isclose(prompt_cost, (600 * 1.2e-06) + (300 * 1e-07) + (100 * 3e-06), rel_tol=1e-10)
assert math.isclose(completion_cost, 200 * 2.4e-06, rel_tol=1e-10)
peak_prompt_cost, peak_completion_cost = dashscope_cost_per_token(
model="deepseek-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)
assert math.isclose(peak_completion_cost, 200 * 4.8e-06, rel_tol=1e-10)
def test_dashscope_off_peak_window_overrides_the_selected_tier(self):
"""An open off-peak window bills the whole request at the flat off-peak rates, whichever tier
the input volume selected."""
self._register_tiered_model(
"dashscope/qwen-tiered-off-peak-test",
[
{"range": [0, 1000], "input_cost_per_token": 4e-07, "output_cost_per_token": 1.6e-06},
{"range": [1000, 2000], "input_cost_per_token": 8e-07, "output_cost_per_token": 3.2e-06},
],
)
litellm.model_cost["dashscope/qwen-tiered-off-peak-test"]["off_peak_pricing"] = {
"hours_utc": self.OFF_PEAK_WINDOW,
"input_cost_per_token": 1e-07,
"output_cost_per_token": 4e-07,
}
usage = Usage(prompt_tokens=1500, completion_tokens=300)
prompt_cost, completion_cost = dashscope_cost_per_token(
model="qwen-tiered-off-peak-test", usage=usage, current_time=self.INSIDE_WINDOW
)
assert math.isclose(prompt_cost, 1500 * 1e-07, rel_tol=1e-10)
assert math.isclose(completion_cost, 300 * 4e-07, rel_tol=1e-10)
peak_prompt_cost, peak_completion_cost = dashscope_cost_per_token(
model="qwen-tiered-off-peak-test", usage=usage, current_time=self.OUTSIDE_WINDOW
)
assert math.isclose(peak_prompt_cost, 1500 * 8e-07, rel_tol=1e-10)
assert math.isclose(peak_completion_cost, 300 * 3.2e-06, rel_tol=1e-10)
def test_dashscope_off_peak_rates_left_unset_keep_the_standard_rates(self):
"""A block that only overrides the input rate leaves output and cache reads on the standard
rates, and an explicit reasoning rate is never swapped out."""
self._register_off_peak_flat_model(
"dashscope/qwen-partial-off-peak-test",
{"hours_utc": self.OFF_PEAK_WINDOW, "input_cost_per_token": 1.2e-06},
)
litellm.model_cost["dashscope/qwen-partial-off-peak-test"]["output_cost_per_reasoning_token"] = 9e-06
usage = Usage(
prompt_tokens=1000,
completion_tokens=200,
prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=300),
completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=50),
)
prompt_cost, completion_cost = dashscope_cost_per_token(
model="qwen-partial-off-peak-test", usage=usage, current_time=self.INSIDE_WINDOW
)
assert math.isclose(prompt_cost, (700 * 1.2e-06) + (300 * 2e-07), rel_tol=1e-10)
assert math.isclose(completion_cost, (150 * 4.8e-06) + (50 * 9e-06), rel_tol=1e-10)
def test_dashscope_off_peak_output_rate_covers_reasoning_without_a_dedicated_rate(self):
"""Reasoning tokens on a model with no dedicated reasoning rate follow the off-peak output
rate, the same way they follow the standard output rate outside the window."""
self._register_off_peak_flat_model(
"dashscope/qwen-reasoning-off-peak-test",
{"hours_utc": self.OFF_PEAK_WINDOW, "output_cost_per_token": 2.4e-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-off-peak-test", usage=usage, current_time=self.INSIDE_WINDOW
)
assert math.isclose(completion_cost, 200 * 2.4e-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."""
self._register_off_peak_flat_model(
"dashscope/qwen-all-day-off-peak-test",
{"hours_utc": "00:00-00:00", "input_cost_per_token": 1.2e-06, "output_cost_per_token": 2.4e-06},
)
usage = Usage(prompt_tokens=1000, completion_tokens=200)
prompt_cost, completion_cost = dashscope_cost_per_token(model="qwen-all-day-off-peak-test", usage=usage)
assert math.isclose(prompt_cost, 1000 * 1.2e-06, rel_tol=1e-10)
assert math.isclose(completion_cost, 200 * 2.4e-06, rel_tol=1e-10)