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 httpx1819from findata_core.redact import redact2021BASE_URL = "https://firstratedata.com/api"22FUNDAMENTALS_BASE_URL = "https://firstratedata.com/api-f"23CHUNK = 4 * 1024 * 102424MAX_RETRIES = 525TIMEOUT_META = 60.026TIMEOUT_DOWNLOAD = 7200.027DEFAULT_TRANSPORT: httpx.BaseTransport | None = None # tests inject a MockTransport here282930class FRDError(RuntimeError):31 pass323334@dataclass35class Downloaded:36 path: Path37 sha256: str38 bytes: int39 final_url_host: str40 fetched_at: datetime414243class FRDClient:44 def __init__(self, userid: str, base_url: str = BASE_URL, min_interval_s: float = 1.0,45 transport: httpx.BaseTransport | None = None):46 if not userid:47 raise FRDError("FRD_USERID is not configured")48 self._userid = userid49 self.base_url = base_url.rstrip("/")50 self.min_interval_s = min_interval_s51 self._last_call = 0.052 self._client = httpx.Client(follow_redirects=True, timeout=TIMEOUT_META, transport=transport or DEFAULT_TRANSPORT,53 headers={"User-Agent": "FinDataOS/0.1 (+private research lakehouse)"})5455 # ----------------------------------------------------------------- helpers56 def _url(self, endpoint: str, params: dict[str, Any]) -> str:57 q = {k: str(v) for k, v in params.items() if v is not None}58 q["userid"] = self._userid59 return str(httpx.URL(f"{self.base_url}/{endpoint}", params=q))6061 def sanitized_url(self, endpoint: str, params: dict[str, Any]) -> str:62 q = {k: str(v) for k, v in params.items() if v is not None}63 return str(httpx.URL(f"{self.base_url}/{endpoint}", params=q)) + "&userid=<REDACTED>"6465 def _throttle(self) -> None:66 wait = self.min_interval_s - (time.monotonic() - self._last_call)67 if wait > 0:68 time.sleep(wait)69 self._last_call = time.monotonic()7071 def _redact(self, text: Any) -> str:72 return redact(text, [self._userid])7374 def _get_text(self, endpoint: str, params: dict[str, Any]) -> str:75 last: Exception | None = None76 for attempt in range(1, MAX_RETRIES + 1):77 self._throttle()78 try:79 r = self._client.get(self._url(endpoint, params))80 if r.status_code in (429, 500, 502, 503, 504):81 raise FRDError(f"HTTP {r.status_code} from {endpoint}")82 if r.status_code != 200:83 raise FRDError(f"HTTP {r.status_code} from {endpoint}: {self._redact(r.text[:200])}")84 return r.text85 except (httpx.TransportError, FRDError) as e: # retry86 last = e87 if isinstance(e, FRDError) and not any(c in str(e) for c in ("429", "500", "502", "503", "504")):88 raise FRDError(self._redact(e)) from None89 time.sleep(min(60, 2 ** attempt) + random.uniform(0, 1))90 raise FRDError(self._redact(f"{endpoint} failed after {MAX_RETRIES} attempts: {last}"))9192 # ----------------------------------------------------------------- metadata93 def last_update(self, type_: str, full: bool = False) -> date:94 txt = self._get_text("last_update", {"type": type_, "is_full_update": "true" if full else None}).strip()95 for fmt in ("%Y-%m-%d", "%Y%m%d"):96 try:97 return datetime.strptime(txt[:10] if fmt == "%Y-%m-%d" else txt[:8], fmt).date()98 except ValueError:99 continue100 raise FRDError(f"unparseable last_update for {type_}: {self._redact(txt[:40])!r}")101102 def ticker_listing(self, type_: str) -> str:103 return self._get_text("ticker_listing", {"type": type_})104105 def update_log(self, type_: str, log_type: str) -> str:106 return self._get_text("update_log", {"type": type_, "log_type": log_type})107108 # ----------------------------------------------------------------- archives109 def download(self, endpoint: str, params: dict[str, Any], dest: Path) -> Downloaded:110 """Stream an archive to `dest` (atomic: `.part` then rename), computing SHA-256 on the fly."""111 dest.parent.mkdir(parents=True, exist_ok=True)112 part = dest.with_suffix(dest.suffix + ".part")113 last: Exception | None = None114 for attempt in range(1, MAX_RETRIES + 1):115 self._throttle()116 h = hashlib.sha256()117 size = 0118 try:119 with self._client.stream("GET", self._url(endpoint, params), timeout=TIMEOUT_DOWNLOAD) as r:120 if r.status_code != 200:121 body = r.read()[:300]122 raise FRDError(f"HTTP {r.status_code} downloading {endpoint}: {self._redact(body)}")123 ctype = r.headers.get("content-type", "")124 if "text/html" in ctype:125 body = r.read()[:300]126 raise FRDError(f"HTML instead of archive from {endpoint}: {self._redact(body)}")127 with open(part, "wb") as f:128 for chunk in r.iter_bytes(CHUNK):129 f.write(chunk)130 h.update(chunk)131 size += len(chunk)132 host = r.url.host133 part.replace(dest)134 return Downloaded(dest, h.hexdigest(), size, host, datetime.now().astimezone())135 except (httpx.TransportError, FRDError, OSError) as e:136 last = e137 part.unlink(missing_ok=True)138 if isinstance(e, FRDError) and "HTTP 4" in str(e):139 raise FRDError(self._redact(e)) from None140 time.sleep(min(120, 5 * 2 ** attempt) + random.uniform(0, 2))141 raise FRDError(self._redact(f"download {endpoint} failed after {MAX_RETRIES} attempts: {last}"))142143 def close(self) -> None:144 self._client.close()145