diff --git a/README.md b/README.md
index be8b0c42..34204715 100644
--- a/README.md
+++ b/README.md
@@ -64,11 +64,15 @@ prefix nearly free, `/eco` shrinks the fresh suffix every turn actually pays for
# Run long agentic coding sessions for *pennies*
-### Cache-hit input bills at **`~$0.0435` / 1M tokens** — about **230× cheaper** than Claude Fable 5 (`$10` / 1M).
+### Cache-hit input bills at **`$0.022` / 1M tokens** — about **450× cheaper** than Claude Fable 5 (`$10` / 1M).
ClawCodex keeps your request prefix **byte-stable**, so DeepSeek's prompt cache covers your whole
`system + tools + history` span across turns. **The longer you code, the more you save.**
+`deepseek-v4-pro` off-peak, checked 2026-08-25. DeepSeek doubles every rate during peak hours
+(01:00–04:00 and 06:00–10:00 UTC, Mon–Fri) — still ~227× cheaper there. `/cost` follows the
+schedule.
+
***
diff --git a/src/services/cost_restore.py b/src/services/cost_restore.py
index 662604ee..2d11f3f1 100644
--- a/src/services/cost_restore.py
+++ b/src/services/cost_restore.py
@@ -68,6 +68,14 @@ def build_cost_block() -> dict[str, Any]:
# display or budget gate (those keep reading the billed ``total_cost_usd``
# / ``cost_usd``); it exists so downstream trajectory/leaderboard tooling
# has a comparable cost column.
+ #
+ # Computed from AGGREGATED per-model totals, so it is exact only for
+ # models with no per-request rate tier. It already over-prices a
+ # context-tiered model (gpt-5.6-luna doubles above 272K prompt tokens)
+ # by treating a whole session as one giant request; for a time-tiered
+ # one (DeepSeek V4 doubles during UTC peak hours) it prices the whole
+ # session at the rate in force when the snapshot is written. Both need
+ # per-request cost records to fix, not a different call here.
from src.services.pricing import compute_cost
estimated_cost_usd = 0.0
diff --git a/src/services/cost_tracker.py b/src/services/cost_tracker.py
index a4bea75e..40962026 100644
--- a/src/services/cost_tracker.py
+++ b/src/services/cost_tracker.py
@@ -155,9 +155,17 @@ def get_total_output_tokens(self) -> int:
return self._total_output_tokens
def get_cache_savings(self) -> float:
+ # ``request_time=event.timestamp`` matters for time-tiered cards
+ # (DeepSeek V4 doubles during UTC peak hours): this runs at DISPLAY
+ # time, arbitrarily later than the requests it is summing, so pricing
+ # each event at "now" would report an off-peak session's savings at
+ # peak rates once the clock crossed 01:00 UTC.
total_savings = 0.0
for event in self._events:
- pricing = _get_pricing(event.model) or DEFAULT_PRICING
+ pricing = (
+ _get_pricing(event.model, request_time=event.timestamp)
+ or DEFAULT_PRICING
+ )
saved_per_token = pricing["input"] - pricing["cache_read"]
total_savings += event.cache_read_input_tokens * saved_per_token
return total_savings
diff --git a/src/services/pricing.py b/src/services/pricing.py
index 3fd70f2e..55fc2d7b 100644
--- a/src/services/pricing.py
+++ b/src/services/pricing.py
@@ -5,6 +5,12 @@
``add_to_total_cost_state`` and friends); this module just computes the
dollar cost of a usage record.
+One exception to "pure", added for DeepSeek's peak/off-peak card: when a
+caller omits ``request_time``, ``get_pricing`` reads the wall clock to decide
+which side of that schedule a request falls on. Every caller that knows the
+real request time should pass it; "now" is only correct because the live path
+prices a response the moment it arrives.
+
Pricing mirrors ``typescript/src/utils/modelCost.ts``: published Anthropic
list prices per million tokens for first-party direct calls. Proxies
(litellm, openrouter, bedrock, vertex) may apply different rates;
@@ -19,6 +25,8 @@
from __future__ import annotations
+import time
+from datetime import datetime, timezone
from typing import Any
@@ -75,24 +83,77 @@
"cache_creation": 0.30 / 1_000_000,
"cache_read": 0.03 / 1_000_000,
}
-# DeepSeek V4 (USD per million tokens). DeepSeek's automatic prefix cache
-# bills cache HITS at the low ``cache_read`` rate and cache MISSES at the
-# normal input rate; there is no separate cache-write charge, so
-# ``cache_creation`` mirrors ``input`` (a non-cached token is just input).
-# DeepSeekProvider maps its usage onto the Anthropic convention
-# (``input_tokens`` = miss, ``cache_read_input_tokens`` = hit), so these tiers
-# price correctly through the generic ``compute_cost``.
-_TIER_DEEPSEEK_FLASH = {
- "input": 0.14 / 1_000_000,
- "output": 0.28 / 1_000_000,
- "cache_creation": 0.14 / 1_000_000,
- "cache_read": 0.0028 / 1_000_000,
+# DeepSeek V4 (USD per million tokens) — checked 2026-08-25 against
+# https://api-docs.deepseek.com/quick_start/pricing/
+#
+# DeepSeek's automatic prefix cache bills cache HITS at the low ``cache_read``
+# rate and cache MISSES at the normal input rate; there is no separate
+# cache-write charge, so ``cache_creation`` mirrors ``input`` (a non-cached
+# token is just input). DeepSeekProvider maps its usage onto the Anthropic
+# convention (``input_tokens`` = miss, ``cache_read_input_tokens`` = hit), so
+# these tiers price correctly through the generic ``compute_cost``.
+#
+# TIME-TIERED since 2026-08-16: every rate DOUBLES during peak hours
+# (01:00-04:00 and 06:00-10:00 UTC, Monday through Friday); that is 7 hours
+# on each of 5 days, so off-peak covers 133 of every 168 hours. This is the
+# file's third tier axis, and the only one that is not a property of the
+# request's content — a prompt does not get cheaper at midnight, the clock
+# does. See ``is_deepseek_peak``.
+#
+# CORRECTED 2026-08-25 (issue #904): these rows held DeepSeek's
+# pre-2026-08-16 card — 0.14/0.28 flash, 0.435/0.87 pro — which was itself a
+# time-limited promotion, and no axis existed for the schedule that replaced
+# it. ``cache_read`` drove the error: at ~96% of agentic input tokens,
+# 0.003625 against a real 0.022 understated a deepseek-v4-pro session 2.3x
+# off-peak and 4.5x peak. A number that low still reads as authoritative, in
+# exactly the way ``get_pricing`` returning None does not. Hence the
+# checked-on date above: of everything in this file, DeepSeek is the row most
+# likely to move again, and a promo card that rots into silent
+# under-reporting is the same failure the Luna row below was written about.
+#
+# Not registered: ``deepseek-v4-flash-vision-exp``, which shares the flash
+# card on the vendor's page but has no row in ``models/configs.py`` and is
+# unreachable through the provider (``supports_vision=False``).
+_TIER_DEEPSEEK_FLASH_OFF_PEAK = {
+ "input": 0.22 / 1_000_000,
+ "output": 0.66 / 1_000_000,
+ "cache_creation": 0.22 / 1_000_000,
+ "cache_read": 0.007 / 1_000_000,
}
-_TIER_DEEPSEEK_PRO = {
- "input": 0.435 / 1_000_000,
- "output": 0.87 / 1_000_000,
- "cache_creation": 0.435 / 1_000_000,
- "cache_read": 0.003625 / 1_000_000,
+_TIER_DEEPSEEK_FLASH_PEAK = {
+ "input": 0.44 / 1_000_000,
+ "output": 1.32 / 1_000_000,
+ "cache_creation": 0.44 / 1_000_000,
+ "cache_read": 0.014 / 1_000_000,
+}
+_TIER_DEEPSEEK_PRO_OFF_PEAK = {
+ "input": 0.66 / 1_000_000,
+ "output": 1.98 / 1_000_000,
+ "cache_creation": 0.66 / 1_000_000,
+ "cache_read": 0.022 / 1_000_000,
+}
+_TIER_DEEPSEEK_PRO_PEAK = {
+ "input": 1.32 / 1_000_000,
+ "output": 3.96 / 1_000_000,
+ "cache_creation": 1.32 / 1_000_000,
+ "cache_read": 0.044 / 1_000_000,
+}
+# Peak windows as half-open ``[start_hour, end_hour)`` in UTC, applied Monday
+# through Friday. Half-open is the reading that makes the two windows tile
+# without overlap: 03:59:59 UTC is peak, 04:00:00 is not. The vendor page
+# states the windows to the hour and says nothing finer, so hour granularity
+# is exact rather than a rounding.
+_DEEPSEEK_PEAK_WINDOWS_UTC: tuple[tuple[int, int], ...] = ((1, 4), (6, 10))
+# Canonical model id -> (off-peak card, peak card).
+_DEEPSEEK_TIERS: dict[str, tuple[dict[str, float], dict[str, float]]] = {
+ "deepseek-v4-flash": (
+ _TIER_DEEPSEEK_FLASH_OFF_PEAK,
+ _TIER_DEEPSEEK_FLASH_PEAK,
+ ),
+ "deepseek-v4-pro": (
+ _TIER_DEEPSEEK_PRO_OFF_PEAK,
+ _TIER_DEEPSEEK_PRO_PEAK,
+ ),
}
# MiniMax M3 pay-as-you-go rates in USD per million tokens. Prompt size is the
# complete request input, including cache creation and cache read tokens.
@@ -263,8 +324,15 @@
# DeepSeek V4 (api.deepseek.com). OpenRouter's ``deepseek/…`` ids resolve
# here too via get_pricing's vendor-prefix strip — consistent with how
# every proxied model is priced at its upstream rate.
- "deepseek-v4-flash": _TIER_DEEPSEEK_FLASH,
- "deepseek-v4-pro": _TIER_DEEPSEEK_PRO,
+ # VALUES UNUSED, same as the gpt-5.6-luna rows below: these two entries
+ # are membership gates for ``get_pricing``'s ``model in PRICING`` checks,
+ # and the live card is picked by request time in ``_get_exact_pricing``,
+ # which returns before reaching ``PRICING.get(model)``. They point at the
+ # off-peak card so that anything reading the table directly (the legacy
+ # ``services.cost_tracker`` fallback path) gets the rate that covers 133
+ # of every 168 hours rather than a number picked for tidiness.
+ "deepseek-v4-flash": _TIER_DEEPSEEK_FLASH_OFF_PEAK,
+ "deepseek-v4-pro": _TIER_DEEPSEEK_PRO_OFF_PEAK,
"MiniMax-M3": _TIER_MINIMAX_M3_STANDARD,
"MiniMax-M2.7": _TIER_MINIMAX_M27,
# Meta Muse Spark (api.meta.ai)
@@ -322,11 +390,36 @@
]
+def is_deepseek_peak(request_time: float | None = None) -> bool:
+ """True if ``request_time`` falls inside DeepSeek's peak-rate schedule.
+
+ ``request_time`` is POSIX epoch seconds (what ``time.time()`` returns);
+ ``None`` means now. Peak is 01:00-04:00 and 06:00-10:00 UTC, Monday
+ through Friday, evaluated in UTC — the vendor's schedule is stated in UTC
+ and does not follow the caller's local calendar, so a Friday 23:00
+ US/Pacific request is a Saturday in UTC and off-peak.
+
+ Public because it is the only way to explain a DeepSeek cost figure
+ without re-deriving the schedule: the same request costs 2x more at
+ 07:00 UTC on a Tuesday than at 07:00 UTC on a Sunday, and a caller that
+ wants to say so in a status bar or a ``/cost`` breakdown should ask here
+ rather than reimplement the windows.
+ """
+ ts = time.time() if request_time is None else request_time
+ when = datetime.fromtimestamp(ts, timezone.utc)
+ if when.weekday() > 4: # Saturday/Sunday — off-peak all day.
+ return False
+ return any(
+ start <= when.hour < end for start, end in _DEEPSEEK_PEAK_WINDOWS_UTC
+ )
+
+
def _get_exact_pricing(
model: str,
*,
input_tokens: int,
service_tier: str,
+ request_time: float | None,
) -> dict[str, float] | None:
# Context-tiered models: the published rate depends on how big THIS
# request's prompt is. ``model`` is already the canonical bare key here
@@ -337,6 +430,15 @@ def _get_exact_pricing(
if input_tokens > _GPT_56_LUNA_INPUT_TIER_LIMIT
else _TIER_GPT_56_LUNA
)
+ # Time-tiered models: the published rate depends on WHEN the request was
+ # sent and on nothing inside it. Neither existing axis can carry this —
+ # ``input_tokens`` is prompt size, and ``service_tier`` is whatever the
+ # provider declared in its response (DeepSeek declares nothing, so it
+ # always resolves to "standard").
+ deepseek = _DEEPSEEK_TIERS.get(model)
+ if deepseek is not None:
+ off_peak, peak = deepseek
+ return peak if is_deepseek_peak(request_time) else off_peak
if model != "MiniMax-M3":
return PRICING.get(model)
@@ -359,6 +461,7 @@ def get_pricing(
*,
input_tokens: int = 0,
service_tier: str = "standard",
+ request_time: float | None = None,
) -> dict[str, float] | None:
"""Return per-token prices for ``model``, or ``None`` if unknown.
@@ -366,6 +469,16 @@ def get_pricing(
uses it with ``service_tier`` to select its standard/priority and
short/long-context rate.
+ ``request_time`` is POSIX epoch seconds for when the request was sent;
+ DeepSeek V4 uses it to pick its peak or off-peak card (see
+ ``is_deepseek_peak``). ``None`` means now, which is correct for the live
+ path — ``cost_tracker.record_api_usage`` prices a response the moment it
+ arrives — and an approximation anywhere a stored usage record is repriced
+ later. Callers holding a real timestamp should pass it. Note that a
+ request straddling a boundary is priced by the single instant handed in;
+ the vendor page does not say whether it bills by request start or by
+ completion, and both are within one turn of each other.
+
Lookup order:
1. Exact match in ``PRICING``.
2. Strip a leading ``/`` segment (openrouter convention,
@@ -388,6 +501,7 @@ def get_pricing(
model,
input_tokens=input_tokens,
service_tier=service_tier,
+ request_time=request_time,
)
if "/" in model:
bare = model.split("/", 1)[1]
@@ -396,6 +510,7 @@ def get_pricing(
bare,
input_tokens=input_tokens,
service_tier=service_tier,
+ request_time=request_time,
)
for prefix, pricing in _FAMILY_PREFIXES:
if bare.startswith(prefix):
@@ -413,8 +528,13 @@ def is_known_pricing(model: str) -> bool:
return get_pricing(model) is not None
-def compute_cost(model: str, usage: dict[str, Any]) -> float:
- """Compute USD cost for a usage record. Pure function.
+def compute_cost(
+ model: str,
+ usage: dict[str, Any],
+ *,
+ request_time: float | None = None,
+) -> float:
+ """Compute USD cost for a usage record.
Returns 0.0 when the model has no pricing entry (rather than
guessing with ``DEFAULT_PRICING``). The legacy cost-tracker facade
@@ -425,6 +545,16 @@ def compute_cost(model: str, usage: dict[str, Any]) -> float:
``cache_creation_input_tokens``, and ``cache_read_input_tokens``
from ``usage``. Missing keys default to zero so callers that only
track input+output still get a sensible result.
+
+ ``request_time`` (POSIX epoch seconds) is forwarded to ``get_pricing``
+ for DeepSeek's peak/off-peak card; ``None`` prices at the current clock.
+ Callers that recompute from an AGGREGATED ``model_usage`` block rather
+ than from one response — ``cost_restore.build_cost_block``,
+ ``eval/harbor/advisor_cost.py`` — have no per-request time to pass and
+ inherit the same aggregate inexactness they already carry for the
+ context tier: a session spanning a peak boundary gets priced entirely on
+ one side of it. Fixing that needs per-request cost records, not a
+ different default here.
"""
input_tokens = int(usage.get("input_tokens", 0) or 0)
output_tokens = int(usage.get("output_tokens", 0) or 0)
@@ -435,6 +565,7 @@ def compute_cost(model: str, usage: dict[str, Any]) -> float:
model,
input_tokens=prompt_tokens,
service_tier=str(usage.get("service_tier") or "standard"),
+ request_time=request_time,
)
if pricing is None:
return 0.0
@@ -521,6 +652,7 @@ def compute_session_cost(
"PRICING",
"DEFAULT_PRICING",
"get_pricing",
+ "is_deepseek_peak",
"is_known_pricing",
"compute_cost",
"compute_session_cost",
diff --git a/tests/test_deepseek_peak_pricing.py b/tests/test_deepseek_peak_pricing.py
new file mode 100644
index 00000000..5337fc03
--- /dev/null
+++ b/tests/test_deepseek_peak_pricing.py
@@ -0,0 +1,324 @@
+"""DeepSeek V4's peak/off-peak rate schedule (issue #904).
+
+Since 2026-08-16 DeepSeek publishes a peak and an off-peak card: every rate
+doubles during 01:00-04:00 and 06:00-10:00 UTC, Monday through Friday. That
+is the pricing table's third tier axis, and the only one that is not a
+property of the request's content — ``input_tokens`` is prompt size and
+``service_tier`` is what the provider declared in its response, so neither
+could carry it.
+
+Two things are worth pinning here and are pinned separately:
+
+* the SCHEDULE — which instants are peak — because an off-by-one on a window
+ boundary or a missed weekend rule mis-prices a whole class of requests
+ silently; and
+* the CARD — the absolute published rates — because the values this issue
+ replaced were internally consistent (correct ratios, mirrored
+ cache_creation) and still 3x low. Only an external number catches that,
+ which is the same lesson the gpt-5.6-luna row in ``services/pricing.py``
+ records.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from src.services.pricing import (
+ compute_cost,
+ get_pricing,
+ is_deepseek_peak,
+)
+
+
+MODELS = ("deepseek-v4-flash", "deepseek-v4-pro")
+
+
+def _utc(year: int, month: int, day: int, hour: int, minute: int = 0,
+ second: int = 0) -> float:
+ return datetime(
+ year, month, day, hour, minute, second, tzinfo=timezone.utc
+ ).timestamp()
+
+
+# 2026-08-24 is a Monday, 2026-08-28 a Friday, 2026-08-29 a Saturday and
+# 2026-08-30 a Sunday.
+MON, FRI, SAT, SUN = 24, 28, 29, 30
+
+# One instant on each side of the schedule, for the tests that care which
+# card is in force rather than where the boundaries are.
+OFF_PEAK_TS = _utc(2026, 8, MON, 12)
+PEAK_TS = _utc(2026, 8, MON, 2)
+
+
+# --------------------------------------------------------------------------- #
+# The schedule
+# --------------------------------------------------------------------------- #
+
+@pytest.mark.parametrize("hour", [1, 2, 3, 6, 7, 8, 9])
+def test_weekday_peak_hours(hour: int) -> None:
+ assert is_deepseek_peak(_utc(2026, 8, MON, hour)) is True
+
+
+@pytest.mark.parametrize("hour", [0, 4, 5, 10, 11, 17, 23])
+def test_weekday_off_peak_hours(hour: int) -> None:
+ """Including 04:00 and 10:00 — the windows are half-open, so the hour a
+ window ends on is already off-peak."""
+ assert is_deepseek_peak(_utc(2026, 8, MON, hour)) is False
+
+
+def test_window_boundaries_are_half_open() -> None:
+ assert is_deepseek_peak(_utc(2026, 8, MON, 0, 59, 59)) is False
+ assert is_deepseek_peak(_utc(2026, 8, MON, 1, 0, 0)) is True
+ assert is_deepseek_peak(_utc(2026, 8, MON, 3, 59, 59)) is True
+ assert is_deepseek_peak(_utc(2026, 8, MON, 4, 0, 0)) is False
+ assert is_deepseek_peak(_utc(2026, 8, MON, 5, 59, 59)) is False
+ assert is_deepseek_peak(_utc(2026, 8, MON, 6, 0, 0)) is True
+ assert is_deepseek_peak(_utc(2026, 8, MON, 9, 59, 59)) is True
+ assert is_deepseek_peak(_utc(2026, 8, MON, 10, 0, 0)) is False
+
+
+@pytest.mark.parametrize("day", [SAT, SUN])
+@pytest.mark.parametrize("hour", [1, 2, 3, 6, 7, 9])
+def test_weekends_are_off_peak_even_inside_the_windows(day: int, hour: int) -> None:
+ assert is_deepseek_peak(_utc(2026, 8, day, hour)) is False
+
+
+def test_monday_and_friday_are_both_weekdays() -> None:
+ """Fencepost on the Mon-Fri rule: a `weekday() >= 5` weekend test and a
+ `1 <= weekday() <= 5` one differ only on these two days."""
+ assert is_deepseek_peak(_utc(2026, 8, MON, 2)) is True
+ assert is_deepseek_peak(_utc(2026, 8, FRI, 2)) is True
+
+
+def test_schedule_is_evaluated_in_utc_not_local_time() -> None:
+ """Friday 23:00 US/Pacific is Saturday 06:00 UTC — inside a peak window by
+ the local calendar, off-peak by the vendor's."""
+ assert is_deepseek_peak(_utc(2026, 8, SAT, 6)) is False
+
+
+def test_off_peak_covers_133_of_168_hours() -> None:
+ """The vendor's schedule leaves 35 peak hours a week — 7 hours (3 + 4) on
+ each of 5 days — so 133 hours are off-peak. A sweep of every hour in a
+ week is the cheapest guard against a window silently widening."""
+ start = _utc(2026, 8, MON, 0)
+ peak_hours = sum(
+ is_deepseek_peak(start + h * 3600) for h in range(7 * 24)
+ )
+ assert peak_hours == 35
+ assert 7 * 24 - peak_hours == 133
+
+
+class _FakeClock:
+ """Stands in for the ``time`` module inside ``services.pricing`` only.
+
+ Patching the name in that module's namespace rather than ``time.time``
+ itself keeps the fake off pytest's own clock.
+ """
+
+ def __init__(self, ts: float) -> None:
+ self.ts = ts
+
+ def time(self) -> float:
+ return self.ts
+
+
+def test_omitting_request_time_reads_the_clock(monkeypatch) -> None:
+ """``request_time=None`` means "price at the current clock", which is what
+ makes the live path correct without passing anything: cost is computed the
+ moment a response arrives. Pinned end to end — through
+ ``is_deepseek_peak``, ``get_pricing`` and ``compute_cost`` — because a
+ default that silently stopped reaching the clock would leave every
+ production call site on one card with nothing failing."""
+ usage = {"input_tokens": 1_000_000}
+
+ monkeypatch.setattr("src.services.pricing.time", _FakeClock(PEAK_TS))
+ assert is_deepseek_peak() is True
+ assert get_pricing("deepseek-v4-pro")["input"] == 1.32 / 1_000_000
+ assert compute_cost("deepseek-v4-pro", usage) == pytest.approx(1.32)
+
+ monkeypatch.setattr("src.services.pricing.time", _FakeClock(OFF_PEAK_TS))
+ assert is_deepseek_peak() is False
+ assert get_pricing("deepseek-v4-pro")["input"] == 0.66 / 1_000_000
+ assert compute_cost("deepseek-v4-pro", usage) == pytest.approx(0.66)
+
+
+# --------------------------------------------------------------------------- #
+# The card
+# --------------------------------------------------------------------------- #
+
+# Published USD per 1M tokens, read 2026-08-25 from
+# https://api-docs.deepseek.com/quick_start/pricing/
+PUBLISHED = {
+ "deepseek-v4-flash": {
+ "off_peak": {"input": 0.22, "output": 0.66, "cache_read": 0.007},
+ "peak": {"input": 0.44, "output": 1.32, "cache_read": 0.014},
+ },
+ "deepseek-v4-pro": {
+ "off_peak": {"input": 0.66, "output": 1.98, "cache_read": 0.022},
+ "peak": {"input": 1.32, "output": 3.96, "cache_read": 0.044},
+ },
+}
+
+@pytest.mark.parametrize("model", MODELS)
+@pytest.mark.parametrize("window", ["off_peak", "peak"])
+def test_published_rates(model: str, window: str) -> None:
+ ts = OFF_PEAK_TS if window == "off_peak" else PEAK_TS
+ pricing = get_pricing(model, request_time=ts)
+ assert pricing is not None
+ for field, dollars in PUBLISHED[model][window].items():
+ assert pricing[field] == dollars / 1_000_000, field
+ # No separate cache-WRITE charge on this provider: a miss is just input.
+ assert pricing["cache_creation"] == pricing["input"]
+
+
+@pytest.mark.parametrize("model", MODELS)
+def test_peak_is_exactly_double_off_peak(model: str) -> None:
+ off = get_pricing(model, request_time=OFF_PEAK_TS)
+ peak = get_pricing(model, request_time=PEAK_TS)
+ assert off.keys() == peak.keys()
+ for field in off:
+ assert peak[field] == pytest.approx(2 * off[field], rel=1e-12), field
+
+
+def test_pro_is_three_times_flash_except_on_cache_read() -> None:
+ """The vendor prices pro at exactly 3x flash on input and output — but NOT
+ on cache read, where $0.022 against $0.007 is 22/7, not 3.
+
+ Pinned because that is exactly the kind of near-ratio that invites
+ deriving one row from the other. The cache-read rate has to be read off
+ the page, and at ~96% of agentic input tokens it is the field that moves
+ the bill most.
+ """
+ for ts in (OFF_PEAK_TS, PEAK_TS):
+ flash = get_pricing("deepseek-v4-flash", request_time=ts)
+ pro = get_pricing("deepseek-v4-pro", request_time=ts)
+ for field in ("input", "output", "cache_creation"):
+ assert pro[field] == pytest.approx(3 * flash[field], rel=1e-9), field
+ assert pro["cache_read"] == pytest.approx(
+ flash["cache_read"] * 22 / 7, rel=1e-9
+ )
+ assert pro["cache_read"] != pytest.approx(
+ 3 * flash["cache_read"], rel=1e-9
+ )
+
+
+# --------------------------------------------------------------------------- #
+# Wiring: the axis reaches compute_cost, and reaches nothing else
+# --------------------------------------------------------------------------- #
+
+def _agent_mix(total: int = 1_000_000) -> dict[str, int]:
+ """The token mix issue #904 measured on a real agent trace: 95.64% cache
+ read, 4.07% cache miss, 0.29% output. ``cache_read`` dominating is why a
+ wrong cache-read rate moved the total more than input and output combined.
+ """
+ return {
+ "cache_read_input_tokens": round(total * 0.9564),
+ "input_tokens": round(total * 0.0407),
+ "output_tokens": round(total * 0.0029),
+ "cache_creation_input_tokens": 0,
+ }
+
+
+@pytest.mark.parametrize("model", MODELS)
+def test_compute_cost_doubles_inside_a_peak_window(model: str) -> None:
+ usage = _agent_mix()
+ off = compute_cost(model, usage, request_time=OFF_PEAK_TS)
+ peak = compute_cost(model, usage, request_time=PEAK_TS)
+ assert off > 0
+ assert peak == pytest.approx(2 * off, rel=1e-12)
+
+
+def test_agent_mix_cost_matches_the_published_card() -> None:
+ """The end-to-end number issue #904 reported as 2.3x/4.5x low. Recomputed
+ from the published card rather than from the tiers, so a future edit to
+ ``services/pricing.py`` alone cannot make this pass."""
+ usage = _agent_mix()
+ for model in MODELS:
+ for window, ts in (("off_peak", OFF_PEAK_TS), ("peak", PEAK_TS)):
+ card = PUBLISHED[model][window]
+ expected = (
+ usage["input_tokens"] * card["input"]
+ + usage["output_tokens"] * card["output"]
+ + usage["cache_read_input_tokens"] * card["cache_read"]
+ ) / 1_000_000
+ got = compute_cost(model, usage, request_time=ts)
+ assert got == pytest.approx(expected, rel=1e-12), (model, window)
+
+
+def test_pro_off_peak_agent_mix_is_the_issues_number() -> None:
+ """$0.0536 per 1M tokens off-peak, $0.1073 peak — against the $0.0237 the
+ stale card returned. Pinned as an absolute so a regression to any card
+ that merely has the right shape is visible as a dollar figure."""
+ usage = _agent_mix()
+ assert compute_cost(
+ "deepseek-v4-pro", usage, request_time=OFF_PEAK_TS
+ ) == pytest.approx(0.0536, abs=5e-5)
+ assert compute_cost(
+ "deepseek-v4-pro", usage, request_time=PEAK_TS
+ ) == pytest.approx(0.1073, abs=5e-5)
+
+
+def test_vendor_prefix_stripped_ids_follow_the_schedule() -> None:
+ """OpenRouter's ``deepseek/…`` ids resolve to the upstream card via
+ get_pricing's prefix strip, so they must carry the timestamp too."""
+ for model in MODELS:
+ for ts in (OFF_PEAK_TS, PEAK_TS):
+ assert get_pricing(f"deepseek/{model}", request_time=ts) == (
+ get_pricing(model, request_time=ts)
+ )
+
+
+@pytest.mark.parametrize(
+ "model",
+ ["claude-opus-5", "claude-sonnet-5", "MiniMax-M3", "kimi-k3",
+ "gpt-5.6-luna", "muse-spark-1.1"],
+)
+def test_no_other_model_is_time_tiered(model: str) -> None:
+ """Scope gate: the new axis is DeepSeek-only. Every other row must return
+ the same card at every instant of the week."""
+ baseline = get_pricing(model, request_time=OFF_PEAK_TS)
+ assert baseline is not None
+ start = _utc(2026, 8, MON, 0)
+ for h in range(7 * 24):
+ assert get_pricing(model, request_time=start + h * 3600) == baseline
+
+
+def test_unknown_models_still_return_none_at_every_hour() -> None:
+ start = _utc(2026, 8, MON, 0)
+ for h in range(0, 7 * 24, 6):
+ assert get_pricing("totally-unknown-model-xyz",
+ request_time=start + h * 3600) is None
+
+
+def test_cache_savings_price_events_at_their_own_timestamp() -> None:
+ """``get_cache_savings`` runs at DISPLAY time, arbitrarily later than the
+ requests it sums. An off-peak session's savings must not be restated at
+ peak rates because the user happened to open ``/cost`` at 02:00 UTC."""
+ from src.services.cost_tracker import CostTracker
+
+ tracker = CostTracker()
+ tracker.record_usage("deepseek-v4-pro", {
+ "input_tokens": 10_000,
+ "output_tokens": 1_000,
+ "cache_read_input_tokens": 900_000,
+ })
+ # Back-date the recorded event into an off-peak window, then read the
+ # savings back as if the clock had since moved into a peak one.
+ tracker._events[0].timestamp = OFF_PEAK_TS
+ saved = tracker.get_cache_savings()
+ off = get_pricing("deepseek-v4-pro", request_time=OFF_PEAK_TS)
+ expected = 900_000 * (off["input"] - off["cache_read"])
+ assert saved == pytest.approx(expected, rel=1e-12)
+
+
+def test_leap_second_free_arithmetic_across_a_dst_shift() -> None:
+ """UTC has no DST, so a fixed 24h offset lands on the same wall hour. This
+ guards against anyone reimplementing the window check in local time."""
+ base = _utc(2026, 3, 27, 2) # Friday 02:00 UTC, inside a peak window
+ assert is_deepseek_peak(base) is True
+ day_later = (
+ datetime.fromtimestamp(base, timezone.utc) + timedelta(days=1)
+ ).timestamp()
+ assert is_deepseek_peak(day_later) is False # Saturday
diff --git a/tests/test_deepseek_prefix_cache.py b/tests/test_deepseek_prefix_cache.py
index 9cfdf127..e1d9ab2f 100644
--- a/tests/test_deepseek_prefix_cache.py
+++ b/tests/test_deepseek_prefix_cache.py
@@ -13,10 +13,14 @@
DeepSeek, keeping the system+history prefix byte-stable even when volatile
sections (e.g. the mutable MEMORY.md body) change — while leaving every
other provider's request bytes identical.
+* The DeepSeek rate card, including its peak/off-peak schedule (the schedule
+ itself is exercised in ``tests/test_deepseek_peak_pricing.py``).
"""
from __future__ import annotations
+from datetime import datetime, timezone
+
from src.context_system.prompt_assembly import build_full_system_prompt_blocks
from src.models import get_context_window_for_model, get_model_max_output_tokens
from src.providers.base import BaseProvider
@@ -166,24 +170,67 @@ def test_other_provider_usage_unchanged_by_cache_fields():
# Cost wiring (services/pricing.py)
# --------------------------------------------------------------------------- #
+# Pinned instants for the peak/off-peak card. Rates are time-tiered, so
+# every pricing assertion here passes an explicit ``request_time`` rather
+# than letting it default to whenever CI happens to run. The full schedule
+# lives in ``tests/test_deepseek_peak_pricing.py``.
+_OFF_PEAK = datetime(2026, 8, 24, 12, 0, tzinfo=timezone.utc).timestamp() # Mon
+_PEAK = datetime(2026, 8, 24, 2, 0, tzinfo=timezone.utc).timestamp() # Mon
+
+
def test_deepseek_pricing_registered():
+ """Published rates, pinned as absolutes on both sides of the schedule.
+
+ Checked 2026-08-25 against api-docs.deepseek.com/quick_start/pricing/.
+ Pinning absolutes (rather than ratios) is what catches a stale card —
+ the pre-2026-08-16 values in issue #904 were internally consistent and
+ still 3x low.
+ """
from src.services.pricing import get_pricing
- flash = get_pricing("deepseek-v4-flash")
- pro = get_pricing("deepseek-v4-pro")
+ flash = get_pricing("deepseek-v4-flash", request_time=_OFF_PEAK)
+ pro = get_pricing("deepseek-v4-pro", request_time=_OFF_PEAK)
assert flash is not None and pro is not None
- assert flash["input"] == 0.14 / 1_000_000
- assert flash["cache_read"] == 0.0028 / 1_000_000
- assert pro["input"] == 0.435 / 1_000_000
- assert pro["output"] == 0.87 / 1_000_000
+ assert flash["input"] == 0.22 / 1_000_000
+ assert flash["output"] == 0.66 / 1_000_000
+ assert flash["cache_read"] == 0.007 / 1_000_000
+ assert pro["input"] == 0.66 / 1_000_000
+ assert pro["output"] == 1.98 / 1_000_000
+ assert pro["cache_read"] == 0.022 / 1_000_000
+
+ flash_peak = get_pricing("deepseek-v4-flash", request_time=_PEAK)
+ pro_peak = get_pricing("deepseek-v4-pro", request_time=_PEAK)
+ assert flash_peak["input"] == 0.44 / 1_000_000
+ assert flash_peak["output"] == 1.32 / 1_000_000
+ assert flash_peak["cache_read"] == 0.014 / 1_000_000
+ assert pro_peak["input"] == 1.32 / 1_000_000
+ assert pro_peak["output"] == 3.96 / 1_000_000
+ assert pro_peak["cache_read"] == 0.044 / 1_000_000
+
+
+def test_deepseek_cache_creation_mirrors_input_on_both_cards():
+ """DeepSeek has no cache-WRITE charge — a miss is just input. The mirror
+ is what makes the generic ``compute_cost`` correct for this provider, and
+ it has to survive on the peak card too."""
+ from src.services.pricing import get_pricing
+
+ for ts in (_OFF_PEAK, _PEAK):
+ for model in ("deepseek-v4-flash", "deepseek-v4-pro"):
+ p = get_pricing(model, request_time=ts)
+ assert p["cache_creation"] == p["input"]
def test_openrouter_deepseek_pricing_via_vendor_strip():
"""Consistent with how all proxied models are priced at the upstream rate
- (get_pricing strips the ``deepseek/`` vendor prefix)."""
+ (get_pricing strips the ``deepseek/`` vendor prefix). The strip must carry
+ ``request_time`` through, or the proxied id would price off a different
+ clock reading than the bare one."""
from src.services.pricing import get_pricing
- assert get_pricing("deepseek/deepseek-v4-pro") == get_pricing("deepseek-v4-pro")
+ for ts in (_OFF_PEAK, _PEAK):
+ assert get_pricing("deepseek/deepseek-v4-pro", request_time=ts) == (
+ get_pricing("deepseek-v4-pro", request_time=ts)
+ )
def test_deepseek_cost_credits_cache_hit_end_to_end():
@@ -197,12 +244,15 @@ def test_deepseek_cost_credits_cache_hit_end_to_end():
prompt_tokens=1_000_000, completion_tokens=0, total_tokens=1_000_000,
prompt_cache_hit_tokens=900_000, prompt_cache_miss_tokens=100_000,
))
- cost = compute_cost("deepseek-v4-flash", usage)
- expected = 100_000 * 0.14 / 1_000_000 + 900_000 * 0.0028 / 1_000_000
+ cost = compute_cost("deepseek-v4-flash", usage, request_time=_OFF_PEAK)
+ expected = 100_000 * 0.22 / 1_000_000 + 900_000 * 0.007 / 1_000_000
assert abs(cost - expected) < 1e-12
# ~9x cheaper than pricing the whole prompt as uncached input.
- full = 1_000_000 * 0.14 / 1_000_000
+ full = 1_000_000 * 0.22 / 1_000_000
assert cost < full / 5
+ # The same response costs exactly twice as much inside a peak window.
+ peak = compute_cost("deepseek-v4-flash", usage, request_time=_PEAK)
+ assert abs(peak - 2 * expected) < 1e-12
def test_cost_command_surfaces_cache_hit_rate():
diff --git a/tests/test_pricing_status_bar.py b/tests/test_pricing_status_bar.py
index 8e02557b..1f147b2f 100644
--- a/tests/test_pricing_status_bar.py
+++ b/tests/test_pricing_status_bar.py
@@ -81,8 +81,12 @@ def test_openrouter_vendor_prefix_stripped(self) -> None:
self.assertEqual(p["input"], 5.0 / 1_000_000)
# The same strip prices DeepSeek-via-OpenRouter at the upstream
# DeepSeek rate (a directional estimate; the proxy may add markup).
- d = get_pricing("deepseek/deepseek-v4-pro")
- self.assertEqual(d["input"], 0.435 / 1_000_000)
+ # Time-tiered, so the instant is pinned: Monday 12:00 UTC is off-peak.
+ from datetime import datetime, timezone
+
+ off_peak = datetime(2026, 8, 24, 12, 0, tzinfo=timezone.utc).timestamp()
+ d = get_pricing("deepseek/deepseek-v4-pro", request_time=off_peak)
+ self.assertEqual(d["input"], 0.66 / 1_000_000)
def test_unknown_model_returns_none(self) -> None:
# Critic C1: unknowns return None instead of mispricing as