xsl
c97f8ae3e1
CI / lint / typecheck / import-linter / tests / secrets (push) Canceled after 0s
ob_ai_call is the only input to V2 pricing, and it was dropping the two numbers needed to interpret its own cost column. tokens_reasoning was captured from the provider response but never written to the table, so the cost of thinking mode was invisible once it left the process. Folding it into tokens_out was not an option either -- it is billed at the output rate, and mixing the two hides exactly the difference we now have to watch. rate_tier records which price list applied. Without it a row's cost_cents cannot be rechecked: DeepSeek's peak rate is double off-peak from 2026-08-17, so recomputing an old row at "the current rate" silently produces a different answer than what was actually billed. Migration 0002 adds both columns. tokens_reasoning defaults to 0 rather than NULL so cost aggregates do not collapse to NULL on historical rows; rate_tier stays nullable because rows written before this change genuinely have an unknown tier, and NULL says that more honestly than a guessed 'flat'. The smoke script now asserts both: reasoning tokens must be 0 (a non-zero value means a provider default re-enabled thinking mode behind our backs) and rate_tier must be present. Verified against the real API -- tokens_reasoning=0, rate_tier=off_peak, 3.9s round trip.
102 lines
3.2 KiB
Python
102 lines
3.2 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 (
|
|
TIER_FLAT,
|
|
TIER_OFF_PEAK,
|
|
TIER_PEAK,
|
|
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
|
|
|
|
|
|
def test_tier_is_reported_so_cost_can_be_rechecked_later() -> None:
|
|
# 没有档位，事后就无法判断某行 cost_cents 是按哪套单价算的
|
|
table = _table()
|
|
assert table.current_tier(datetime(2026, 8, 20, 10, 0, tzinfo=BEIJING)) == TIER_PEAK
|
|
assert table.current_tier(datetime(2026, 8, 20, 13, 0, tzinfo=BEIJING)) == TIER_OFF_PEAK
|
|
|
|
|
|
def test_single_rate_tables_report_a_flat_tier() -> None:
|
|
assert RateTable({MODEL: OFF_PEAK}, default=OFF_PEAK).current_tier() == TIER_FLAT
|
|
assert _table(enabled=False).current_tier() == TIER_FLAT
|