From cafa03fb5272da64763a728b4dced15e1d46b738 Mon Sep 17 00:00:00 2001 From: dunmin1980 Date: Sun, 19 Jul 2026 08:54:40 +0800 Subject: [PATCH] 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 --- 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: /reports/tickflow_pro_probe/)", + ) + 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//`. + +## 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)