1"""Typed, redacting client for the FinData OS API (https://findata.spboucher.ai/api/v1).23Design rules (docs/API_INTEGRATION.md):4- the bearer token is read from the environment or a mode-600 file, never logged, never placed in a URL;5- bounded exponential backoff with jitter on 429/5xx and transport errors (idempotent GETs; export creation only when6 no export id was obtained);7- every call appends one provenance line (method, path, status, request id, dataset id/version, elapsed) to a JSONL log;8- Parquet exports are validated against the published bars schema before being accepted.9"""10from __future__ import annotations1112import hashlib13import json14import os15import random16import re17import time18from dataclasses import dataclass, asdict19from pathlib import Path20from typing import Any, Iterator2122import httpx23import threading2425_TOKEN_RE = re.compile(r"fd_[A-Za-z0-9_.\-]+")262728def _redact(text: str) -> str:29 return _TOKEN_RE.sub("fd_<REDACTED>", text)303132class FinDataError(RuntimeError):33 def __init__(self, status: int, code: str, message: str, request_id: str | None = None):34 super().__init__(f"HTTP {status} {code}: {_redact(message)} (request_id={request_id})")35 self.status, self.code, self.request_id = status, code, request_id363738def load_api_key() -> str:39 key = os.environ.get("FINDATA_API_KEY")40 if key:41 return key.strip()42 path = os.path.expanduser(os.environ.get("FINDATA_API_KEY_FILE", "~/findata-secrets/api-key-wp39.txt"))43 if not os.path.exists(path):44 path = os.path.expanduser("~/Desktop/Cluster/secrets/findata/api-key-wp39.txt")45 with open(path) as fh:46 return fh.read().strip()474849@dataclass50class ExportResult:51 export_id: str52 dataset_id: str53 dataset_version: int | None54 symbols: list[str]55 start: str56 end: str57 adjustment: str58 row_count: int59 bytes: int60 sha256: str61 path: str62 request_id: str | None63 exported_at: str646566class _RateLimiter:67 """Process-wide token bucket shared by every client/thread (default 6 req/s; override with WP39_RATE)."""68 _lock = threading.Lock()69 _last = 0.07071 @classmethod72 def wait(cls) -> None:73 rate = float(os.environ.get("WP39_RATE", "6"))74 if rate <= 0:75 return76 with cls._lock:77 gap = 1.0 / rate - (time.monotonic() - cls._last)78 if gap > 0:79 time.sleep(gap)80 cls._last = time.monotonic()818283class FinDataClient:84 def __init__(85 self,86 base_url: str | None = None,87 api_key: str | None = None,88 log_path: str | os.PathLike | None = None,89 timeout: float = 60.0,90 max_attempts: int = 8,91 ):92 self.base_url = (base_url or os.environ.get("FINDATA_BASE_URL") or "https://findata.spboucher.ai/api/v1").rstrip("/")93 self._key = api_key or load_api_key()94 self.max_attempts = max_attempts95 self.log_path = Path(log_path) if log_path else None96 self._client = httpx.Client(timeout=timeout, headers={"Authorization": f"Bearer {self._key}", "User-Agent": "wp39-uqo/0.1"})9798 # ------------------------------------------------------------------ core99 def _log(self, rec: dict[str, Any]) -> None:100 if not self.log_path:101 return102 self.log_path.parent.mkdir(parents=True, exist_ok=True)103 with open(self.log_path, "a") as fh:104 fh.write(json.dumps(rec, default=str) + "\n")105106 def _request(self, method: str, path: str, *, params=None, json_body=None, stream_to: Path | None = None,107 retry: bool = True) -> httpx.Response:108 url = path if path.startswith("http") else f"{self.base_url}{path}"109 attempt, delay = 0, 0.5110 while True:111 attempt += 1112 _RateLimiter.wait()113 t0 = time.perf_counter()114 try:115 if stream_to is not None:116 with self._client.stream(method, url, params=params, json=json_body) as r:117 if r.status_code >= 400:118 r.read()119 resp = r120 else:121 stream_to.parent.mkdir(parents=True, exist_ok=True)122 h = hashlib.sha256()123 with open(stream_to, "wb") as fh:124 for chunk in r.iter_bytes():125 fh.write(chunk)126 h.update(chunk)127 r.sha256 = h.hexdigest() # type: ignore[attr-defined]128 resp = r129 else:130 resp = self._client.request(method, url, params=params, json=json_body)131 elapsed = time.perf_counter() - t0132 self._log({133 "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "method": method, "path": path, "status": resp.status_code,134 "request_id": resp.headers.get("X-Request-ID"), "dataset_id": resp.headers.get("X-Dataset-Id"),135 "dataset_version": resp.headers.get("X-Dataset-Version"), "elapsed_s": round(elapsed, 3), "attempt": attempt,136 })137 if resp.status_code < 400:138 return resp139 retryable = resp.status_code in (429, 500, 502, 503, 504)140 if resp.status_code == 429:141 ra = resp.headers.get("Retry-After")142 delay = max(delay, float(ra) if ra and ra.replace(".", "", 1).isdigit() else 10.0)143 retry = True # a rate-limited call never executed server-side: always safe to retry144 if not (retry and retryable) or attempt >= self.max_attempts:145 try:146 err = resp.json().get("error", {})147 except Exception:148 err = {"code": "http_error", "message": resp.text[:300]}149 raise FinDataError(resp.status_code, err.get("code", "http_error"), err.get("message", ""), err.get("request_id"))150 except (httpx.TransportError, httpx.TimeoutException) as exc:151 self._log({"ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "method": method, "path": path, "status": None,152 "error": _redact(type(exc).__name__), "attempt": attempt})153 if not retry or attempt >= self.max_attempts:154 raise RuntimeError(f"FinData transport failure after {attempt} attempts on {path}: {_redact(str(exc))}") from None155 time.sleep(delay + random.uniform(0, delay / 2))156 delay = min(delay * 2, 60.0)157158 def get(self, path: str, **params) -> Any:159 return self._request("GET", path, params={k: v for k, v in params.items() if v is not None}).json()160161 def post(self, path: str, body: dict, retry: bool = False) -> Any:162 return self._request("POST", path, json_body=body, retry=retry).json()163164 # --------------------------------------------------------------- catalog165 def health(self) -> dict:166 return self.get("/health")167168 def openapi(self) -> dict:169 return self.get("/openapi.json")170171 def datasets(self) -> list[dict]:172 return self.get("/catalog/datasets")["data"]173174 def dataset(self, dataset_id: str) -> dict:175 return self.get(f"/catalog/datasets/{dataset_id}")176177 def schema_bars(self) -> dict:178 return self.get("/catalog/schema/bars")179180 # ----------------------------------------------------------- instruments181 def instrument(self, symbol: str) -> dict:182 return self.get(f"/instruments/{symbol}")183184 def instruments_search(self, query: str | None = None, asset_class: str | None = None, status: str | None = None,185 limit: int = 200) -> Iterator[dict]:186 cursor = None187 while True:188 page = self.get("/instruments", query=query, asset_class=asset_class, status=status, limit=limit, cursor=cursor)189 yield from page.get("data", [])190 cursor = page.get("next_cursor")191 if not cursor:192 return193194 def coverage_symbols(self, dataset_id: str, min_sessions: int = 1, limit: int = 5000) -> Iterator[dict]:195 cursor = None196 while True:197 page = self.get("/coverage/symbols", dataset_id=dataset_id, min_sessions=min_sessions, limit=limit, cursor=cursor)198 yield from page.get("data", [])199 cursor = page.get("next_cursor")200 if not cursor:201 return202203 def coverage(self, dataset_id: str, symbol: str, start: str, end: str, limit: int = 5000) -> list[dict]:204 return self.get("/coverage", dataset_id=dataset_id, symbol=symbol, start=start, end=end, limit=limit)["data"]205206 # ---------------------------------------------------------- fundamentals207 def fundamentals(self, entity_id: str, as_of: str | None = None, metrics: list[str] | None = None,208 period_type: str | None = "q", limit: int = 20000, include_superseded: bool = False) -> dict:209 return self.get(f"/fundamentals/{entity_id}", as_of=as_of, metrics=",".join(metrics) if metrics else None,210 period_type=period_type, limit=limit, include_superseded=str(include_superseded).lower())211212 def fundamentals_periods(self, entity_id: str) -> list[dict]:213 return self.get(f"/fundamentals/{entity_id}/periods")["periods"]214215 # ----------------------------------------------------------------- query216 def query(self, sql: str, max_rows: int = 1000) -> tuple[list[str], list[list[Any]], dict]:217 d = self.post("/query", {"sql": sql, "max_rows": max_rows}, retry=True)218 return [c["name"] for c in d["columns"]], d["rows"], d.get("meta", {})219220 # --------------------------------------------------------------- exports221 def export(self, dataset_id: str, symbols: list[str], start: str, end: str, dest: Path, adjustment: str = "raw",222 poll_s: float = 1.0, max_wait_s: float = 600.0) -> ExportResult:223 body = {"dataset_id": dataset_id, "symbols": symbols, "start": start, "end": end, "adjustment": adjustment, "format": "parquet"}224 r = self._request("POST", "/exports", json_body=body, retry=False)225 created = r.json()226 export_id, request_id = created["id"], r.headers.get("X-Request-ID")227 t0 = time.time()228 while True:229 st = self.get(f"/exports/{export_id}")230 if st["status"] == "ready":231 break232 if st["status"] in ("failed", "expired"):233 raise FinDataError(500, "export_" + st["status"], str(st.get("error")), request_id)234 if time.time() - t0 > max_wait_s:235 raise FinDataError(504, "export_timeout", f"export {export_id} not ready after {max_wait_s}s", request_id)236 time.sleep(poll_s)237 resp = self._request("GET", f"/exports/{export_id}/download", stream_to=dest)238 snap = st.get("snapshot") or {}239 return ExportResult(export_id=export_id, dataset_id=dataset_id, dataset_version=snap.get("version"), symbols=symbols,240 start=start, end=end, adjustment=adjustment, row_count=int(st.get("row_count") or 0),241 bytes=int(st.get("bytes") or 0), sha256=getattr(resp, "sha256", ""), path=str(dest),242 request_id=request_id, exported_at=time.strftime("%Y-%m-%dT%H:%M:%S%z"))243244 def close(self) -> None:245 self._client.close()246247248def export_result_dict(res: ExportResult) -> dict:249 return asdict(res)250