mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-08 03:08:45 +00:00
feat(cost): support day-of-week qualified off-peak windows
This commit is contained in:
mateo-berri
parent
cc3ea1fb08
commit
7abed91523
3 changed files with 259 additions and 12 deletions
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue