fix(cost): honor off_peak_pricing in the fireworks_ai and perplexity cost calculators

litellm/litellm_core_utils/llm_cost_calc/utils.py

@@ -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)

litellm/llms/fireworks_ai/cost_calculator.py

@@ -2,6 +2,7 @@
 For calculating cost of fireworks ai serverless inference models.
 """
 
+from datetime import datetime
 from typing import Final
 
 from litellm.constants import (
@@ -10,7 +11,8 @@ from litellm.constants import (
     FIREWORKS_AI_56_B_MOE,
     FIREWORKS_AI_176_B_MOE,
 )
-from litellm.types.utils import Usage
+from litellm.litellm_core_utils.llm_cost_calc.utils import apply_off_peak_pricing
+from litellm.types.utils import ModelInfo, Usage
 from litellm.utils import get_model_info
 
 
@@ -54,44 +56,46 @@ def get_base_model_for_pricing(model_name: str) -> str:
     return "fireworks-ai-default"
 
 
-def cost_per_token(model: str, usage: Usage) -> tuple[float, float]:
+def _resolve_model_info(model: str) -> ModelInfo:
+    try:
+        return get_model_info(model=model, custom_llm_provider="fireworks_ai")
+    except Exception:
+        base_model: Final = get_base_model_for_pricing(model_name=model)
+        return get_model_info(model=base_model, custom_llm_provider="fireworks_ai")
+
+
+def cost_per_token(model: str, usage: Usage, current_time: datetime | None = None) -> tuple[float, float]:
     """
-    Calculates the cost per token for a given model, prompt tokens, and completion tokens.
+    Calculates the cost per token for a given model, prompt tokens, and completion tokens,
+    swapping in the model's off_peak_pricing rates while one of its windows is open.
 
     Input:
         - model: str, the model name without provider prefix
         - usage: LiteLLM Usage block, containing anthropic caching information
+        - 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
     """
-    ## check if model mapped, else use default pricing
-    try:
-        model_info = get_model_info(model=model, custom_llm_provider="fireworks_ai")
-    except Exception:
-        base_model: Final = get_base_model_for_pricing(model_name=model)
+    model_info: Final = _resolve_model_info(model)
+    standard_input_rate: Final[float] = model_info["input_cost_per_token"] or 0.0
+    standard_cache_read_rate: Final = model_info.get("cache_read_input_token_cost")
+    input_rate, output_rate, cache_read_rate = apply_off_peak_pricing(
+        model_info,
+        current_time,
+        standard_input_rate,
+        model_info["output_cost_per_token"] or 0.0,
+        standard_cache_read_rate if standard_cache_read_rate is not None else standard_input_rate,
+    )
 
-        ## GET MODEL INFO
-        model_info = get_model_info(model=base_model, custom_llm_provider="fireworks_ai")
-
-    ## CALCULATE INPUT COST
     prompt_tokens_details: Final = usage.prompt_tokens_details
     cached_tokens: Final[int] = (
         prompt_tokens_details.cached_tokens
         if prompt_tokens_details is not None and prompt_tokens_details.cached_tokens is not None
         else 0
     )
-    input_cost_per_token: Final[float] = model_info["input_cost_per_token"] or 0.0
-    cache_read_input_token_cost: Final = model_info.get("cache_read_input_token_cost")
-    cache_read_cost_per_token: Final[float] = (
-        cache_read_input_token_cost if cache_read_input_token_cost is not None else input_cost_per_token
-    )
     non_cached_prompt_tokens: Final[int] = max(usage.prompt_tokens - cached_tokens, 0)
-
-    prompt_cost: float = non_cached_prompt_tokens * input_cost_per_token + cached_tokens * cache_read_cost_per_token
-
-    ## CALCULATE OUTPUT COST
-    output_cost_per_token: Final[float] = model_info["output_cost_per_token"] or 0.0
-    completion_cost: Final[float] = usage.completion_tokens * output_cost_per_token
+    prompt_cost: Final[float] = non_cached_prompt_tokens * input_rate + cached_tokens * cache_read_rate
+    completion_cost: Final[float] = usage.completion_tokens * output_rate
 
     return prompt_cost, completion_cost

litellm/llms/perplexity/cost_calculator.py

@@ -3,19 +3,23 @@ Helper util for handling perplexity-specific cost calculation
 - e.g.: citation tokens, search queries
 """
 
+from datetime import datetime
 from typing import Final
 
+from litellm.litellm_core_utils.llm_cost_calc.utils import apply_off_peak_pricing
 from litellm.types.utils import Usage
 from litellm.utils import get_model_info
 
 
-def cost_per_token(model: str, usage: Usage) -> tuple[float, float]:
+def cost_per_token(model: str, usage: Usage, current_time: datetime | None = None) -> tuple[float, float]:
     """
     Calculates the cost per token for a given model, prompt tokens, and completion tokens.
+    The manual fallback swaps in the model's off_peak_pricing rates while one of its windows is open.
 
     Input:
         - model: str, the model name without provider prefix
         - usage: LiteLLM Usage block, containing perplexity-specific usage information
+        - 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
@@ -48,8 +52,15 @@ def cost_per_token(model: str, usage: Usage) -> tuple[float, float]:
         except (ValueError, TypeError):
             return default
 
+    input_cost_per_token, output_cost_per_token, _ = apply_off_peak_pricing(
+        model_info,
+        current_time,
+        _safe_float_cast(model_info.get("input_cost_per_token")),
+        _safe_float_cast(model_info.get("output_cost_per_token")),
+        0.0,
+    )
+
     ## CALCULATE INPUT COST
-    input_cost_per_token: Final = _safe_float_cast(model_info.get("input_cost_per_token"))
     prompt_cost: float = (usage.prompt_tokens or 0) * input_cost_per_token
 
     ## ADD CITATION TOKENS COST (if present)
@@ -60,8 +71,6 @@ def cost_per_token(model: str, usage: Usage) -> tuple[float, float]:
         prompt_cost += citation_tokens * citation_cost_per_token
 
     ## CALCULATE OUTPUT COST
-    output_cost_per_token: Final = _safe_float_cast(model_info.get("output_cost_per_token"))
-
     reasoning_tokens = getattr(usage, "reasoning_tokens", 0) or 0
     if reasoning_tokens == 0 and hasattr(usage, "completion_tokens_details") and usage.completion_tokens_details:
         reasoning_tokens = getattr(usage.completion_tokens_details, "reasoning_tokens", 0) or 0

tests/test_litellm/llms/fireworks_ai/test_fireworks_ai_cost_calculator.py

@@ -1,4 +1,7 @@
 
+import math
+from datetime import datetime, timezone
+
 import pytest
 
 
@@ -64,3 +67,75 @@ def test_no_cached_tokens_matches_full_input_rate():
 
     assert prompt_cost == pytest.approx(prompt_tokens * INPUT_COST)
     assert completion_cost == pytest.approx(completion_tokens * OUTPUT_COST)
+
+
+OFF_PEAK_MODEL = "accounts/fireworks/models/off-peak-test"
+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)
+STANDARD_INPUT_COST = 1.5e-07
+STANDARD_OUTPUT_COST = 6e-07
+STANDARD_CACHE_READ_COST = 1.5e-08
+
+
+def _register_off_peak_model(off_peak_pricing: dict) -> None:
+    litellm.model_cost[f"fireworks_ai/{OFF_PEAK_MODEL}"] = {
+        "litellm_provider": "fireworks_ai",
+        "mode": "chat",
+        "input_cost_per_token": STANDARD_INPUT_COST,
+        "output_cost_per_token": STANDARD_OUTPUT_COST,
+        "cache_read_input_token_cost": STANDARD_CACHE_READ_COST,
+        "off_peak_pricing": off_peak_pricing,
+    }
+
+
+def test_off_peak_window_swaps_in_the_off_peak_rates():
+    """
+    Regression (LIT-6874): a deployment configured with off_peak_pricing kept billing the
+    standard fireworks_ai rates inside its window, while the same block on a deepseek
+    deployment billed the off-peak rates.
+    """
+    _register_off_peak_model(
+        {
+            "hours_utc": OFF_PEAK_WINDOW,
+            "input_cost_per_token": 1e-08,
+            "output_cost_per_token": 2e-08,
+            "cache_read_input_token_cost": 1e-09,
+        }
+    )
+    usage = _usage(prompt_tokens=1000, cached_tokens=300, completion_tokens=200)
+
+    prompt_cost, completion_cost = cost_per_token(model=OFF_PEAK_MODEL, usage=usage, current_time=INSIDE_WINDOW)
+
+    assert math.isclose(prompt_cost, (700 * 1e-08) + (300 * 1e-09), rel_tol=1e-10)
+    assert math.isclose(completion_cost, 200 * 2e-08, rel_tol=1e-10)
+
+    peak_prompt_cost, peak_completion_cost = cost_per_token(
+        model=OFF_PEAK_MODEL, usage=usage, current_time=OUTSIDE_WINDOW
+    )
+
+    assert math.isclose(peak_prompt_cost, (700 * STANDARD_INPUT_COST) + (300 * STANDARD_CACHE_READ_COST), rel_tol=1e-10)
+    assert math.isclose(peak_completion_cost, 200 * STANDARD_OUTPUT_COST, rel_tol=1e-10)
+
+
+def test_off_peak_rates_left_unset_keep_the_standard_rates():
+    """A block that only overrides the input rate leaves output and cache reads on the standard rates."""
+    _register_off_peak_model({"hours_utc": OFF_PEAK_WINDOW, "input_cost_per_token": 1e-08})
+    usage = _usage(prompt_tokens=1000, cached_tokens=300, completion_tokens=200)
+
+    prompt_cost, completion_cost = cost_per_token(model=OFF_PEAK_MODEL, usage=usage, current_time=INSIDE_WINDOW)
+
+    assert math.isclose(prompt_cost, (700 * 1e-08) + (300 * STANDARD_CACHE_READ_COST), rel_tol=1e-10)
+    assert math.isclose(completion_cost, 200 * STANDARD_OUTPUT_COST, rel_tol=1e-10)
+
+
+def test_off_peak_defaults_to_the_current_time():
+    """The proxy's cost dispatch passes no clock, so an all-day window has to apply on the
+    default current time."""
+    _register_off_peak_model({"hours_utc": "00:00-00:00", "input_cost_per_token": 1e-08, "output_cost_per_token": 2e-08})
+    usage = _usage(prompt_tokens=1000, cached_tokens=0, completion_tokens=200)
+
+    prompt_cost, completion_cost = cost_per_token(model=OFF_PEAK_MODEL, usage=usage)
+
+    assert math.isclose(prompt_cost, 1000 * 1e-08, rel_tol=1e-10)
+    assert math.isclose(completion_cost, 200 * 2e-08, rel_tol=1e-10)

tests/test_litellm/llms/perplexity/test_perplexity_cost_calculator.py

@@ -8,6 +8,7 @@ search queries, and reasoning tokens.
 import json
 import math
 import os
+from datetime import datetime, timezone
 from unittest.mock import patch
 
 import pytest
@@ -523,3 +524,94 @@ class TestPerplexityCostCalculator:
         )
 
         assert math.isclose(total_cost, 1000 * 1.4e-06 + 500 * 4.4e-06, rel_tol=1e-9)
+
+    OFF_PEAK_MODEL = "sonar-off-peak-test"
+    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_model(self, off_peak_pricing: dict) -> None:
+        litellm.model_cost[f"perplexity/{self.OFF_PEAK_MODEL}"] = {
+            "litellm_provider": "perplexity",
+            "mode": "chat",
+            "input_cost_per_token": 1e-06,
+            "output_cost_per_token": 1e-06,
+            "output_cost_per_reasoning_token": 3e-06,
+            "citation_cost_per_token": 2e-06,
+            "search_context_cost_per_query": {"search_context_size_low": 0.005},
+            "off_peak_pricing": off_peak_pricing,
+        }
+
+    def test_off_peak_window_swaps_in_the_off_peak_rates(self):
+        """
+        Regression (LIT-6874): a deployment configured with off_peak_pricing kept billing the
+        standard perplexity rates inside its window, while the same block on a deepseek
+        deployment billed the off-peak rates.
+        """
+        self._register_off_peak_model(
+            {"hours_utc": self.OFF_PEAK_WINDOW, "input_cost_per_token": 1e-07, "output_cost_per_token": 2e-07}
+        )
+        usage = Usage(prompt_tokens=1000, completion_tokens=200, total_tokens=1200)
+
+        prompt_cost, completion_cost = perplexity_cost_per_token(
+            model=self.OFF_PEAK_MODEL, usage=usage, current_time=self.INSIDE_WINDOW
+        )
+
+        assert math.isclose(prompt_cost, 1000 * 1e-07, rel_tol=1e-10)
+        assert math.isclose(completion_cost, 200 * 2e-07, rel_tol=1e-10)
+
+        peak_prompt_cost, peak_completion_cost = perplexity_cost_per_token(
+            model=self.OFF_PEAK_MODEL, usage=usage, current_time=self.OUTSIDE_WINDOW
+        )
+
+        assert math.isclose(peak_prompt_cost, 1000 * 1e-06, rel_tol=1e-10)
+        assert math.isclose(peak_completion_cost, 200 * 1e-06, rel_tol=1e-10)
+
+    def test_off_peak_rates_leave_citation_search_and_reasoning_fees_alone(self):
+        """Inside the window only the plain input and output rates change: citation tokens, the
+        per-request search fee, and a dedicated reasoning rate keep billing as published."""
+        self._register_off_peak_model(
+            {"hours_utc": self.OFF_PEAK_WINDOW, "input_cost_per_token": 1e-07, "output_cost_per_token": 2e-07}
+        )
+        usage = Usage(
+            prompt_tokens=1000,
+            completion_tokens=200,
+            total_tokens=1200,
+            prompt_tokens_details=PromptTokensDetailsWrapper(web_search_requests=1),
+            completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=50),
+        )
+        usage.citation_tokens = 100
+
+        prompt_cost, completion_cost = perplexity_cost_per_token(
+            model=self.OFF_PEAK_MODEL, usage=usage, current_time=self.INSIDE_WINDOW
+        )
+
+        assert math.isclose(prompt_cost, (1000 * 1e-07) + (100 * 2e-06), rel_tol=1e-10)
+        assert math.isclose(completion_cost, (150 * 2e-07) + (50 * 3e-06) + 0.005, rel_tol=1e-10)
+
+    def test_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_model(
+            {"hours_utc": "00:00-00:00", "input_cost_per_token": 1e-07, "output_cost_per_token": 2e-07}
+        )
+        usage = Usage(prompt_tokens=1000, completion_tokens=200, total_tokens=1200)
+
+        prompt_cost, completion_cost = perplexity_cost_per_token(model=self.OFF_PEAK_MODEL, usage=usage)
+
+        assert math.isclose(prompt_cost, 1000 * 1e-07, rel_tol=1e-10)
+        assert math.isclose(completion_cost, 200 * 2e-07, rel_tol=1e-10)
+
+    def test_provider_stated_cost_still_wins_inside_an_off_peak_window(self):
+        """A response that carries Perplexity's own metered cost bills that cost whatever the
+        window says; the caller strips it when the deployment carries custom pricing."""
+        self._register_off_peak_model(
+            {"hours_utc": "00:00-00:00", "input_cost_per_token": 1e-07, "output_cost_per_token": 2e-07}
+        )
+        usage = Usage(prompt_tokens=1000, completion_tokens=200, total_tokens=1200)
+        usage.cost = {"total_cost": 0.00501}
+
+        prompt_cost, completion_cost = perplexity_cost_per_token(model=self.OFF_PEAK_MODEL, usage=usage)
+
+        assert prompt_cost == 0.0
+        assert completion_cost == 0.00501