xsl
f46ecc5eb0
Two defects that only surface against the real DeepSeek API, both of which break the cost model rather than the code: deepseek-v4-flash enables thinking mode by default. The same single-turn question took 13.5s with it on and 1.8s with it off, and the thinking tokens are billed at the output rate. That misses NFR-PERF-02b (15s per turn) and inflates every conversational call. SceneConfig now carries an explicit `thinking` flag, all four V1 scenes set it to false, and the provider always sends the parameter rather than relying on the upstream default. Reasoning tokens are recorded separately in Usage.tokens_reasoning so the cost of ever turning it back on is visible. DeepSeek moves to peak/off-peak pricing on 2026-08-17, with peak (Beijing 09:00-12:00 and 14:00-18:00) at twice the off-peak rate. Reports are generated mostly during peak hours, so a single blended rate would leave ob_ai_call -- the only input to V2 pricing -- off by nearly a factor of two. PeakAwareRateTable resolves the rate at call time; cost is computed and stored then, never recomputed later at whatever the current rate happens to be. Also here: - Replace the live API key in the redaction test with a synthetic one. The test that proves we redact secrets was itself a secret leak waiting for its first commit. - Correct LLM_COST_RATES_JSON, which held placeholder rates. - Make ids.SnowflakeGenerator take an injectable clock so rollover and clock-rollback are testable without patching a slots class. - Rename exceptions to the -Error suffix (N818) and clear the remaining ruff, mypy and import-linter findings, including two dead lines: a duplicate pytestmark assignment and a no-op assert on a None return. Gates: ruff, mypy (96 files incl. tests), 4/4 import contracts, detect-secrets, 166 unit + 3 integration tests, migration up/down round-trip.
87 lines
2.6 KiB
Python
87 lines
2.6 KiB
Python
"""峰谷定价（DeepSeek 自 2026-08-17 起高峰单价翻倍）。
|
|
|
|
保护的是：同一次调用在高峰与空闲时段必须记出不同的成本。若只配一套单价，
|
|
``ob_ai_call`` 会与真实账单差出接近 2 倍，而这张表是 V2 定价的唯一依据。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta, timezone
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
|
|
from packages.llm.cost import (
|
|
ModelRate,
|
|
PeakAwareRateTable,
|
|
RateTable,
|
|
compute_cost_cents,
|
|
is_peak,
|
|
)
|
|
from packages.llm.types import Usage
|
|
|
|
BEIJING = timezone(timedelta(hours=8))
|
|
|
|
OFF_PEAK = ModelRate(Decimal("150"), Decimal("5"), Decimal("450"))
|
|
PEAK = ModelRate(Decimal("300"), Decimal("10"), Decimal("900"))
|
|
MODEL = "deepseek-v4-flash"
|
|
|
|
|
|
def _table(*, enabled: bool = True) -> PeakAwareRateTable:
|
|
return PeakAwareRateTable(
|
|
RateTable({MODEL: OFF_PEAK}, default=OFF_PEAK),
|
|
RateTable({MODEL: PEAK}, default=PEAK),
|
|
enabled=enabled,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("hour", "expected"),
|
|
[
|
|
(0, False),
|
|
(8, False),
|
|
(9, True), # 高峰左闭
|
|
(11, True),
|
|
(12, False), # 高峰右开
|
|
(13, False),
|
|
(14, True),
|
|
(17, True),
|
|
(18, False),
|
|
(23, False),
|
|
],
|
|
)
|
|
def test_peak_windows_follow_the_published_schedule(hour: int, expected: bool) -> None:
|
|
assert is_peak(datetime(2026, 8, 20, hour, 30, tzinfo=BEIJING)) is expected
|
|
|
|
|
|
def test_utc_input_is_converted_to_beijing_before_judging() -> None:
|
|
# 02:00 UTC = 10:00 北京时间，属高峰。按 UTC 判定会把高峰算成空闲、成本少记一半
|
|
assert is_peak(datetime(2026, 8, 20, 2, 0, tzinfo=UTC)) is True
|
|
|
|
|
|
def test_peak_rate_is_double_the_off_peak_rate() -> None:
|
|
table = _table()
|
|
usage = Usage(tokens_in=10_000, tokens_out=2_000, tokens_cached=0)
|
|
|
|
off_peak_cost = compute_cost_cents(
|
|
usage, table.for_model(MODEL, datetime(2026, 8, 20, 13, 0, tzinfo=BEIJING))
|
|
)
|
|
peak_cost = compute_cost_cents(
|
|
usage, table.for_model(MODEL, datetime(2026, 8, 20, 10, 0, tzinfo=BEIJING))
|
|
)
|
|
|
|
assert peak_cost == off_peak_cost * 2
|
|
|
|
|
|
def test_disabling_peak_pricing_keeps_one_rate() -> None:
|
|
# 供应商若取消峰谷定价，改一个开关即可，不必回滚代码
|
|
table = _table(enabled=False)
|
|
assert table.for_model(MODEL, datetime(2026, 8, 20, 10, 0, tzinfo=BEIJING)) is OFF_PEAK
|
|
|
|
|
|
def test_unknown_model_still_costs_money_in_both_tiers() -> None:
|
|
table = _table()
|
|
for hour in (10, 13):
|
|
rate = table.for_model("some-new-model", datetime(2026, 8, 20, hour, tzinfo=BEIJING))
|
|
assert rate.input_cents_per_mtok > 0
|