spb/wp46_uqo

1"""Typed, redacting client for the FinData OS API (https://findata.spboucher.ai/api/v1).23Contract verified against the live OpenAPI on 2026-11-09 (see docs/findata_audit.md). Rules: bearer token from the4environment or the owner's secret file only (never logged, never in URLs); process-wide token bucket (default 5 req/s;5the server allows 600 req/min per key); bounded exponential backoff with jitter on 429/5xx; every call appends a6provenance record (method, path, status, X-Request-ID, X-Dataset-Version, elapsed) to a JSONL log; exports are hashed on7disk. Nothing here prints bar values.8"""9from __future__ import annotations1011import hashlib12import json13import os14import random15import re16import threading17import time18from dataclasses import asdict, dataclass19from datetime import datetime, timezone20from pathlib import Path21from typing import Any2223import httpx24import pyarrow as pa25import pyarrow.ipc as ipc2627_TOKEN_RE = re.compile(r"fd_[A-Za-z0-9_.\-]+")28DEFAULT_BASE = "https://findata.spboucher.ai/api/v1"29SECRET_CANDIDATES = ["~/findata-secrets/api-key-wp46.txt", "~/findata-secrets/api-key-claude-code.txt",30                     "~/Desktop/Cluster/secrets/findata/api-key-claude-code.txt"]31MAX_DAYS_PER_CALL = {"1m": 31, "5m": 186, "30m": 3650, "1h": 3650, "1d": 3650}323334def redact(text: Any) -> str:35    return _TOKEN_RE.sub("fd_<REDACTED>", str(text))363738class FinDataError(RuntimeError):39    def __init__(self, status: int, code: str, message: str, request_id: str | None = None):40        super().__init__(f"HTTP {status} {code}: {redact(message)} (request_id={request_id})")41        self.status, self.code, self.request_id = status, code, request_id424344def load_api_key() -> str:45    key = os.environ.get("FINDATA_API_KEY")46    if key:47        return key.strip()48    f = os.environ.get("FINDATA_API_KEY_FILE")49    for c in ([f] if f else []) + SECRET_CANDIDATES:50        p = Path(os.path.expanduser(c))51        if p.exists():52            return p.read_text().strip()53    raise FileNotFoundError("FinData API key not found (set FINDATA_API_KEY or FINDATA_API_KEY_FILE)")545556def base_url() -> str:57    return os.environ.get("FINDATA_BASE_URL", DEFAULT_BASE).rstrip("/")585960class _RateLimiter:61    """Process-wide token bucket shared by every client/thread (override with WP46_RATE requests per second)."""62    _lock = threading.Lock()63    _last = 0.06465    @classmethod66    def wait(cls) -> None:67        rate = float(os.environ.get("WP46_RATE", "5"))68        if rate <= 0:69            return70        with cls._lock:71            gap = 1.0 / rate - (time.monotonic() - cls._last)72            if gap > 0:73                time.sleep(gap)74            cls._last = time.monotonic()757677@dataclass78class Provenance:79    ts: str80    method: str81    path: str82    status: int83    request_id: str | None84    dataset_version: str | None85    elapsed_ms: int86    note: str | None = None878889def sha256_file(path: str | Path) -> str:90    h = hashlib.sha256()91    with open(path, "rb") as fh:92        for chunk in iter(lambda: fh.read(1 << 20), b""):93            h.update(chunk)94    return h.hexdigest()959697class FinDataClient:98    def __init__(self, base: str | None = None, api_key: str | None = None, log_path: str | Path | None = None,99                 timeout: float = 300.0, max_retries: int = 6):100        self.base = (base or base_url()).rstrip("/")101        self._key = api_key or load_api_key()102        self.log_path = Path(log_path) if log_path else None103        self.max_retries = max_retries104        self._client = httpx.Client(base_url=self.base, timeout=timeout,105                                    headers={"Authorization": f"Bearer {self._key}", "User-Agent": "wp46-garch-benchmark/0.1"})106        self._lock = threading.Lock()107108    # ------------------------------------------------------------------ low level109    def _log(self, rec: Provenance) -> None:110        if not self.log_path:111            return112        self.log_path.parent.mkdir(parents=True, exist_ok=True)113        with self._lock, open(self.log_path, "a") as fh:114            fh.write(json.dumps(asdict(rec)) + "\n")115116    def request(self, method: str, path: str, *, params: dict | None = None, json_body: dict | None = None,117                headers: dict | None = None, note: str | None = None) -> httpx.Response:118        attempt = 0119        while True:120            _RateLimiter.wait()121            t0 = time.monotonic()122            try:123                r = self._client.request(method, path, params=params, json=json_body, headers=headers)124            except (httpx.TransportError, httpx.TimeoutException) as e:  # network: retry125                attempt += 1126                if attempt > self.max_retries:127                    raise FinDataError(0, "transport", redact(e)) from None128                time.sleep(min(60.0, (2 ** attempt) + random.random()))129                continue130            el = int((time.monotonic() - t0) * 1000)131            self._log(Provenance(datetime.now(timezone.utc).isoformat(), method, path, r.status_code,132                                 r.headers.get("x-request-id"), r.headers.get("x-dataset-version"), el, note))133            if r.status_code in (429, 500, 502, 503, 504) and attempt < self.max_retries:134                attempt += 1135                ra = r.headers.get("retry-after")136                wait = float(ra) if ra and ra.replace(".", "", 1).isdigit() else min(60.0, (2 ** attempt) + random.random())137                time.sleep(wait)138                continue139            if r.status_code >= 400:140                try:141                    err = r.json().get("error", {})142                except Exception:143                    err = {}144                raise FinDataError(r.status_code, err.get("code", "http_error"), err.get("message", r.text[:300]),145                                   r.headers.get("x-request-id"))146            return r147148    def get(self, path: str, **params: Any) -> Any:149        return self.request("GET", path, params={k: v for k, v in params.items() if v is not None}).json()150151    def post(self, path: str, body: dict) -> Any:152        return self.request("POST", path, json_body=body).json()153154    # ------------------------------------------------------------------ catalog155    def health(self) -> dict:156        return self.get("/health")157158    def openapi(self) -> dict:159        return self.get("/openapi.json")160161    def datasets(self) -> list[dict]:162        return self.get("/catalog/datasets")["data"]163164    def dataset(self, dataset_id: str) -> dict:165        return self.get(f"/catalog/datasets/{dataset_id}")166167    def instrument(self, symbol: str) -> dict:168        return self.get(f"/instruments/{symbol}")169170    def bars_schema(self) -> dict:171        return self.get("/catalog/schema/bars")172173    # ------------------------------------------------------------------ data174    def sql(self, sql: str, max_rows: int = 10000) -> pa.Table:175        d = self.post("/query", {"sql": sql, "max_rows": max_rows})176        cols = [c["name"] for c in d["columns"]]177        rows = d["rows"]178        if not rows:179            return pa.table({c: [] for c in cols})180        if isinstance(rows[0], dict):181            return pa.Table.from_pylist(rows)182        return pa.table({c: [r[i] for r in rows] for i, c in enumerate(cols)})183184    def bars(self, symbol: str, interval: str, start: str, end: str, adjustment: str = "raw",185             limit: int = 10000) -> pa.Table:186        """All pages of GET /market/bars for one symbol/interval/range (range must respect MAX_DAYS_PER_CALL)."""187        tables: list[pa.Table] = []188        cursor = None189        while True:190            r = self.request("GET", "/market/bars", params={k: v for k, v in dict(191                symbol=symbol, interval=interval, start=start, end=end, adjustment=adjustment, limit=limit,192                cursor=cursor).items() if v is not None}, headers={"Accept": "application/vnd.apache.arrow.stream"})193            ctype = r.headers.get("content-type", "")194            if "arrow" in ctype:195                tables.append(ipc.open_stream(r.content).read_all())196            else:197                d = r.json()198                tables.append(pa.Table.from_pylist(d.get("data", [])))199            cursor = r.headers.get("x-next-cursor")200            if not cursor:201                break202        return pa.concat_tables([t for t in tables if t.num_rows]) if any(t.num_rows for t in tables) else tables[0]203204    def export(self, dataset_id: str, symbols: list[str], start: str, end: str, dest: str | Path,205               adjustment: str | None = None, poll: float = 2.0) -> dict:206        body: dict[str, Any] = {"dataset_id": dataset_id, "symbols": symbols, "start": start, "end": end}207        if adjustment:208            body["adjustment"] = adjustment209        job = self.post("/exports", body)210        eid = job["id"]211        while True:212            st = self.get(f"/exports/{eid}")213            if st["status"] == "ready":214                break215            if st["status"] in ("failed", "expired"):216                raise FinDataError(500, "export_" + st["status"], str(st.get("error")))217            time.sleep(poll)218        dest = Path(dest)219        dest.parent.mkdir(parents=True, exist_ok=True)220        r = self.request("GET", f"/exports/{eid}/download")221        dest.write_bytes(r.content)222        return {"export_id": eid, "dataset_id": dataset_id, "dataset_version": (st.get("snapshot") or {}).get("version"),223                "symbols": symbols, "start": start, "end": end, "row_count": st.get("row_count"),224                "bytes": st.get("bytes"), "sha256": sha256_file(dest), "path": str(dest),225                "retrieved_at": datetime.now(timezone.utc).isoformat()}226