tzt
a0afeec5cb
- pricing.py：is_offpeak（含跨午夜窗口、非法回落、±1min 边界语义）； compute 全整数毫元——上游成本叠加 offpeak_factor（空闲半价）， 学生售价按差异化折扣（in 0.5 / out 0.8，不叠加热闲系数，全天一口价）， margin = charged - cost 恒等式 - 未配置价格模型三值零（调用方可拦截）；配置换算加载期唯一 - 测试 +9（窗口边界/跨午夜/回落/黄金手算 #1#2/恒等式 10 组扫/零用量/ 稳定性/未知模型/换算不漂移），全量 354 passed
82 lines
3.3 KiB
Python
82 lines
3.3 KiB
Python
"""计价（T-P3）：峰谷窗口 + 毫元整数计算。
|
||
|
||
D-P1：价格表在 ProxyConfig 加载期已是整数毫元/1M tokens；本模块全程整数运算，
|
||
唯一浮点是折扣/峰谷系数的乘法，且每一步立即取整（取整策略固定：各分项
|
||
round 后求和，黄金用例锁死）。
|
||
|
||
毛利恒等式：margin_milli = charged_milli - upstream_cost_milli（测试断言）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import time
|
||
from typing import Any, Dict
|
||
|
||
from gateway.proxy.ledgerutil import YUAN_TO_MILLI
|
||
|
||
|
||
def _hhmm_to_minutes(text: str, fallback: int) -> int:
|
||
parts = str(text or "").strip().split(":")
|
||
if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit():
|
||
h, m = int(parts[0]), int(parts[1])
|
||
if 0 <= h <= 23 and 0 <= m <= 59:
|
||
return h * 60 + m
|
||
return fallback
|
||
|
||
|
||
def is_offpeak(ts: float, peak_start: str = "08:30",
|
||
peak_end: str = "23:59") -> bool:
|
||
"""峰谷窗口判定（本地时区 HH:MM；窗口含头含尾：start<=t<=end 为高峰）。
|
||
|
||
支持跨午夜窗口（如 start=23:00, end=06:00）。
|
||
"""
|
||
lt = time.localtime(ts)
|
||
cur = lt.tm_hour * 60 + lt.tm_min
|
||
s = _hhmm_to_minutes(peak_start, 8 * 60 + 30)
|
||
e = _hhmm_to_minutes(peak_end, 23 * 60 + 59)
|
||
if s <= e:
|
||
return not (s <= cur <= e)
|
||
# 跨午夜：高峰 = [s, 24h) ∪ [0, e]
|
||
return not (cur >= s or cur <= e)
|
||
|
||
|
||
def _apply_factor(milli: int, factor: float) -> int:
|
||
return int(round(milli * factor))
|
||
|
||
|
||
def compute(usage: Dict[str, int], model: str, ts: float, cfg) -> Dict[str, int]:
|
||
"""按 usage 计费（§6 签名）-> CostBreakdown（全整数毫元）。
|
||
|
||
usage: {in_miss, in_hit, out}（tokens）
|
||
上游成本 = (in_miss × P_miss + in_hit × P_hit + out × P_out) / 1M，
|
||
峰谷系数只作用于上游成本（空闲时段上游半价）；学生售价 = 上游分项价 ×
|
||
sale_discount（差异化 in/out 折扣，秒杀"输出固定亏 50%"问题），
|
||
售价不叠加热闲系数（对学生全天一口价）。
|
||
"""
|
||
price = cfg.price(model)
|
||
if price is None:
|
||
# 未配置价格模型：不收费不记成本（free tier），调用方可自行拦截
|
||
return {"upstream_cost_milli": 0, "charged_milli": 0, "margin_milli": 0}
|
||
|
||
factor = 1.0 if not is_offpeak(ts, cfg.peak_start, cfg.peak_end) else cfg.offpeak_factor
|
||
|
||
# ---- 上游成本（毫元，每 1M tokens = 1e6）----
|
||
in_miss_cost = usage.get("in_miss", 0) * price.in_miss / 1_000_000
|
||
in_hit_cost = usage.get("in_hit", 0) * price.in_hit / 1_000_000
|
||
out_cost = usage.get("out", 0) * price.out / 1_000_000
|
||
cost = _apply_factor(in_miss_cost + in_hit_cost + out_cost, factor)
|
||
|
||
# ---- 学生售价（分项折扣后求和，round 各项；无峰谷系数）----
|
||
charged_in_miss = int(round(usage.get("in_miss", 0) * price.in_miss
|
||
* cfg.sale_in / 1_000_000))
|
||
charged_in_hit = int(round(usage.get("in_hit", 0) * price.in_hit
|
||
* cfg.sale_in / 1_000_000))
|
||
charged_out = int(round(usage.get("out", 0) * price.out
|
||
* cfg.sale_out / 1_000_000))
|
||
charged = charged_in_miss + charged_in_hit + charged_out
|
||
|
||
return {
|
||
"upstream_cost_milli": int(cost),
|
||
"charged_milli": int(charged),
|
||
"margin_milli": int(charged) - int(cost),
|
||
}
|