1"""HTTP client for the FirstRate Data API. The user id is appended at request time and never logged.23Vendor facts (verified 2026-11-01): auth = `userid` query parameter; archive endpoints answer HTTP 302 to a CDN4(x01.frd-data.com / backblazeb2.com) serving a zip; `last_update` returns a bare date (fx uses yyyyMMdd);5no documented rate limit — we still cap concurrency and sleep between metadata calls.6"""7from __future__ import annotations89import hashlib10import random11import time12from dataclasses import dataclass13from datetime import date, datetime14from pathlib import Path15from typing import Any1617import httpx18from findata_core.redact import redact1920BASE_URL = "https://firstratedata.com/api"21FUNDAMENTALS_BASE_URL = "https://firstratedata.com/api-f"22CHUNK = 4 * 1024 * 102423MAX_RETRIES = 524TIMEOUT_META = 60.025TIMEOUT_DOWNLOAD = 7200.026DEFAULT_TRANSPORT: httpx.BaseTransport | None = None # tests inject a MockTransport here272829class FRDError(RuntimeError):30 pass313233@dataclass34class Downloaded:35 path: Path36 sha256: str37 bytes: int38 final_url_host: str39 fetched_at: datetime404142class FRDClient:43 def __init__(self, userid: str, base_url: str = BASE_URL, min_interval_s: float = 1.0,44 transport: httpx.BaseTransport | None = None):45 if not userid:46 raise FRDError("FRD_USERID is not configured")47 self._userid = userid48 self.base_url = base_url.rstrip("/")49 self.min_interval_s = min_interval_s50 self._last_call = 0.051 self._client = httpx.Client(follow_redirects=True, timeout=TIMEOUT_META, transport=transport or DEFAULT_TRANSPORT,52 headers={"User-Agent": "FinDataOS/0.1 (+private research lakehouse)"})5354 # ----------------------------------------------------------------- helpers55 def _url(self, endpoint: str, params: dict[str, Any]) -> str:56 q = {k: str(v) for k, v in params.items() if v is not None}57 q["userid"] = self._userid58 return str(httpx.URL(f"{self.base_url}/{endpoint}", params=q))5960 def sanitized_url(self, endpoint: str, params: dict[str, Any]) -> str:61 q = {k: str(v) for k, v in params.items() if v is not None}62 return str(httpx.URL(f"{self.base_url}/{endpoint}", params=q)) + "&userid=<REDACTED>"6364 def _throttle(self) -> None:65 wait = self.min_interval_s - (time.monotonic() - self._last_call)66 if wait > 0:67 time.sleep(wait)68 self._last_call = time.monotonic()6970 def _redact(self, text: Any) -> str:71 return redact(text, [self._userid])7273 def _get_text(self, endpoint: str, params: dict[str, Any]) -> str:74 last: Exception | None = None75 for attempt in range(1, MAX_RETRIES + 1):76 self._throttle()77 try:78 r = self._client.get(self._url(endpoint, params))79 if r.status_code in (429, 500, 502, 503, 504):80 raise FRDError(f"HTTP {r.status_code} from {endpoint}")81 if r.status_code != 200:82 raise FRDError(f"HTTP {r.status_code} from {endpoint}: {self._redact(r.text[:200])}")83 return r.text84 except (httpx.TransportError, FRDError) as e: # retry85 last = e86 if isinstance(e, FRDError) and not any(c in str(e) for c in ("429", "500", "502", "503", "504")):87 raise FRDError(self._redact(e)) from None88 time.sleep(min(60, 2 ** attempt) + random.uniform(0, 1))89 raise FRDError(self._redact(f"{endpoint} failed after {MAX_RETRIES} attempts: {last}"))9091 # ----------------------------------------------------------------- metadata92 def last_update(self, type_: str, full: bool = False) -> date:93 txt = self._get_text("last_update", {"type": type_, "is_full_update": "true" if full else None}).strip()94 for fmt in ("%Y-%m-%d", "%Y%m%d"):95 try:96 return datetime.strptime(txt[:10] if fmt == "%Y-%m-%d" else txt[:8], fmt).date()97 except ValueError:98 continue99 raise FRDError(f"unparseable last_update for {type_}: {self._redact(txt[:40])!r}")100101 def ticker_listing(self, type_: str) -> str:102 return self._get_text("ticker_listing", {"type": type_})103104 def update_log(self, type_: str, log_type: str) -> str:105 return self._get_text("update_log", {"type": type_, "log_type": log_type})106107 # ----------------------------------------------------------------- archives108 def download(self, endpoint: str, params: dict[str, Any], dest: Path) -> Downloaded:109 """Stream an archive to `dest` (atomic: `.part` then rename), computing SHA-256 on the fly."""110 dest.parent.mkdir(parents=True, exist_ok=True)111 part = dest.with_suffix(dest.suffix + ".part")112 last: Exception | None = None113 for attempt in range(1, MAX_RETRIES + 1):114 self._throttle()115 h = hashlib.sha256()116 size = 0117 try:118 with self._client.stream("GET", self._url(endpoint, params), timeout=TIMEOUT_DOWNLOAD) as r:119 if r.status_code != 200:120 body = r.read()[:300]121 raise FRDError(f"HTTP {r.status_code} downloading {endpoint}: {self._redact(body)}")122 ctype = r.headers.get("content-type", "")123 if "text/html" in ctype:124 body = r.read()[:300]125 raise FRDError(f"HTML instead of archive from {endpoint}: {self._redact(body)}")126 with open(part, "wb") as f:127 for chunk in r.iter_bytes(CHUNK):128 f.write(chunk)129 h.update(chunk)130 size += len(chunk)131 host = r.url.host132 part.replace(dest)133 return Downloaded(dest, h.hexdigest(), size, host, datetime.now().astimezone())134 except (httpx.TransportError, FRDError, OSError) as e:135 last = e136 part.unlink(missing_ok=True)137 if isinstance(e, FRDError) and "HTTP 4" in str(e):138 raise FRDError(self._redact(e)) from None139 time.sleep(min(120, 5 * 2 ** attempt) + random.uniform(0, 2))140 raise FRDError(self._redact(f"download {endpoint} failed after {MAX_RETRIES} attempts: {last}"))141142 def close(self) -> None:143 self._client.close()144