fix: apply process-local 80% TickFlow rpm safety budget

Scale resolve_limit rpm by SAFETY_RPM_FACTOR and document that the
in-process slot limiter is not a cross-container account budget.
Keep Stage A isolation via off-peak A-share usage guidance.

Co-authored-by: Cursor <cursoragent@cursor.com>

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
 

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,
     )
 
 

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

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，禁止静默混源。