From 0745fdbb7d4235de4bc9e4805732daee78833b91 Mon Sep 17 00:00:00 2001
From: dunmin1980 <dunmin1980@gmail.com>
Date: Sun, 19 Jul 2026 08:50:10 +0800
Subject: [PATCH 1/6] 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         | 12 ++--
 backend/app/tickflow/rate_limits.py           | 33 +++++++++--
 .../tests/test_rate_limit_safety_budget.py    | 55 +++++++++++++++++++
 docs/tickflow-pro-shared-rate-limit.md        | 26 +++++++++
 4 files changed, 118 insertions(+), 8 deletions(-)
 create mode 100644 backend/tests/test_rate_limit_safety_budget.py
 create mode 100644 docs/tickflow-pro-shared-rate-limit.md

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

From f65aaa88f6d474eda5ac6b390267160f0f18095c Mon Sep 17 00:00:00 2001
From: dunmin1980 <dunmin1980@gmail.com>
Date: Sun, 19 Jul 2026 08:50:26 +0800
Subject: [PATCH 2/6] docs: clarify rate limit is process-local, not
 cross-product

Align the Pro rate-limit note with the critique: 80% applies inside
one process; Stage A cross-product safety relies on off-peak scheduling.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 docs/tickflow-pro-shared-rate-limit.md | 27 +++++++++++++-------------
 1 file changed, 14 insertions(+), 13 deletions(-)

diff --git a/docs/tickflow-pro-shared-rate-limit.md b/docs/tickflow-pro-shared-rate-limit.md
index 1f3b21d..26bfe4d 100644
--- a/docs/tickflow-pro-shared-rate-limit.md
+++ b/docs/tickflow-pro-shared-rate-limit.md
@@ -1,12 +1,12 @@
-# TickFlow Pro 共享限频（Gold Shadow + A 股研究）
+# TickFlow Pro 限频安全预算（进程内 + 跨产品错峰）
 
-## 决策
-- 同一 TickFlow API Key 的所有调用方必须共享进程级限频器：`backend/app/tickflow/rate_limits.py`。
-- 项目安全预算 = 套餐额度 × **80%**（见 `tiers.yaml` 的 `pro` 段）。
-- Gold Shadow Stage A 与 A 股同步/回测**不得**各建一套限频，避免叠加超限。
+## 决策（经 ChatGPT/Codex 评析修订）
+- `backend/app/tickflow/rate_limits.py` 提供**单 Python 进程内**的 rpm 槽位限速与 `SAFETY_RPM_FACTOR=0.8`。
+- **不要**把「各进程各扣 80%」当成账户级共享限频：Gold Shadow 容器与 A 股面板进程状态独立，理论聚合可达 160%。
+- Stage A 期间跨产品靠**错峰**，不在 Gold 上部署分布式限频重构。
 
-## 预算表示例（Pro）
-| capability | 套餐 rpm | 80% 预算 |
+## 预算表示例（Pro，单进程 80%）
+| capability | 套餐 rpm | 进程内 80% |
 |---|---:|---:|
 | quote.batch | 120 | 96 |
 | quote.pool | 60 | 48 |
@@ -15,12 +15,13 @@
 | depth5.batch | 30 | 24 |
 | adj_factor | 60 | 48 |
 
-## 错峰
-- 盘中（A 股交易时段）：优先保证 Gold Shadow 观察与 legacy 监控。
-- 盘后：A 股日线/研究批量、分钟按需回填。
+## 错峰（Stage A）
+- 盘中：优先 Gold Shadow 观察与 legacy `gold-monitor`。
+- A 股 Pro 探测/大批量同步：建议 **16:00 后**。
 - 禁止全市场一年分钟一次性回填。
 
 ## 实现要点
-- 调用方只传 `rpm`/`batch`，统一走 `sleep_between_batches` / `_reserve_slot`。
-- 应用 80% 预算：在 resolve 后对 rpm 做 `floor(rpm * 0.8)`（后续 Phase 实现，Phase 0 仅冻结约定）。
-- 任一 fallback / 429 必须写入 DatasetEvidence，禁止静默混源。
+- `resolve_limit(..., apply_safety=True)` 默认对 rpm 做 `floor(rpm * 0.8)`。
+- `sleep_between_batches`：`index=0` 只占槽不 sleep（首批突发行为已有测试文档化）；后续 batch 按槽位等待。
+- 诊断可传 `apply_safety=False`；跨容器账户预算需另设（Stage A 不做）。
+- 任一 429 / fallback 应记入证据链，禁止静默混源。

From cafa03fb5272da64763a728b4dced15e1d46b738 Mon Sep 17 00:00:00 2001
From: dunmin1980 <dunmin1980@gmail.com>
Date: Sun, 19 Jul 2026 08:54:40 +0800
Subject: [PATCH 3/6] feat: add TickFlow Pro Phase 1 probe scaffold with
 off-peak gate

Dry-run by default; live calls require a key and wait until 16:00 Asia/Shanghai unless --force, so Stage A off-peak probing stays intentional.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 backend/scripts/probe_tickflow_pro.py         | 220 ++++++++++++++++++
 backend/tests/test_probe_tickflow_pro_gate.py |  41 ++++
 docs/tickflow-pro-phase1-probe.md             |  33 +++
 3 files changed, 294 insertions(+)
 create mode 100644 backend/scripts/probe_tickflow_pro.py
 create mode 100644 backend/tests/test_probe_tickflow_pro_gate.py
 create mode 100644 docs/tickflow-pro-phase1-probe.md

diff --git a/backend/scripts/probe_tickflow_pro.py b/backend/scripts/probe_tickflow_pro.py
new file mode 100644
index 0000000..e2a997b
--- /dev/null
+++ b/backend/scripts/probe_tickflow_pro.py
@@ -0,0 +1,220 @@
+#!/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 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
+
+
+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 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": [
+            "quote.batch",
+            "kline.daily.batch",
+            "kline.minute.by_symbol",
+            "adj_factor",
+            "rate_limit_observation",
+        ],
+        "safety": {
+            "SAFETY_RPM_FACTOR": 0.8,
+            "offpeak_hour": OFFPEAK_HOUR,
+            "note": "Live probe must not overlap Gold Stage A peak window without --force.",
+        },
+        "status": "READY_FOR_LIVE",
+    }
+    (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: READY_FOR_LIVE\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) -> dict:
+    """Minimal live smoke: capability detect + one daily pull for golden symbols."""
+    out_dir.mkdir(parents=True, exist_ok=True)
+    # Prefer env for one-shot CLI; secrets_store if app context available.
+    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
+    from app.tickflow.rate_limits import SAFETY_RPM_FACTOR, apply_safety_rpm
+
+    base = _base_url() or PAID_ENDPOINT
+    tf = TickFlow(api_key=key, base_url=base)
+    samples: dict[str, object] = {}
+    errors: list[str] = []
+
+    # daily batch for 3 symbols — do not log key
+    try:
+        raw = tf.kline.daily(symbols=list(GOLDEN), limit=5)  # type: ignore[attr-defined]
+        samples["kline.daily"] = _sanitize_sample(raw if not hasattr(raw, "to_dict") else raw.to_dict())
+    except Exception as exc:  # noqa: BLE001
+        errors.append(f"kline.daily: {type(exc).__name__}")
+
+    try:
+        raw = tf.quote.get(symbols=list(GOLDEN))  # type: ignore[attr-defined]
+        samples["quote"] = _sanitize_sample(raw if not hasattr(raw, "to_dict") else raw.to_dict())
+    except Exception as exc:  # noqa: BLE001
+        # SDK surface varies; record type only
+        errors.append(f"quote: {type(exc).__name__}: try alternate API in follow-up")
+
+    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)
+    # #region agent log
+    try:
+        _dbg = Path(__file__).resolve().parents[2] / ".cursor" / "debug-976372.log"
+        _dbg.parent.mkdir(parents=True, exist_ok=True)
+        _dbg.open("a").write(
+            json.dumps(
+                {
+                    "sessionId": "976372",
+                    "runId": "phase1-scaffold",
+                    "hypothesisId": "H1",
+                    "location": "probe_tickflow_pro.py:main",
+                    "message": "probe gate decision",
+                    "data": asdict(gate),
+                    "timestamp": int(datetime.now(SHANGHAI).timestamp() * 1000),
+                },
+                ensure_ascii=False,
+            )
+            + "\n"
+        )
+    except Exception:
+        pass
+    # #endregion
+    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:
+        # Ensure app imports resolve when run as script
+        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())
diff --git a/backend/tests/test_probe_tickflow_pro_gate.py b/backend/tests/test_probe_tickflow_pro_gate.py
new file mode 100644
index 0000000..6706662
--- /dev/null
+++ b/backend/tests/test_probe_tickflow_pro_gate.py
@@ -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
diff --git a/docs/tickflow-pro-phase1-probe.md b/docs/tickflow-pro-phase1-probe.md
new file mode 100644
index 0000000..9637854
--- /dev/null
+++ b/docs/tickflow-pro-phase1-probe.md
@@ -0,0 +1,33 @@
+# 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>/`.
+
+## Live probe (off-peak)
+
+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)

From 6cb7f9f50c18a9d6327a58bd71f907985324dd59 Mon Sep 17 00:00:00 2001
From: dunmin1980 <dunmin1980@gmail.com>
Date: Sun, 19 Jul 2026 08:55:39 +0800
Subject: [PATCH 4/6] chore: drop probe debug instrumentation and ignore probe
 artifacts

Keep the Phase 1 CLI production-clean; local probe reports and debug NDJSON stay out of git.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .gitignore                            |  2 ++
 backend/scripts/probe_tickflow_pro.py | 22 ----------------------
 2 files changed, 2 insertions(+), 22 deletions(-)

diff --git a/.gitignore b/.gitignore
index fe9f019..6accff0 100644
--- a/.gitignore
+++ b/.gitignore
@@ -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
diff --git a/backend/scripts/probe_tickflow_pro.py b/backend/scripts/probe_tickflow_pro.py
index e2a997b..4192b34 100644
--- a/backend/scripts/probe_tickflow_pro.py
+++ b/backend/scripts/probe_tickflow_pro.py
@@ -176,28 +176,6 @@ def main(argv: list[str] | None = None) -> int:
     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)
-    # #region agent log
-    try:
-        _dbg = Path(__file__).resolve().parents[2] / ".cursor" / "debug-976372.log"
-        _dbg.parent.mkdir(parents=True, exist_ok=True)
-        _dbg.open("a").write(
-            json.dumps(
-                {
-                    "sessionId": "976372",
-                    "runId": "phase1-scaffold",
-                    "hypothesisId": "H1",
-                    "location": "probe_tickflow_pro.py:main",
-                    "message": "probe gate decision",
-                    "data": asdict(gate),
-                    "timestamp": int(datetime.now(SHANGHAI).timestamp() * 1000),
-                },
-                ensure_ascii=False,
-            )
-            + "\n"
-        )
-    except Exception:
-        pass
-    # #endregion
     print(json.dumps(asdict(gate), ensure_ascii=False))
     if not gate.allowed:
         return 2

From a7b272628eca492807e76c3823800950c7205f62 Mon Sep 17 00:00:00 2001
From: dunmin1980 <dunmin1980@gmail.com>
Date: Sun, 19 Jul 2026 09:09:31 +0800
Subject: [PATCH 5/6] chore: ignore local rate-limit branch transfer bundle

Keep the offline git bundle out of version control; it is only a push fallback when origin credentials lack write access.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .gitignore | 1 +
 1 file changed, 1 insertion(+)

diff --git a/.gitignore b/.gitignore
index 6accff0..8e7abbd 100644
--- a/.gitignore
+++ b/.gitignore
@@ -108,3 +108,4 @@ packaging/installed_run.log
 TickFlowStockPanel.app/
 TickFlowStockPanel-macos*.dmg
 TickFlowStockPanel-macos*.zip
+reports/cursor_tickflow_pro_rate_limit_p0.bundle

From 431550859f833c2dc84440125d3dc5e20a756e0a Mon Sep 17 00:00:00 2001
From: dunmin1980 <dunmin1980@gmail.com>
Date: Sun, 19 Jul 2026 10:06:26 +0800
Subject: [PATCH 6/6] fix: align Phase 1 probe with TickFlow SDK namespaces

Use klines.batch/quotes.get, rename dry-run status to DRY_RUN_OK, document
first-batch burst limits, and cover run_live with a fake client.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 backend/app/tickflow/rate_limits.py           |  21 ++--
 backend/scripts/probe_tickflow_pro.py         | 108 ++++++++++++------
 backend/tests/test_probe_tickflow_pro_live.py |  69 +++++++++++
 .../tests/test_rate_limit_safety_budget.py    |   6 +-
 docs/tickflow-pro-phase1-probe.md             |   6 +-
 docs/tickflow-pro-shared-rate-limit.md        |   3 +-
 6 files changed, 167 insertions(+), 46 deletions(-)
 create mode 100644 backend/tests/test_probe_tickflow_pro_live.py

diff --git a/backend/app/tickflow/rate_limits.py b/backend/app/tickflow/rate_limits.py
index e6d29af..cc3b6e3 100644
--- a/backend/app/tickflow/rate_limits.py
+++ b/backend/app/tickflow/rate_limits.py
@@ -20,11 +20,10 @@ T = TypeVar("T")
 # 不要把「各进程各扣 80%」误当成账户级共享限频。
 SAFETY_RPM_FACTOR = 0.8
 
-# 进程级共享限速器: 原先每个调用方各自本地 sleep(60/rpm), 并发同步 (kline/index/
-# depth/watchlist/custom) 时聚合请求速率会成倍超过单能力 rpm → 429。
-# 这里用一张按 rpm 分桶的「下一个可用时刻」表 (Lock 守护), 所有调用方按同一时间轴
-# 排队, 使跨调用方的聚合发包间隔 >= 60/rpm。以 rpm 为键 (调用方签名只带 rpm, 不带 cap;
-# rpm 是各能力速率的代理); 恰好同 rpm 的不同能力会共享一队, 偏保守但绝不超速。
+# 进程级共享限速器: 按 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] = {}
 
@@ -99,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)
diff --git a/backend/scripts/probe_tickflow_pro.py b/backend/scripts/probe_tickflow_pro.py
index 4192b34..e98ab62 100644
--- a/backend/scripts/probe_tickflow_pro.py
+++ b/backend/scripts/probe_tickflow_pro.py
@@ -15,6 +15,7 @@ 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")
@@ -32,6 +33,19 @@ class ProbeGate:
     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
@@ -68,6 +82,15 @@ def _sanitize_sample(obj: object) -> object:
     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 = {
@@ -75,69 +98,87 @@ def run_dry_run(out_dir: Path) -> dict:
         "as_of": datetime.now(SHANGHAI).isoformat(timespec="seconds"),
         "symbols": list(GOLDEN),
         "planned_checks": [
-            "quote.batch",
-            "kline.daily.batch",
-            "kline.minute.by_symbol",
-            "adj_factor",
+            "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": "Live probe must not overlap Gold Stage A peak window without --force.",
+            "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."
+            ),
         },
-        "status": "READY_FOR_LIVE",
+        # 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: READY_FOR_LIVE\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) -> dict:
-    """Minimal live smoke: capability detect + one daily pull for golden symbols."""
+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)
-    # Prefer env for one-shot CLI; secrets_store if app context available.
-    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
     from app.tickflow.rate_limits import SAFETY_RPM_FACTOR, apply_safety_rpm
 
-    base = _base_url() or PAID_ENDPOINT
-    tf = TickFlow(api_key=key, base_url=base)
+    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] = []
 
-    # daily batch for 3 symbols — do not log key
     try:
-        raw = tf.kline.daily(symbols=list(GOLDEN), limit=5)  # type: ignore[attr-defined]
-        samples["kline.daily"] = _sanitize_sample(raw if not hasattr(raw, "to_dict") else raw.to_dict())
+        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"kline.daily: {type(exc).__name__}")
+        errors.append(f"klines.batch.1d: {type(exc).__name__}")
 
     try:
-        raw = tf.quote.get(symbols=list(GOLDEN))  # type: ignore[attr-defined]
-        samples["quote"] = _sanitize_sample(raw if not hasattr(raw, "to_dict") else raw.to_dict())
+        raw = tf.quotes.get(symbols=list(GOLDEN), as_dataframe=False)
+        samples["quotes.get"] = _to_sample(raw)
     except Exception as exc:  # noqa: BLE001
-        # SDK surface varies; record type only
-        errors.append(f"quote: {type(exc).__name__}: try alternate API in follow-up")
+        errors.append(f"quotes.get: {type(exc).__name__}")
 
     report = {
         "mode": "live",
@@ -185,7 +226,6 @@ def main(argv: list[str] | None = None) -> int:
     if dry_run:
         report = run_dry_run(out_dir)
     else:
-        # Ensure app imports resolve when run as script
         root = _repo_root() / "backend"
         if str(root) not in sys.path:
             sys.path.insert(0, str(root))
diff --git a/backend/tests/test_probe_tickflow_pro_live.py b/backend/tests/test_probe_tickflow_pro_live.py
new file mode 100644
index 0000000..17ce96e
--- /dev/null
+++ b/backend/tests/test_probe_tickflow_pro_live.py
@@ -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"])
diff --git a/backend/tests/test_rate_limit_safety_budget.py b/backend/tests/test_rate_limit_safety_budget.py
index 24b27a0..00a56cc 100644
--- a/backend/tests/test_rate_limit_safety_budget.py
+++ b/backend/tests/test_rate_limit_safety_budget.py
@@ -37,7 +37,11 @@ def test_resolve_limit_can_skip_safety_for_diagnostics() -> None:
 
 
 def test_two_first_batches_do_not_wait_even_when_slot_reserved() -> None:
-    """Documents current behavior called out in Codex critique (first-batch burst)."""
+    """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()
diff --git a/docs/tickflow-pro-phase1-probe.md b/docs/tickflow-pro-phase1-probe.md
index 9637854..cfad58b 100644
--- a/docs/tickflow-pro-phase1-probe.md
+++ b/docs/tickflow-pro-phase1-probe.md
@@ -9,10 +9,14 @@ cd backend
 python3 scripts/probe_tickflow_pro.py
 ```
 
-Writes sanitized plan under `reports/tickflow_pro_probe/<timestamp>/`.
+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)
diff --git a/docs/tickflow-pro-shared-rate-limit.md b/docs/tickflow-pro-shared-rate-limit.md
index 26bfe4d..d11d7bb 100644
--- a/docs/tickflow-pro-shared-rate-limit.md
+++ b/docs/tickflow-pro-shared-rate-limit.md
@@ -22,6 +22,7 @@
 
 ## 实现要点
 - `resolve_limit(..., apply_safety=True)` 默认对 rpm 做 `floor(rpm * 0.8)`。
-- `sleep_between_batches`：`index=0` 只占槽不 sleep（首批突发行为已有测试文档化）；后续 batch 按槽位等待。
+- `sleep_between_batches`：`index=0` 只占槽不 sleep（首批突发；**并发多个 index=0 仍可能超 rpm**）；后续 batch 按槽位等待。
+- Phase 1 / Stage A：**不要并发**启动多个 probe 或大批量 sync。
 - 诊断可传 `apply_safety=False`；跨容器账户预算需另设（Stage A 不做）。
 - 任一 429 / fallback 应记入证据链，禁止静默混源。