From 207ec88b3a54606be9185001cf6544a1cea1f593 Mon Sep 17 00:00:00 2001 From: shrey kharbanda Date: Thu, 24 Sep 2026 04:09:55 +0000 Subject: [PATCH] feat(cost): add off_peak_dates and weekday_dates holiday exceptions to off_peak_pricing --- .../litellm_core_utils/llm_cost_calc/utils.py | 37 +++- litellm/types/utils.py | 7 +- .../llm_cost_calc/test_llm_cost_calc_utils.py | 186 ++++++++++++++++++ .../test_litellm/test_model_prices_schema.py | 5 + 4 files changed, 228 insertions(+), 7 deletions(-) diff --git a/litellm/litellm_core_utils/llm_cost_calc/utils.py b/litellm/litellm_core_utils/llm_cost_calc/utils.py index 46bf2ec2960..8a5deb293b0 100644 --- a/litellm/litellm_core_utils/llm_cost_calc/utils.py +++ b/litellm/litellm_core_utils/llm_cost_calc/utils.py @@ -466,18 +466,32 @@ def _weekday_calendar(weekday_timezone: object) -> tzinfo: return timezone.utc -def _matches_weekdays(reference_utc: datetime, weekdays: object, weekday_timezone: object) -> bool: +_MONDAY_TO_FRIDAY: Final = frozenset({1, 2, 3, 4, 5}) + + +def _calendar_date(reference_utc: datetime, weekday_timezone: object) -> str: + return reference_utc.astimezone(_weekday_calendar(weekday_timezone)).date().isoformat() + + +def _matches_weekdays( + reference_utc: datetime, weekdays: object, weekday_timezone: object, weekday_dates: frozenset[str] +) -> 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. + window past 16:00 UTC is where an Asia/Shanghai weekday diverges from the UTC one. A + calendar date listed in weekday_dates is a make-up workday, so it follows the + Monday-to-Friday rules whatever its real isoweekday is. """ 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 + local: Final = reference_utc.astimezone(_weekday_calendar(weekday_timezone)) + if local.date().isoformat() in weekday_dates: + return not allowed.isdisjoint(_MONDAY_TO_FRIDAY) + return local.isoweekday() in allowed def _as_window_strings(value: object) -> tuple[str, ...]: @@ -488,29 +502,40 @@ def _as_window_strings(value: object) -> tuple[str, ...]: return () +def _as_date_strings(value: object) -> frozenset[str]: + if isinstance(value, str) or not isinstance(value, Sequence): + return frozenset() + return frozenset(entry for entry in value if isinstance(entry, str)) + + 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. + hours apply only on its weekdays. off_peak_dates lists calendar dates that are off-peak + all day, and weekday_dates lists dates that follow the Monday-to-Friday rules even on a + weekend; both are read on the weekday_timezone calendar. """ reference: Final = current_time if current_time is not None else current_billing_time() reference_utc: Final = ( reference.astimezone(timezone.utc) if reference.tzinfo is not None else reference.replace(tzinfo=timezone.utc) ) + weekday_timezone: Final = off_peak.get("weekday_timezone") + if _calendar_date(reference_utc, weekday_timezone) in _as_date_strings(off_peak.get("off_peak_dates")): + return True 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") + weekday_dates: Final = _as_date_strings(off_peak.get("weekday_dates")) 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): + if not _matches_weekdays(reference_utc, rule.get("weekdays"), weekday_timezone, weekday_dates): continue if _is_within_off_peak_window(rule_windows, reference_utc): return True diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 064e3040054..0b8dc965427 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -247,12 +247,17 @@ class OffPeakPricing(TypedDict, total=False): 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. + to UTC. off_peak_dates lists YYYY-MM-DD dates, read on that calendar, that are off-peak + all day (public holidays), and weekday_dates lists dates that follow the Monday-to-Friday + rules even on a weekend (make-up workdays). Any rate left unset falls back to the + standard rate. """ hours_utc: ReadOnly[str | Sequence[str]] windows: ReadOnly[Sequence[OffPeakWindow]] weekday_timezone: ReadOnly[str] + off_peak_dates: ReadOnly[Sequence[str]] + weekday_dates: ReadOnly[Sequence[str]] input_cost_per_token: ReadOnly[float] output_cost_per_token: ReadOnly[float] output_cost_per_reasoning_token: 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 781a3a7c4ed..1b2d1f7f781 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 @@ -600,6 +600,192 @@ def test_is_off_peak_weekday_qualified_windows_deepseek_schedule(): assert _is_off_peak(deepseek, when) is True, f"{when.isoformat()} should bill off-peak" +def test_is_off_peak_holiday_weekday_is_off_peak_all_day(): + """A date in off_peak_dates is off-peak around the clock, including inside the peak + windows a plain weekday would bill at standard rates.""" + from datetime import datetime, timezone + + deepseek = { + "weekday_timezone": "Asia/Shanghai", + "off_peak_dates": ["2026-01-01", "2026-10-01"], + "weekday_dates": ["2026-01-04", "2026-02-14"], + "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]}, + ], + } + holiday = datetime(2026, 1, 1, 2, 0, tzinfo=timezone.utc) + holiday_second_window = datetime(2026, 1, 1, 7, 30, tzinfo=timezone.utc) + normal_thursday = datetime(2026, 1, 8, 2, 0, tzinfo=timezone.utc) + normal_thursday_second_window = datetime(2026, 1, 8, 7, 30, tzinfo=timezone.utc) + assert _is_off_peak(deepseek, holiday) is True, f"{holiday.isoformat()} is a public holiday" + assert _is_off_peak(deepseek, holiday_second_window) is True, ( + f"{holiday_second_window.isoformat()} is a public holiday" + ) + assert _is_off_peak(deepseek, normal_thursday) is False, ( + f"{normal_thursday.isoformat()} is a normal Thursday peak hour" + ) + assert _is_off_peak(deepseek, normal_thursday_second_window) is False, ( + f"{normal_thursday_second_window.isoformat()} is a normal Thursday peak hour" + ) + + +def test_is_off_peak_make_up_weekend_day_follows_weekday_rules(): + """A Sunday listed in weekday_dates is a make-up workday: the weekday windows apply, so + its peak hours bill standard and its weekday off-peak hours bill off-peak.""" + from datetime import datetime, timezone + + deepseek = { + "weekday_timezone": "Asia/Shanghai", + "off_peak_dates": ["2026-01-01", "2026-10-01"], + "weekday_dates": ["2026-01-04", "2026-02-14"], + "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]}, + ], + } + make_up_sunday_peak = datetime(2026, 1, 4, 2, 0, tzinfo=timezone.utc) + make_up_sunday_peak_second = datetime(2026, 1, 4, 7, 30, tzinfo=timezone.utc) + make_up_sunday_off_peak = datetime(2026, 1, 4, 5, 0, tzinfo=timezone.utc) + make_up_sunday_off_peak_second = datetime(2026, 1, 4, 12, 0, tzinfo=timezone.utc) + plain_sunday = datetime(2026, 1, 11, 2, 0, tzinfo=timezone.utc) + assert _is_off_peak(deepseek, make_up_sunday_peak) is False, ( + f"{make_up_sunday_peak.isoformat()} is a make-up workday peak hour" + ) + assert _is_off_peak(deepseek, make_up_sunday_peak_second) is False, ( + f"{make_up_sunday_peak_second.isoformat()} is a make-up workday peak hour" + ) + assert _is_off_peak(deepseek, make_up_sunday_off_peak) is True, ( + f"{make_up_sunday_off_peak.isoformat()} is a make-up workday off-peak hour" + ) + assert _is_off_peak(deepseek, make_up_sunday_off_peak_second) is True, ( + f"{make_up_sunday_off_peak_second.isoformat()} is a make-up workday off-peak hour" + ) + assert _is_off_peak(deepseek, plain_sunday) is True, ( + f"{plain_sunday.isoformat()} is a normal Sunday, off-peak all day" + ) + + +def test_is_off_peak_date_exceptions_read_on_weekday_timezone_calendar(): + """Both date lists are read on the weekday_timezone calendar, so the Shanghai date can + diverge from the UTC date over 16:00-24:00 UTC.""" + from datetime import datetime, timezone + + holiday_saturday_rule = { + "weekday_timezone": "Asia/Shanghai", + "windows": [{"hours_utc": "16:00-17:00", "weekdays": [6, 7]}], + "off_peak_dates": ["2026-01-01"], + } + shanghai_new_year = datetime(2025, 12, 31, 16, 30, tzinfo=timezone.utc) + utc_new_year = datetime(2026, 1, 1, 16, 30, tzinfo=timezone.utc) + assert _is_off_peak(holiday_saturday_rule, shanghai_new_year) is True, ( + f"{shanghai_new_year.isoformat()} is already 2026-01-01 in Shanghai, a holiday" + ) + assert _is_off_peak(holiday_saturday_rule, utc_new_year) is False, ( + f"{utc_new_year.isoformat()} is 2026-01-02 in Shanghai, no longer the holiday" + ) + + utc_holiday_saturday_rule = { + "windows": [{"hours_utc": "16:00-17:00", "weekdays": [6, 7]}], + "off_peak_dates": ["2026-01-01"], + } + assert _is_off_peak(utc_holiday_saturday_rule, shanghai_new_year) is False, ( + f"{shanghai_new_year.isoformat()} is still 2025-12-31 on the UTC calendar" + ) + assert _is_off_peak(utc_holiday_saturday_rule, utc_new_year) is True, ( + f"{utc_new_year.isoformat()} is 2026-01-01 on the UTC calendar" + ) + + make_up_weekday_rule = { + "weekday_timezone": "Asia/Shanghai", + "windows": [{"hours_utc": "16:00-17:00", "weekdays": [1, 2, 3, 4, 5]}], + "weekday_dates": ["2026-01-04"], + } + shanghai_make_up_sunday = datetime(2026, 1, 3, 16, 30, tzinfo=timezone.utc) + shanghai_saturday = datetime(2026, 1, 2, 16, 30, tzinfo=timezone.utc) + assert _is_off_peak(make_up_weekday_rule, shanghai_make_up_sunday) is True, ( + f"{shanghai_make_up_sunday.isoformat()} is Sunday 2026-01-04 in Shanghai, a make-up workday" + ) + assert _is_off_peak(make_up_weekday_rule, shanghai_saturday) is False, ( + f"{shanghai_saturday.isoformat()} is Saturday 2026-01-03 in Shanghai" + ) + + +def test_is_off_peak_make_up_day_does_not_match_weekend_only_rules(): + """A make-up workday follows the Monday-to-Friday rules, so a weekend-only rule must not + match it even though its real isoweekday is Sunday.""" + from datetime import datetime, timezone + + weekend_only = { + "windows": [{"hours_utc": "00:00-00:00", "weekdays": [6, 7]}], + "weekday_dates": ["2026-01-04"], + } + make_up_sunday = datetime(2026, 1, 4, 12, 0, tzinfo=timezone.utc) + plain_saturday = datetime(2026, 1, 3, 12, 0, tzinfo=timezone.utc) + assert _is_off_peak(weekend_only, make_up_sunday) is False, ( + f"{make_up_sunday.isoformat()} is a make-up workday, not a weekend day" + ) + assert _is_off_peak(weekend_only, plain_saturday) is True, ( + f"{plain_saturday.isoformat()} is a normal Saturday" + ) + + +def test_is_off_peak_ignores_malformed_date_exceptions(): + """A bare string or non-string entries are ignored, matching how a bare weekdays string + is treated: malformed config can never silently widen the off-peak hours.""" + from datetime import datetime, timezone + + holiday = datetime(2026, 1, 1, 12, 0, tzinfo=timezone.utc) + weekend_whole_day = [{"hours_utc": "00:00-00:00", "weekdays": [6, 7]}] + assert _is_off_peak({"off_peak_dates": "2026-01-01", "windows": weekend_whole_day}, holiday) is False, ( + "a bare off_peak_dates string is malformed and must be ignored" + ) + block = {"off_peak_dates": ["01/01/2026", 20260101, None], "windows": weekend_whole_day} + assert _is_off_peak(block, holiday) is False, "unparseable entries are simply never equal to a date" + make_up_sunday = datetime(2026, 1, 4, 12, 0, tzinfo=timezone.utc) + block = {"weekday_dates": "2026-01-04", "windows": weekend_whole_day} + assert _is_off_peak(block, make_up_sunday) is True, ( + f"{make_up_sunday.isoformat()} is a plain Sunday once the bare weekday_dates string is ignored" + ) + + +def test_get_token_base_cost_off_peak_date_bills_holiday_rates(): + """End to end through _get_token_base_cost: a holiday listed in off_peak_dates bills the + off-peak input rate during what would be a weekday peak hour.""" + 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": { + "off_peak_dates": ["2026-01-01"], + "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) + + holiday_peak_hours = _get_token_base_cost( + model_info, usage, current_time=datetime(2026, 1, 1, 2, 0, tzinfo=timezone.utc) + ) + assert holiday_peak_hours[:2] == (5e-7, 1e-6) + + normal_thursday_same_hours = _get_token_base_cost( + model_info, usage, current_time=datetime(2026, 1, 8, 2, 0, tzinfo=timezone.utc) + ) + assert normal_thursday_same_hours[:2] == (1e-6, 2e-6) + + 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: diff --git a/tests/test_litellm/test_model_prices_schema.py b/tests/test_litellm/test_model_prices_schema.py index 052278631e2..4d92ecaed8c 100644 --- a/tests/test_litellm/test_model_prices_schema.py +++ b/tests/test_litellm/test_model_prices_schema.py @@ -137,6 +137,8 @@ OFF_PEAK_ENTRY: Final = MappingProxyType( "hours_utc": "16:30-00:30", "windows": [{"hours_utc": ["00:30-02:00"], "weekdays": [6, "Sunday", "mon", "THURS"]}], "weekday_timezone": "Asia/Shanghai", + "off_peak_dates": ["2026-01-01"], + "weekday_dates": ["2026-01-04"], "input_cost_per_token": 1e-6, "output_cost_per_token": 4e-6, "cache_read_input_token_cost": 1e-7, @@ -167,6 +169,9 @@ def test_generator_classifies_off_peak_pricing_as_a_windowed_rate_block(): {"hours_utc": "25:00-01:00", "input_cost_per_token": 1e-6}, {"hours_utc": ["16:30-00:30", "4pm-midnight"], "input_cost_per_token": 1e-6}, {"windows": [{"hours_utc": "00:30-02:00", "weekdays": ["Funday"]}], "input_cost_per_token": 1e-6}, + {"hours_utc": "16:30-00:30", "off_peak_dates": ["2026-1-1"], "input_cost_per_token": 1e-6}, + {"hours_utc": "16:30-00:30", "off_peak_dates": "2026-01-01", "input_cost_per_token": 1e-6}, + {"hours_utc": "16:30-00:30", "weekday_dates": [], "input_cost_per_token": 1e-6}, ], ) def test_generated_off_peak_schema_rejects_malformed_blocks(block: dict):