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 httpx2324_TOKEN_RE = re.compile(r"fd_[A-Za-z0-9_.\-]+")252627def _redact(text: str) -> str:28 return _TOKEN_RE.sub("fd_<REDACTED>", text)293031class FinDataError(RuntimeError):32 def __init__(self, status: int, code: str, message: str, request_id: str | None = None):33 super().__init__(f"HTTP {status} {code}: {_redact(message)} (request_id={request_id})")34 self.status, self.code, self.request_id = status, code, request_id353637def load_api_key() -> str:38 key = os.environ.get("FINDATA_API_KEY")39 if key:40 return key.strip()41 path = os.path.expanduser(os.environ.get("FINDATA_API_KEY_FILE", "~/findata-secrets/api-key-wp39.txt"))42 if not os.path.exists(path):43 path = os.path.expanduser("~/Desktop/Cluster/secrets/findata/api-key-wp39.txt")44 with open(path) as fh:45 return fh.read().strip()464748@dataclass49class ExportResult:50 export_id: str51 dataset_id: str52 dataset_version: int | None53 symbols: list[str]54 start: str55 end: str56 adjustment: str57 row_count: int58 bytes: int59 sha256: str60 path: str61 request_id: str | None62 exported_at: str636465class FinDataClient:66 def __init__(67 self,68 base_url: str | None = None,69 api_key: str | None = None,70 log_path: str | os.PathLike | None = None,71 timeout: float = 60.0,72 max_attempts: int = 5,73 ):74 self.base_url = (base_url or os.environ.get("FINDATA_BASE_URL") or "https://findata.spboucher.ai/api/v1").rstrip("/")75 self._key = api_key or load_api_key()76 self.max_attempts = max_attempts77 self.log_path = Path(log_path) if log_path else None78 self._client = httpx.Client(timeout=timeout, headers={"Authorization": f"Bearer {self._key}", "User-Agent": "wp39-uqo/0.1"})7980 # ------------------------------------------------------------------ core81 def _log(self, rec: dict[str, Any]) -> None:82 if not self.log_path:83 return84 self.log_path.parent.mkdir(parents=True, exist_ok=True)85 with open(self.log_path, "a") as fh:86 fh.write(json.dumps(rec, default=str) + "\n")8788 def _request(self, method: str, path: str, *, params=None, json_body=None, stream_to: Path | None = None,89 retry: bool = True) -> httpx.Response:90 url = path if path.startswith("http") else f"{self.base_url}{path}"91 attempt, delay = 0, 0.592 while True:93 attempt += 194 t0 = time.perf_counter()95 try:96 if stream_to is not None:97 with self._client.stream(method, url, params=params, json=json_body) as r:98 if r.status_code >= 400:99 r.read()100 resp = r101 else:102 stream_to.parent.mkdir(parents=True, exist_ok=True)103 h = hashlib.sha256()104 with open(stream_to, "wb") as fh:105 for chunk in r.iter_bytes():106 fh.write(chunk)107 h.update(chunk)108 r.sha256 = h.hexdigest() # type: ignore[attr-defined]109 resp = r110 else:111 resp = self._client.request(method, url, params=params, json=json_body)112 elapsed = time.perf_counter() - t0113 self._log({114 "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "method": method, "path": path, "status": resp.status_code,115 "request_id": resp.headers.get("X-Request-ID"), "dataset_id": resp.headers.get("X-Dataset-Id"),116 "dataset_version": resp.headers.get("X-Dataset-Version"), "elapsed_s": round(elapsed, 3), "attempt": attempt,117 })118 if resp.status_code < 400:119 return resp120 retryable = resp.status_code in (429, 500, 502, 503, 504)121 if not (retry and retryable) or attempt >= self.max_attempts:122 try:123 err = resp.json().get("error", {})124 except Exception:125 err = {"code": "http_error", "message": resp.text[:300]}126 raise FinDataError(resp.status_code, err.get("code", "http_error"), err.get("message", ""), err.get("request_id"))127 except (httpx.TransportError, httpx.TimeoutException) as exc:128 self._log({"ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "method": method, "path": path, "status": None,129 "error": _redact(type(exc).__name__), "attempt": attempt})130 if not retry or attempt >= self.max_attempts:131 raise RuntimeError(f"FinData transport failure after {attempt} attempts on {path}: {_redact(str(exc))}") from None132 time.sleep(delay + random.uniform(0, delay / 2))133 delay = min(delay * 2, 8.0)134135 def get(self, path: str, **params) -> Any:136 return self._request("GET", path, params={k: v for k, v in params.items() if v is not None}).json()137138 def post(self, path: str, body: dict, retry: bool = False) -> Any:139 return self._request("POST", path, json_body=body, retry=retry).json()140141 # --------------------------------------------------------------- catalog142 def health(self) -> dict:143 return self.get("/health")144145 def openapi(self) -> dict:146 return self.get("/openapi.json")147148 def datasets(self) -> list[dict]:149 return self.get("/catalog/datasets")["data"]150151 def dataset(self, dataset_id: str) -> dict:152 return self.get(f"/catalog/datasets/{dataset_id}")153154 def schema_bars(self) -> dict:155 return self.get("/catalog/schema/bars")156157 # ----------------------------------------------------------- instruments158 def instrument(self, symbol: str) -> dict:159 return self.get(f"/instruments/{symbol}")160161 def instruments_search(self, query: str | None = None, asset_class: str | None = None, status: str | None = None,162 limit: int = 200) -> Iterator[dict]:163 cursor = None164 while True:165 page = self.get("/instruments", query=query, asset_class=asset_class, status=status, limit=limit, cursor=cursor)166 yield from page.get("data", [])167 cursor = page.get("next_cursor")168 if not cursor:169 return170171 def coverage_symbols(self, dataset_id: str, min_sessions: int = 1, limit: int = 5000) -> Iterator[dict]:172 cursor = None173 while True:174 page = self.get("/coverage/symbols", dataset_id=dataset_id, min_sessions=min_sessions, limit=limit, cursor=cursor)175 yield from page.get("data", [])176 cursor = page.get("next_cursor")177 if not cursor:178 return179180 def coverage(self, dataset_id: str, symbol: str, start: str, end: str, limit: int = 5000) -> list[dict]:181 return self.get("/coverage", dataset_id=dataset_id, symbol=symbol, start=start, end=end, limit=limit)["data"]182183 # ---------------------------------------------------------- fundamentals184 def fundamentals(self, entity_id: str, as_of: str | None = None, metrics: list[str] | None = None,185 period_type: str | None = "q", limit: int = 20000, include_superseded: bool = False) -> dict:186 return self.get(f"/fundamentals/{entity_id}", as_of=as_of, metrics=",".join(metrics) if metrics else None,187 period_type=period_type, limit=limit, include_superseded=str(include_superseded).lower())188189 def fundamentals_periods(self, entity_id: str) -> list[dict]:190 return self.get(f"/fundamentals/{entity_id}/periods")["periods"]191192 # ----------------------------------------------------------------- query193 def query(self, sql: str, max_rows: int = 1000) -> tuple[list[str], list[list[Any]], dict]:194 d = self.post("/query", {"sql": sql, "max_rows": max_rows}, retry=True)195 return [c["name"] for c in d["columns"]], d["rows"], d.get("meta", {})196197 # --------------------------------------------------------------- exports198 def export(self, dataset_id: str, symbols: list[str], start: str, end: str, dest: Path, adjustment: str = "raw",199 poll_s: float = 1.0, max_wait_s: float = 600.0) -> ExportResult:200 body = {"dataset_id": dataset_id, "symbols": symbols, "start": start, "end": end, "adjustment": adjustment, "format": "parquet"}201 r = self._request("POST", "/exports", json_body=body, retry=False)202 created = r.json()203 export_id, request_id = created["id"], r.headers.get("X-Request-ID")204 t0 = time.time()205 while True:206 st = self.get(f"/exports/{export_id}")207 if st["status"] == "ready":208 break209 if st["status"] in ("failed", "expired"):210 raise FinDataError(500, "export_" + st["status"], str(st.get("error")), request_id)211 if time.time() - t0 > max_wait_s:212 raise FinDataError(504, "export_timeout", f"export {export_id} not ready after {max_wait_s}s", request_id)213 time.sleep(poll_s)214 resp = self._request("GET", f"/exports/{export_id}/download", stream_to=dest)215 snap = st.get("snapshot") or {}216 return ExportResult(export_id=export_id, dataset_id=dataset_id, dataset_version=snap.get("version"), symbols=symbols,217 start=start, end=end, adjustment=adjustment, row_count=int(st.get("row_count") or 0),218 bytes=int(st.get("bytes") or 0), sha256=getattr(resp, "sha256", ""), path=str(dest),219 request_id=request_id, exported_at=time.strftime("%Y-%m-%dT%H:%M:%S%z"))220221 def close(self) -> None:222 self._client.close()223224225def export_result_dict(res: ExportResult) -> dict:226 return asdict(res)227