diff --git a/.gitignore b/.gitignore
index fe9f019..8e7abbd 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
@@ -106,3 +108,4 @@ packaging/installed_run.log
 TickFlowStockPanel.app/
 TickFlowStockPanel-macos*.dmg
 TickFlowStockPanel-macos*.zip
+reports/cursor_tickflow_pro_rate_limit_p0.bundle
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..cc3b6e3 100644
--- a/backend/app/tickflow/rate_limits.py
+++ b/backend/app/tickflow/rate_limits.py
@@ -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)
diff --git a/backend/scripts/probe_tickflow_pro.py b/backend/scripts/probe_tickflow_pro.py
new file mode 100644
index 0000000..e98ab62
--- /dev/null
+++ b/backend/scripts/probe_tickflow_pro.py
@@ -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())
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/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
new file mode 100644
index 0000000..00a56cc
--- /dev/null
+++ b/backend/tests/test_rate_limit_safety_budget.py
@@ -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
diff --git a/docs/tickflow-pro-phase1-probe.md b/docs/tickflow-pro-phase1-probe.md
new file mode 100644
index 0000000..cfad58b
--- /dev/null
+++ b/docs/tickflow-pro-phase1-probe.md
@@ -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)
diff --git a/docs/tickflow-pro-shared-rate-limit.md b/docs/tickflow-pro-shared-rate-limit.md
new file mode 100644
index 0000000..d11d7bb
--- /dev/null
+++ b/docs/tickflow-pro-shared-rate-limit.md
@@ -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 应记入证据链，禁止静默混源。