diff --git a/backend/app/services/depth_service.py b/backend/app/services/depth_service.py index debe045..c239530 100644 --- a/backend/app/services/depth_service.py +++ b/backend/app/services/depth_service.py @@ -33,7 +33,12 @@ from pathlib import Path import polars as pl from app.tickflow.capabilities import Cap -from app.tickflow.rate_limits import chunked, resolve_limit, sleep_between_batches +from app.tickflow.rate_limits import ( + apply_safety_rpm, + chunked, + resolve_limit, + sleep_between_batches, +) logger = logging.getLogger(__name__) @@ -47,7 +52,6 @@ TIER_INTERVAL_RANGE: dict[str, tuple[float, float]] = { DEFAULT_RANGE = (10.0, 120.0) # 限速余量: 只用 rpm 的 80%, 给系统其他 depth 调用留空间 -RPM_MARGIN = 0.8 # 间隔硬下限/上限(任何套餐) INTERVAL_HARD_MIN = 10.0 INTERVAL_HARD_MAX = 300.0 @@ -511,9 +515,9 @@ class DepthService: raw_user = preferences.get_depth_polling_interval() user_interval = max(lo, min(hi, raw_user)) - # ② 限速安全 clamp + # ② 限速安全 clamp(与 resolve_limit 共用 SAFETY_RPM_FACTOR,不叠乘) batches = max(1, math.ceil(n_symbols / batch_size)) - usable_rpm = rpm * RPM_MARGIN + usable_rpm = apply_safety_rpm(rpm) or 1 calls_per_min = usable_rpm / batches if batches > 0 else usable_rpm safe_interval = 60.0 / calls_per_min if calls_per_min > 0 else INTERVAL_HARD_MAX diff --git a/backend/app/tickflow/rate_limits.py b/backend/app/tickflow/rate_limits.py index f68546e..e6d29af 100644 --- a/backend/app/tickflow/rate_limits.py +++ b/backend/app/tickflow/rate_limits.py @@ -14,6 +14,12 @@ from app.tickflow.capabilities import Cap, CapabilitySet T = TypeVar("T") +# TickFlow Pro 单进程安全预算: 套餐标称 rpm 的 80%。 +# 注意: _next_slot 仅在本 Python 进程内共享, 不能跨 Gold 容器与 A 股面板进程。 +# Stage A 期间跨产品靠错峰(盘中 Gold 优先, A 股 Pro 批处理建议 16:00 后), +# 不要把「各进程各扣 80%」误当成账户级共享限频。 +SAFETY_RPM_FACTOR = 0.8 + # 进程级共享限速器: 原先每个调用方各自本地 sleep(60/rpm), 并发同步 (kline/index/ # depth/watchlist/custom) 时聚合请求速率会成倍超过单能力 rpm → 429。 # 这里用一张按 rpm 分桶的「下一个可用时刻」表 (Lock 守护), 所有调用方按同一时间轴 @@ -23,6 +29,13 @@ _slot_lock = threading.Lock() _next_slot: dict[int, float] = {} +def apply_safety_rpm(rpm: int | None, *, factor: float = SAFETY_RPM_FACTOR) -> int | None: + """Scale a package rpm by the shared safety factor (default 80%).""" + if rpm is None or rpm <= 0: + return rpm + return max(1, int(rpm * factor)) + + def _reserve_slot(rpm: int, interval: float) -> float: """在共享时间轴上为一次请求预约一个发包槽, 返回需等待的秒数 (>=0)。 @@ -50,14 +63,26 @@ def resolve_limit( default_batch: int | None = None, default_rpm: int | None = None, default_rpm_when_unset: bool = True, + apply_safety: bool = True, ) -> ResolvedLimit: - """Return a capability's batch/rpm with caller-provided fallbacks.""" + """Return a capability's batch/rpm with caller-provided fallbacks. + + By default rpm is scaled by SAFETY_RPM_FACTOR (0.8) inside this process only. + This is not a cross-container account budget. Pass apply_safety=False for diagnostics. + """ lim = capset.limits(cap) if lim is None: - return ResolvedLimit(batch=default_batch, rpm=default_rpm) + rpm = default_rpm + else: + rpm = lim.rpm if lim.rpm else (default_rpm if default_rpm_when_unset else None) + default_batch = lim.batch if lim.batch else default_batch + if apply_safety: + rpm = apply_safety_rpm(rpm) + if lim is None: + return ResolvedLimit(batch=default_batch, rpm=rpm) return ResolvedLimit( - batch=lim.batch if lim.batch else default_batch, - rpm=lim.rpm if lim.rpm else (default_rpm if default_rpm_when_unset else None), + batch=default_batch, + rpm=rpm, ) diff --git a/backend/tests/test_rate_limit_safety_budget.py b/backend/tests/test_rate_limit_safety_budget.py new file mode 100644 index 0000000..24b27a0 --- /dev/null +++ b/backend/tests/test_rate_limit_safety_budget.py @@ -0,0 +1,55 @@ +"""TickFlow Pro 80% rpm safety budget (process-local).""" + +from __future__ import annotations + +import time + +from app.tickflow.capabilities import Cap, CapabilityLimits, CapabilitySet +from app.tickflow.rate_limits import ( + SAFETY_RPM_FACTOR, + _next_slot, + _slot_lock, + apply_safety_rpm, + resolve_limit, + sleep_between_batches, +) + + +def test_apply_safety_rpm_scales_to_80_percent() -> None: + assert SAFETY_RPM_FACTOR == 0.8 + assert apply_safety_rpm(120) == 96 + assert apply_safety_rpm(60) == 48 + assert apply_safety_rpm(30) == 24 + assert apply_safety_rpm(None) is None + + +def test_resolve_limit_applies_safety_by_default() -> None: + capset = CapabilitySet({Cap.KLINE_DAILY_BATCH: CapabilityLimits(rpm=60, batch=100)}) + lim = resolve_limit(capset, Cap.KLINE_DAILY_BATCH) + assert lim.rpm == 48 + assert lim.batch == 100 + + +def test_resolve_limit_can_skip_safety_for_diagnostics() -> None: + capset = CapabilitySet({Cap.KLINE_DAILY_BATCH: CapabilityLimits(rpm=60, batch=100)}) + lim = resolve_limit(capset, Cap.KLINE_DAILY_BATCH, apply_safety=False) + assert lim.rpm == 60 + + +def test_two_first_batches_do_not_wait_even_when_slot_reserved() -> None: + """Documents current behavior called out in Codex critique (first-batch burst).""" + with _slot_lock: + _next_slot.clear() + t0 = time.perf_counter() + sleep_between_batches(0, rpm=60) + sleep_between_batches(0, rpm=60) + assert time.perf_counter() - t0 < 0.05 + + +def test_second_batch_waits_for_shared_slot() -> None: + with _slot_lock: + _next_slot.clear() + t0 = time.perf_counter() + sleep_between_batches(0, rpm=60) + sleep_between_batches(1, rpm=60) + assert time.perf_counter() - t0 >= 0.9 diff --git a/docs/tickflow-pro-shared-rate-limit.md b/docs/tickflow-pro-shared-rate-limit.md new file mode 100644 index 0000000..1f3b21d --- /dev/null +++ b/docs/tickflow-pro-shared-rate-limit.md @@ -0,0 +1,26 @@ +# TickFlow Pro 共享限频(Gold Shadow + A 股研究) + +## 决策 +- 同一 TickFlow API Key 的所有调用方必须共享进程级限频器:`backend/app/tickflow/rate_limits.py`。 +- 项目安全预算 = 套餐额度 × **80%**(见 `tiers.yaml` 的 `pro` 段)。 +- Gold Shadow Stage A 与 A 股同步/回测**不得**各建一套限频,避免叠加超限。 + +## 预算表示例(Pro) +| capability | 套餐 rpm | 80% 预算 | +|---|---:|---:| +| quote.batch | 120 | 96 | +| quote.pool | 60 | 48 | +| kline.daily.batch | 60 | 48 | +| kline.minute.batch | 30 | 24 | +| depth5.batch | 30 | 24 | +| adj_factor | 60 | 48 | + +## 错峰 +- 盘中(A 股交易时段):优先保证 Gold Shadow 观察与 legacy 监控。 +- 盘后:A 股日线/研究批量、分钟按需回填。 +- 禁止全市场一年分钟一次性回填。 + +## 实现要点 +- 调用方只传 `rpm`/`batch`,统一走 `sleep_between_batches` / `_reserve_slot`。 +- 应用 80% 预算:在 resolve 后对 rpm 做 `floor(rpm * 0.8)`(后续 Phase 实现,Phase 0 仅冻结约定)。 +- 任一 fallback / 429 必须写入 DatasetEvidence,禁止静默混源。