1"""Typed, redacting client for the FinData OS API (contract: docs/DATA_CARD.md §connector).23Rules: bearer token from env/secret file only (never logged, never in URLs); bounded exponential backoff with jitter on4429/5xx; every call appends a provenance record (method, path, status, X-Request-ID, X-Dataset-Id, X-Dataset-Version,5elapsed) to a JSONL log; exports are validated against the published bars schema and hashed on disk.6"""7from __future__ import annotations89import hashlib10import json11import os12import random13import re14import threading15import time16from dataclasses import dataclass, asdict17from datetime import datetime, timezone18from pathlib import Path19from typing import Any, Iterator2021import httpx22import pyarrow as pa23import pyarrow.ipc as ipc24import pyarrow.parquet as pq2526_TOKEN_RE = re.compile(r"fd_[A-Za-z0-9_.\-]+")27DEFAULT_BASE = "https://findata.spboucher.ai/api/v1"28BARS_1M_MAX_DAYS = 3129EXPORT_MAX_SYMBOLS = 500303132def redact(text: str) -> str:33 return _TOKEN_RE.sub("fd_<REDACTED>", str(text))343536class FinDataError(RuntimeError):37 def __init__(self, status: int, code: str, message: str, request_id: str | None = None):38 super().__init__(f"HTTP {status} {code}: {redact(message)} (request_id={request_id})")39 self.status, self.code, self.request_id = status, code, request_id404142def load_api_key() -> str:43 key = os.environ.get("FINDATA_API_KEY")44 if key:45 return key.strip()46 candidates = [os.environ.get("FINDATA_API_KEY_FILE"), "~/findata-secrets/api-key-wp41.txt",47 "~/findata-secrets/api-key-claude-code.txt",48 "~/Desktop/Cluster/secrets/findata/api-key-claude-code.txt"]49 for c in candidates:50 if not c:51 continue52 p = Path(os.path.expanduser(c))53 if p.exists():54 return p.read_text().strip()55 raise FileNotFoundError("FinData API key not found (FINDATA_API_KEY or FINDATA_API_KEY_FILE)")565758def base_url() -> str:59 return os.environ.get("FINDATA_BASE_URL", DEFAULT_BASE).rstrip("/")606162class _RateLimiter:63 """Process-wide token bucket shared by every client/thread (default 5 req/s; override with WP41_RATE)."""64 _lock = threading.Lock()65 _last = 0.06667 @classmethod68 def wait(cls) -> None:69 rate = float(os.environ.get("WP41_RATE", "5"))70 if rate <= 0:71 return72 with cls._lock:73 gap = 1.0 / rate - (time.monotonic() - cls._last)74 if gap > 0:75 time.sleep(gap)76 cls._last = time.monotonic()777879@dataclass80class Provenance:81 ts: str82 method: str83 path: str84 status: int85 request_id: str | None86 dataset_id: str | None87 dataset_version: str | None88 elapsed_ms: int89 note: str | None = None909192@dataclass93class ExportResult:94 export_id: str95 dataset_id: str96 dataset_version: int | None97 symbols: list[str]98 start: str99 end: str100 adjustment: str101 row_count: int102 bytes: int103 sha256: str104 path: str105 request_id: str | None106 retrieved_at: str107 revision_ids: list[int]108109110def sha256_file(path: str | Path) -> str:111 h = hashlib.sha256()112 with open(path, "rb") as fh:113 for chunk in iter(lambda: fh.read(1 << 20), b""):114 h.update(chunk)115 return h.hexdigest()116117118class FinDataClient:119 def __init__(self, base: str | None = None, api_key: str | None = None, log_path: str | Path | None = None,120 timeout: float = 180.0, max_retries: int = 6):121 self.base = (base or base_url()).rstrip("/")122 self._key = api_key or load_api_key()123 self.log_path = Path(log_path) if log_path else None124 self.max_retries = max_retries125 self._client = httpx.Client(base_url=self.base, timeout=timeout,126 headers={"Authorization": f"Bearer {self._key}", "User-Agent": "wp41-chronos-x/0.1"})127 self._lock = threading.Lock()128129 # ---------------------------------------------------------------- low level130 def _log(self, rec: Provenance) -> None:131 if not self.log_path:132 return133 self.log_path.parent.mkdir(parents=True, exist_ok=True)134 with self._lock, open(self.log_path, "a") as fh:135 fh.write(json.dumps(asdict(rec)) + "\n")136137 def request(self, method: str, path: str, *, params: dict | None = None, json_body: dict | None = None,138 headers: dict | None = None, stream_to: Path | None = None, idempotent: bool = True) -> httpx.Response:139 attempt = 0140 while True:141 _RateLimiter.wait()142 t0 = time.monotonic()143 try:144 if stream_to is not None:145 with self._client.stream(method, path, params=params, headers=headers) as r:146 if r.status_code < 400:147 stream_to.parent.mkdir(parents=True, exist_ok=True)148 with open(stream_to, "wb") as fh:149 for chunk in r.iter_bytes(1 << 20):150 fh.write(chunk)151 else:152 r.read()153 resp = r154 else:155 resp = self._client.request(method, path, params=params, json=json_body, headers=headers)156 except (httpx.TransportError,) as exc:157 attempt += 1158 if attempt > self.max_retries or not idempotent:159 raise FinDataError(0, "transport", redact(str(exc)))160 time.sleep(min(30, 0.5 * 2 ** attempt) + random.random())161 continue162 elapsed = int((time.monotonic() - t0) * 1000)163 self._log(Provenance(datetime.now(timezone.utc).isoformat(), method, redact(path), resp.status_code,164 resp.headers.get("x-request-id"), resp.headers.get("x-dataset-id"),165 resp.headers.get("x-dataset-version"), elapsed))166 if resp.status_code in (429, 500, 502, 503, 504) and attempt < self.max_retries and idempotent:167 attempt += 1168 ra = resp.headers.get("retry-after")169 time.sleep(float(ra) if ra and ra.isdigit() else min(60, 0.5 * 2 ** attempt) + random.random())170 continue171 if resp.status_code >= 400:172 try:173 err = resp.json().get("error", {})174 except Exception:175 err = {}176 raise FinDataError(resp.status_code, err.get("code", "http_error"),177 err.get("message", resp.text[:300]), err.get("request_id"))178 return resp179180 def get_json(self, path: str, **params: Any) -> dict:181 return self.request("GET", path, params={k: v for k, v in params.items() if v is not None}).json()182183 def post_json(self, path: str, body: dict, idempotent: bool = False) -> dict:184 return self.request("POST", path, json_body=body, idempotent=idempotent).json()185186 # ---------------------------------------------------------------- catalog & reference187 def health(self) -> dict:188 return self.get_json("/health")189190 def datasets(self) -> list[dict]:191 return self.get_json("/catalog/datasets")["data"]192193 def dataset(self, dataset_id: str) -> dict:194 return self.get_json(f"/catalog/datasets/{dataset_id}")195196 def bars_schema(self) -> dict:197 return self.get_json("/catalog/schema/bars")198199 def openapi(self) -> dict:200 return self.request("GET", "/openapi.json").json()201202 def instruments(self, asset_class: str | None = None, status: str | None = None, query: str = "",203 limit: int = 500) -> Iterator[dict]:204 cursor = None205 while True:206 d = self.get_json("/instruments", query=query or None, asset_class=asset_class, status=status, limit=limit,207 cursor=cursor)208 yield from d["data"]209 cursor = d.get("next_cursor")210 if not cursor or not d["data"]:211 break212213 def instrument(self, symbol: str) -> dict:214 return self.get_json(f"/instruments/{symbol}")215216 def sessions(self, symbol: str, start: str, end: str, interval: str = "1m") -> list[dict]:217 return self.get_json("/market/sessions", symbol=symbol, start=start, end=end, interval=interval)["data"]218219 def coverage(self, symbol: str, start: str, end: str, dataset_id: str = "market_bars.stock.1m.raw",220 limit: int = 5000) -> list[dict]:221 return self.get_json("/coverage", symbol=symbol, start=start, end=end, dataset_id=dataset_id, limit=limit)["data"]222223 def anomalies(self, symbol: str, dataset_id: str = "market_bars.stock.1m.raw", limit: int = 5000) -> list[dict]:224 return self.get_json("/quality/anomalies", symbol=symbol, dataset_id=dataset_id, limit=limit)["data"]225226 def sql(self, sql: str, max_rows: int = 100_000) -> tuple[list[str], list[dict], dict]:227 d = self.post_json("/query", {"sql": sql, "max_rows": max_rows}, idempotent=True)228 return [c["name"] for c in d["columns"]], d["rows"], d["meta"]229230 # ---------------------------------------------------------------- bars (small, paginated, Arrow transport)231 def bars(self, symbol: str, start: str, end: str, interval: str = "1m", adjustment: str = "raw",232 limit: int = 10_000) -> pa.Table:233 """All pages for one symbol/range (1m ≤ 31 days per call enforced by the server)."""234 tables, cursor = [], None235 while True:236 params = {"symbol": symbol, "interval": interval, "start": start, "end": end, "adjustment": adjustment,237 "limit": limit, "format": "arrow"}238 if cursor:239 params["cursor"] = cursor240 r = self.request("GET", "/market/bars", params=params)241 with ipc.open_stream(r.content) as reader:242 tables.append(reader.read_all())243 cursor = r.headers.get("x-next-cursor")244 if not cursor:245 break246 if not tables:247 return pa.table({})248 return pa.concat_tables(tables)249250 # ---------------------------------------------------------------- exports (large, async, pinned version)251 def export(self, symbols: list[str], start: str, end: str, dest: Path, dataset_id: str = "market_bars.stock.1m.raw",252 adjustment: str = "raw", poll: float = 2.0, max_wait: float = 3600.0) -> ExportResult:253 if len(symbols) > EXPORT_MAX_SYMBOLS:254 raise ValueError(f"max {EXPORT_MAX_SYMBOLS} symbols per export")255 body = {"dataset_id": dataset_id, "symbols": symbols, "start": start, "end": end, "adjustment": adjustment,256 "format": "parquet"}257 r = self.request("POST", "/exports", json_body=body, idempotent=False)258 created = r.json()259 eid = created["id"]260 t0 = time.monotonic()261 while True:262 st = self.get_json(f"/exports/{eid}")263 if st["status"] == "ready":264 break265 if st["status"] in ("failed", "expired", "cancelled"):266 raise FinDataError(500, "export_" + st["status"], str(st.get("error")), r.headers.get("x-request-id"))267 if time.monotonic() - t0 > max_wait:268 raise FinDataError(504, "export_timeout", f"export {eid} not ready after {max_wait}s")269 time.sleep(poll)270 tmp = dest.with_suffix(dest.suffix + ".part")271 rd = self.request("GET", f"/exports/{eid}/download", stream_to=tmp)272 meta = pq.read_metadata(tmp)273 tbl_schema = pq.read_schema(tmp)274 validate_bars_schema(tbl_schema)275 rev = pq.read_table(tmp, columns=["revision_id"]).column("revision_id").unique().to_pylist()276 tmp.replace(dest)277 snap = st.get("snapshot") or {}278 return ExportResult(eid, dataset_id, snap.get("version"), symbols, start, end, adjustment,279 int(st.get("row_count") or meta.num_rows), int(st.get("bytes") or dest.stat().st_size),280 sha256_file(dest), str(dest), rd.headers.get("x-request-id") or r.headers.get("x-request-id"),281 datetime.now(timezone.utc).isoformat(), sorted(int(x) for x in rev))282283284BARS_REQUIRED = {"instrument_id", "symbol_as_reported", "asset_class", "interval", "timestamp_utc", "session_date",285 "open", "high", "low", "close", "volume", "adjustment_type", "quality_flags", "revision_id"}286287288def validate_bars_schema(schema: pa.Schema) -> None:289 missing = BARS_REQUIRED - set(schema.names)290 if missing:291 raise ValueError(f"bars schema missing columns: {sorted(missing)}")292 ts = schema.field("timestamp_utc").type293 if not pa.types.is_timestamp(ts) or ts.tz is None:294 raise ValueError(f"timestamp_utc must be tz-aware timestamp, got {ts}")295 for c in ("open", "high", "low", "close", "volume"):296 if not pa.types.is_floating(schema.field(c).type):297 raise ValueError(f"{c} must be float, got {schema.field(c).type}")298