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 pass313233class FRDNoData(FRDError):34 """The vendor answered with a tiny text body instead of an archive (e.g. 'no update this period')."""353637@dataclass38class Downloaded:39 path: Path40 sha256: str41 bytes: int42 final_url_host: str43 fetched_at: datetime444546class FRDClient:47 def __init__(self, userid: str, base_url: str = BASE_URL, min_interval_s: float = 1.0,48 transport: httpx.BaseTransport | None = None):49 if not userid:50 raise FRDError("FRD_USERID is not configured")51 self._userid = userid52 self.base_url = base_url.rstrip("/")53 self.min_interval_s = min_interval_s54 self._last_call = 0.055 self._client = httpx.Client(follow_redirects=True, timeout=TIMEOUT_META, transport=transport or DEFAULT_TRANSPORT,56 headers={"User-Agent": "FinDataOS/0.1 (+private research lakehouse)"})5758 # ----------------------------------------------------------------- helpers59 def _url(self, endpoint: str, params: dict[str, Any]) -> str:60 q = {k: str(v) for k, v in params.items() if v is not None}61 q["userid"] = self._userid62 return str(httpx.URL(f"{self.base_url}/{endpoint}", params=q))6364 def sanitized_url(self, endpoint: str, params: dict[str, Any]) -> str:65 q = {k: str(v) for k, v in params.items() if v is not None}66 return str(httpx.URL(f"{self.base_url}/{endpoint}", params=q)) + "&userid=<REDACTED>"6768 def _throttle(self) -> None:69 wait = self.min_interval_s - (time.monotonic() - self._last_call)70 if wait > 0:71 time.sleep(wait)72 self._last_call = time.monotonic()7374 def _redact(self, text: Any) -> str:75 return redact(text, [self._userid])7677 def _get_text(self, endpoint: str, params: dict[str, Any]) -> str:78 last: Exception | None = None79 for attempt in range(1, MAX_RETRIES + 1):80 self._throttle()81 try:82 r = self._client.get(self._url(endpoint, params))83 if r.status_code in (429, 500, 502, 503, 504):84 raise FRDError(f"HTTP {r.status_code} from {endpoint}")85 if r.status_code != 200:86 raise FRDError(f"HTTP {r.status_code} from {endpoint}: {self._redact(r.text[:200])}")87 return r.text88 except (httpx.TransportError, FRDError) as e: # retry89 last = e90 if isinstance(e, FRDError) and not any(c in str(e) for c in ("429", "500", "502", "503", "504")):91 raise FRDError(self._redact(e)) from None92 time.sleep(min(60, 2 ** attempt) + random.uniform(0, 1))93 raise FRDError(self._redact(f"{endpoint} failed after {MAX_RETRIES} attempts: {last}"))9495 # ----------------------------------------------------------------- metadata96 def last_update(self, type_: str, full: bool = False) -> date:97 txt = self._get_text("last_update", {"type": type_, "is_full_update": "true" if full else None}).strip()98 for fmt in ("%Y-%m-%d", "%Y%m%d"):99 try:100 return datetime.strptime(txt[:10] if fmt == "%Y-%m-%d" else txt[:8], fmt).date()101 except ValueError:102 continue103 raise FRDError(f"unparseable last_update for {type_}: {self._redact(txt[:40])!r}")104105 def ticker_listing(self, type_: str) -> str:106 return self._get_text("ticker_listing", {"type": type_})107108 def update_log(self, type_: str, log_type: str) -> str:109 return self._get_text("update_log", {"type": type_, "log_type": log_type})110111 # ----------------------------------------------------------------- archives112 def download(self, endpoint: str, params: dict[str, Any], dest: Path) -> Downloaded:113 """Stream an archive to `dest` (atomic: `.part` then rename), computing SHA-256 on the fly."""114 for i in range(20):115 try:116 dest.parent.mkdir(parents=True, exist_ok=True)117 break118 except InterruptedError:119 time.sleep(0.2 * (i + 1))120 part = dest.with_suffix(dest.suffix + ".part")121 last: Exception | None = None122 for attempt in range(1, MAX_RETRIES + 1):123 self._throttle()124 h = hashlib.sha256()125 size = 0126 try:127 with self._client.stream("GET", self._url(endpoint, params), timeout=TIMEOUT_DOWNLOAD) as r:128 if r.status_code != 200:129 body = r.read()[:300]130 raise FRDError(f"HTTP {r.status_code} downloading {endpoint}: {self._redact(body)}")131 ctype = r.headers.get("content-type", "")132 clen = int(r.headers.get("content-length") or 0)133 if "text/html" in ctype or (clen and clen < 512 and "zip" not in ctype):134 body = r.read()[:300]135 if clen and clen < 512:136 raise FRDNoData(f"no archive available for {endpoint}: {self._redact(body.decode(errors='replace').strip())}")137 raise FRDError(f"HTML instead of archive from {endpoint}: {self._redact(body)}")138 with open(part, "wb") as f:139 for chunk in r.iter_bytes(CHUNK):140 f.write(chunk)141 h.update(chunk)142 size += len(chunk)143 host = r.url.host144 part.replace(dest)145 return Downloaded(dest, h.hexdigest(), size, host, datetime.now().astimezone())146 except (httpx.TransportError, FRDError, OSError) as e:147 last = e148 part.unlink(missing_ok=True)149 if isinstance(e, FRDNoData):150 raise151 if isinstance(e, FRDError) and "HTTP 4" in str(e):152 raise FRDError(self._redact(e)) from None153 time.sleep(min(120, 5 * 2 ** attempt) + random.uniform(0, 2))154 raise FRDError(self._redact(f"download {endpoint} failed after {MAX_RETRIES} attempts: {last}"))155156 def close(self) -> None:157 self._client.close()158