Merge pull request #133 from dunmin1980-ux/cursor/tickflow-pro-rate-limit-p0

TickFlow Pro: process-local 80% RPM safety + Phase 1 probe scaffold
This commit is contained in:
wshy
2026-07-31 19:08:11 +08:00
committed by GitHub
9 changed files with 524 additions and 17 deletions
+3
View File
@@ -92,6 +92,8 @@ data/strategies/custom/
backend._recheck*.py
backend/_verify*.py
backend/_desktop_run.log
reports/tickflow_pro_probe/
.cursor/debug-*.log
# ===== 打包产物 (本地构建的安装包,不入库) =====
backend/TickFlowStockPanel-win-x64.zip
@@ -106,3 +108,4 @@ packaging/installed_run.log
TickFlowStockPanel.app/
TickFlowStockPanel-macos*.dmg
TickFlowStockPanel-macos*.zip
reports/cursor_tickflow_pro_rate_limit_p0.bundle
+8 -4
View File
@@ -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
+41 -13
View File
@@ -14,15 +14,27 @@ from app.tickflow.capabilities import Cap, CapabilitySet
T = TypeVar("T")
# 进程级共享限速器: 原先每个调用方各自本地 sleep(60/rpm), 并发同步 (kline/index/
# depth/watchlist/custom) 时聚合请求速率会成倍超过单能力 rpm → 429
# 这里用一张按 rpm 分桶的「下一个可用时刻」表 (Lock 守护), 所有调用方按同一时间轴
# 排队, 使跨调用方的聚合发包间隔 >= 60/rpm。以 rpm 为键 (调用方签名只带 rpm, 不带 cap;
# rpm 是各能力速率的代理); 恰好同 rpm 的不同能力会共享一队, 偏保守但绝不超速。
# TickFlow Pro 单进程安全预算: 套餐标称 rpm 的 80%。
# 注意: _next_slot 仅在本 Python 进程内共享, 不能跨 Gold 容器与 A 股面板进程
# Stage A 期间跨产品靠错峰(盘中 Gold 优先, A 股 Pro 批处理建议 16:00 后),
# 不要把「各进程各扣 80%」误当成账户级共享限频。
SAFETY_RPM_FACTOR = 0.8
# 进程级共享限速器: 按 rpm 分桶的「下一个可用时刻」表 (Lock 守护)。
# 限制: sleep_between_batches(index=0) 只登记槽位、不 sleep, 因此多个调用方若同时以
# index=0 启动, 仍可能瞬时突发超过单能力 rpm。后续 index>0 批次会按同一时间轴排队。
# Phase 1 / Stage A: 不要并发启动多个 probe 或大批量 sync; 跨进程仍靠错峰, 非账户级限频。
_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 +62,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,
)
@@ -74,16 +98,20 @@ def chunked(items: list[T], batch_size: int | None) -> list[list[T]]:
def sleep_between_batches(index: int, rpm: int | None, *, default_interval: float = 0.0) -> None:
"""Sleep before every batch after the first, using the existing interval formula.
"""Pace batches via the process-local shared slot table.
内部改用进程级共享限速器 (_reserve_slot): 保持「首批不 sleep, 后续每批间隔 60/rpm」
的单调用方观感, 同时让并发调用方按同一时间轴排队, 聚合速率不再超过单能力 rpm。
- index == 0: reserve a slot but do **not** sleep (documented first-batch burst).
- index > 0: wait until the reserved slot time.
Concurrent callers that all pass index=0 can still burst above rpm; only later
batches and single-pipeline callers get full spacing. Do not launch multiple
Phase 1 probes/syncs at once during Stage A.
"""
interval = batch_interval(rpm, default=default_interval)
if interval <= 0:
return
if index <= 0:
# 首批不 sleep, 但登记一个占位槽, 让后续/并发调用方在同一时间轴上排队
# First batch: reserve only (no sleep). Concurrent index=0 calls may burst.
_reserve_slot(rpm or -1, interval)
return
wait = _reserve_slot(rpm or -1, interval)
+238
View File
@@ -0,0 +1,238 @@
#!/usr/bin/env python3
"""TickFlow Pro Phase 1 probe (sanitized).
Default is dry-run (no network). Live probe requires TICKFLOW_API_KEY and
should run after 16:00 Asia/Shanghai unless --force is set (Stage A off-peak).
Never prints API keys. Writes reports under reports/tickflow_pro_probe/.
"""
from __future__ import annotations
import argparse
import json
import os
import sys
from dataclasses import asdict, dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Protocol
from zoneinfo import ZoneInfo
SHANGHAI = ZoneInfo("Asia/Shanghai")
GOLDEN = ("000403.SZ", "600489.SH", "300059.SZ")
OFFPEAK_HOUR = 16
@dataclass
class ProbeGate:
dry_run: bool
force: bool
now_hour: int
has_key: bool
allowed: bool
reason: str
class _KlinesNS(Protocol):
def batch(self, *args: Any, **kwargs: Any) -> Any: ...
class _QuotesNS(Protocol):
def get(self, *args: Any, **kwargs: Any) -> Any: ...
class TickFlowLike(Protocol):
klines: _KlinesNS
quotes: _QuotesNS
def evaluate_gate(*, dry_run: bool, force: bool, has_key: bool, now: datetime | None = None) -> ProbeGate:
now = now or datetime.now(SHANGHAI)
hour = now.hour
if dry_run:
return ProbeGate(True, force, hour, has_key, True, "dry_run")
if not has_key:
return ProbeGate(False, force, hour, False, False, "missing_TICKFLOW_API_KEY")
if hour < OFFPEAK_HOUR and not force:
return ProbeGate(False, force, hour, True, False, f"before_{OFFPEAK_HOUR:02d}00_use_--force_or_wait")
return ProbeGate(False, force, hour, True, True, "live_ok")
def _repo_root() -> Path:
return Path(__file__).resolve().parents[2]
def _sanitize_sample(obj: object) -> object:
"""Keep structure; drop long arrays and obvious secret-like strings."""
if isinstance(obj, dict):
out = {}
for k, v in obj.items():
lk = str(k).lower()
if any(s in lk for s in ("key", "token", "secret", "password", "authorization")):
out[k] = "***"
else:
out[k] = _sanitize_sample(v)
return out
if isinstance(obj, list):
if len(obj) > 3:
return [_sanitize_sample(x) for x in obj[:3]] + [f"...(+{len(obj) - 3} more)"]
return [_sanitize_sample(x) for x in obj]
if isinstance(obj, str) and len(obj) > 120:
return obj[:120] + "..."
return obj
def _to_sample(raw: object) -> object:
if hasattr(raw, "to_dict"):
try:
return _sanitize_sample(raw.to_dict()) # type: ignore[attr-defined]
except Exception:
pass
return _sanitize_sample(raw)
def run_dry_run(out_dir: Path) -> dict:
out_dir.mkdir(parents=True, exist_ok=True)
report = {
"mode": "dry_run",
"as_of": datetime.now(SHANGHAI).isoformat(timespec="seconds"),
"symbols": list(GOLDEN),
"planned_checks": [
"quotes.get",
"klines.batch(period=1d)",
"klines.batch(period=1m)",
"klines.ex_factors",
"rate_limit_observation",
],
"safety": {
"SAFETY_RPM_FACTOR": 0.8,
"offpeak_hour": OFFPEAK_HOUR,
"note": (
"Dry-run only validates CLI gates/plan. "
"Live probe must not overlap Gold Stage A peak window without --force. "
"Do not start multiple Phase 1 probe/sync processes concurrently."
),
},
# Gate-only success; does not prove SDK methods, network, or auth.
"status": "DRY_RUN_OK",
}
(out_dir / "probe_summary.json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n")
(out_dir / "probe_summary.md").write_text(
"# TickFlow Pro probe (dry-run)\n\n"
f"- as_of: {report['as_of']}\n"
f"- symbols: {', '.join(GOLDEN)}\n"
"- status: DRY_RUN_OK (gates/plan only; not a live contract proof)\n"
f"- live command: `TICKFLOW_API_KEY=... python scripts/probe_tickflow_pro.py --live` "
f"(after {OFFPEAK_HOUR}:00 or with --force)\n"
)
return report
def run_live(
out_dir: Path,
*,
client: TickFlowLike | None = None,
endpoint: str | None = None,
) -> dict:
"""Minimal live smoke: daily klines.batch + quotes.get for golden symbols.
Pass ``client`` in tests (fake SDK). Production path builds TickFlow from env key.
"""
out_dir.mkdir(parents=True, exist_ok=True)
from app.tickflow.rate_limits import SAFETY_RPM_FACTOR, apply_safety_rpm
if client is not None:
# Test/injection path: do not import app.tickflow.client (pulls tickflow SDK).
base = endpoint or "https://api.tickflow.org"
tf = client
else:
key = os.environ.get("TICKFLOW_API_KEY", "").strip()
if not key:
try:
from app.secrets_store import get_tickflow_key # type: ignore
key = (get_tickflow_key() or "").strip()
except Exception:
key = ""
if not key:
raise SystemExit("missing TICKFLOW_API_KEY")
from tickflow import TickFlow
from app.tickflow.client import PAID_ENDPOINT, _base_url
base = endpoint or (_base_url() or PAID_ENDPOINT)
tf = TickFlow(api_key=key, base_url=base)
samples: dict[str, object] = {}
errors: list[str] = []
try:
raw = tf.klines.batch(list(GOLDEN), period="1d", count=5, as_dataframe=False)
samples["klines.batch.1d"] = _to_sample(raw)
except Exception as exc: # noqa: BLE001
errors.append(f"klines.batch.1d: {type(exc).__name__}")
try:
raw = tf.quotes.get(symbols=list(GOLDEN), as_dataframe=False)
samples["quotes.get"] = _to_sample(raw)
except Exception as exc: # noqa: BLE001
errors.append(f"quotes.get: {type(exc).__name__}")
report = {
"mode": "live",
"as_of": datetime.now(SHANGHAI).isoformat(timespec="seconds"),
"endpoint": base,
"symbols": list(GOLDEN),
"safety_rpm_factor": SAFETY_RPM_FACTOR,
"example_budget_daily_batch_rpm": apply_safety_rpm(60),
"samples": samples,
"errors": errors,
"status": "LIVE_PARTIAL" if errors else "LIVE_OK",
}
(out_dir / "probe_summary.json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n")
(out_dir / "probe_summary.md").write_text(
"# TickFlow Pro probe (live)\n\n"
f"- as_of: {report['as_of']}\n"
f"- status: {report['status']}\n"
f"- errors: {len(errors)}\n"
"- samples sanitized; no API key written.\n"
)
return report
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--live", action="store_true", help="Call TickFlow APIs (needs key)")
parser.add_argument("--force", action="store_true", help="Allow live probe before 16:00 Asia/Shanghai")
parser.add_argument(
"--out",
type=Path,
default=None,
help="Output directory (default: <repo>/reports/tickflow_pro_probe/<timestamp>)",
)
args = parser.parse_args(argv)
dry_run = not args.live
has_key = bool(os.environ.get("TICKFLOW_API_KEY", "").strip())
gate = evaluate_gate(dry_run=dry_run, force=args.force, has_key=has_key)
print(json.dumps(asdict(gate), ensure_ascii=False))
if not gate.allowed:
return 2
stamp = datetime.now(SHANGHAI).strftime("%Y%m%d_%H%M%S")
out_dir = args.out or (_repo_root() / "reports" / "tickflow_pro_probe" / stamp)
if dry_run:
report = run_dry_run(out_dir)
else:
root = _repo_root() / "backend"
if str(root) not in sys.path:
sys.path.insert(0, str(root))
report = run_live(out_dir)
print(json.dumps({"out": str(out_dir), "status": report.get("status")}, ensure_ascii=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,41 @@
"""Gates for TickFlow Pro Phase 1 probe CLI."""
from __future__ import annotations
from datetime import datetime
from zoneinfo import ZoneInfo
from scripts.probe_tickflow_pro import evaluate_gate
SH = ZoneInfo("Asia/Shanghai")
def test_dry_run_always_allowed_without_key() -> None:
gate = evaluate_gate(dry_run=True, force=False, has_key=False)
assert gate.allowed is True
assert gate.reason == "dry_run"
def test_live_blocked_without_key() -> None:
gate = evaluate_gate(dry_run=False, force=True, has_key=False)
assert gate.allowed is False
assert "missing" in gate.reason
def test_live_blocked_before_offpeak_without_force() -> None:
now = datetime(2026, 7, 19, 10, 0, tzinfo=SH)
gate = evaluate_gate(dry_run=False, force=False, has_key=True, now=now)
assert gate.allowed is False
assert "before_16" in gate.reason
def test_live_allowed_after_offpeak() -> None:
now = datetime(2026, 7, 19, 16, 5, tzinfo=SH)
gate = evaluate_gate(dry_run=False, force=False, has_key=True, now=now)
assert gate.allowed is True
def test_live_allowed_before_offpeak_with_force() -> None:
now = datetime(2026, 7, 19, 10, 0, tzinfo=SH)
gate = evaluate_gate(dry_run=False, force=True, has_key=True, now=now)
assert gate.allowed is True
@@ -0,0 +1,69 @@
"""Fake-client coverage for TickFlow Pro live probe SDK namespaces."""
from __future__ import annotations
from pathlib import Path
from types import SimpleNamespace
from scripts.probe_tickflow_pro import run_dry_run, run_live
class _FakeKlines:
def __init__(self) -> None:
self.calls: list[dict] = []
def batch(self, symbols, **kwargs):
self.calls.append({"symbols": list(symbols), **kwargs})
return [{"symbol": s, "bars": [{"c": 1.0}]} for s in symbols]
class _FakeQuotes:
def __init__(self) -> None:
self.calls: list[dict] = []
def get(self, *, symbols, **kwargs):
self.calls.append({"symbols": list(symbols), **kwargs})
return [{"symbol": s, "last": 1.0} for s in symbols]
class _FakeTickFlow:
def __init__(self) -> None:
self.klines = _FakeKlines()
self.quotes = _FakeQuotes()
def test_dry_run_status_is_dry_run_ok(tmp_path: Path) -> None:
report = run_dry_run(tmp_path)
assert report["status"] == "DRY_RUN_OK"
assert "READY_FOR_LIVE" not in report["status"]
def test_run_live_uses_klines_batch_and_quotes_get(tmp_path: Path) -> None:
fake = _FakeTickFlow()
report = run_live(tmp_path, client=fake, endpoint="https://api.tickflow.org")
assert report["status"] == "LIVE_OK"
assert report["errors"] == []
assert "klines.batch.1d" in report["samples"]
assert "quotes.get" in report["samples"]
assert fake.klines.calls and fake.klines.calls[0]["period"] == "1d"
assert fake.klines.calls[0]["count"] == 5
assert fake.quotes.calls and set(fake.quotes.calls[0]["symbols"]) == {
"000403.SZ",
"600489.SH",
"300059.SZ",
}
def test_run_live_records_partial_on_sdk_errors(tmp_path: Path) -> None:
class _Bad:
def batch(self, *a, **k):
raise AttributeError("nope")
def get(self, *a, **k):
raise AttributeError("nope")
fake = SimpleNamespace(klines=_Bad(), quotes=_Bad())
report = run_live(tmp_path, client=fake, endpoint="https://api.tickflow.org")
assert report["status"] == "LIVE_PARTIAL"
assert any(e.startswith("klines.batch.1d: AttributeError") for e in report["errors"])
assert any(e.startswith("quotes.get: AttributeError") for e in report["errors"])
@@ -0,0 +1,59 @@
"""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:
"""Known limitation: concurrent/serial index=0 calls do not sleep (first-batch burst).
Comments document that aggregate rpm is therefore not guaranteed when many
pipelines start with index=0; Phase 1 must avoid concurrent probe/sync.
"""
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
+37
View File
@@ -0,0 +1,37 @@
# TickFlow Pro Phase 1 probe
Process-local rate-limit safety (80% RPM) is already on branch `cursor/tickflow-pro-rate-limit-p0`.
## Dry-run (safe anytime)
```bash
cd backend
python3 scripts/probe_tickflow_pro.py
```
Writes sanitized plan under `reports/tickflow_pro_probe/<timestamp>/` with status `DRY_RUN_OK`
(gates/plan only — not proof of SDK methods, network, or auth).
## Live probe (off-peak)
Uses real SDK namespaces aligned with the app: `tf.klines.batch` and `tf.quotes.get`.
Do not start multiple Phase 1 probe/sync processes concurrently.
Requirements:
- `TICKFLOW_API_KEY` in the environment (never commit the key)
- Prefer after **16:00 Asia/Shanghai** (Stage A off-peak vs Gold)
- Before 16:00 only with `--force`
```bash
cd backend
TICKFLOW_API_KEY=... python3 scripts/probe_tickflow_pro.py --live
```
Golden symbols: `000403.SZ`, `600489.SH`, `300059.SZ`.
## Out of scope
- Full-market backfill
- Gold Shadow deploy / Telegram ownership
- Claiming cross-product shared rate limits (still process-local)
+28
View File
@@ -0,0 +1,28 @@
# TickFlow Pro 限频安全预算(进程内 + 跨产品错峰)
## 决策(经 ChatGPT/Codex 评析修订)
- `backend/app/tickflow/rate_limits.py` 提供**单 Python 进程内**的 rpm 槽位限速与 `SAFETY_RPM_FACTOR=0.8`
- **不要**把「各进程各扣 80%」当成账户级共享限频:Gold Shadow 容器与 A 股面板进程状态独立,理论聚合可达 160%。
- Stage A 期间跨产品靠**错峰**,不在 Gold 上部署分布式限频重构。
## 预算表示例(Pro,单进程 80%)
| 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 |
## 错峰(Stage A
- 盘中:优先 Gold Shadow 观察与 legacy `gold-monitor`
- A 股 Pro 探测/大批量同步:建议 **16:00 后**
- 禁止全市场一年分钟一次性回填。
## 实现要点
- `resolve_limit(..., apply_safety=True)` 默认对 rpm 做 `floor(rpm * 0.8)`
- `sleep_between_batches``index=0` 只占槽不 sleep(首批突发;**并发多个 index=0 仍可能超 rpm**);后续 batch 按槽位等待。
- Phase 1 / Stage A**不要并发**启动多个 probe 或大批量 sync。
- 诊断可传 `apply_safety=False`;跨容器账户预算需另设(Stage A 不做)。
- 任一 429 / fallback 应记入证据链,禁止静默混源。