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

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

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

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"])

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

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)

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 应记入证据链，禁止静默混源。