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//`. +Writes sanitized plan under `reports/tickflow_pro_probe//` 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 应记入证据链，禁止静默混源。