"""FirstRate Data US Fundamentals (`/api-f/v1`). Verified 2026-11-01: auth via `userid` query parameter (the page advertises `api_key` but that returns 403). JSON envelope `{status, data:{ticker, count, financial_data:[...]}}`. Point-in-time policy -------------------- The vendor supplies `filing_date` (SEC filing date) and one value per (ticker, period): no revision history. FinData sets `available_at = filing_date 00:00 America/New_York + 1 day` (the filing is certainly public on the next trading morning; EDGAR acceptance after 17:30 ET is disseminated the next business day). `published_at` is unknown → NULL. Each ingestion run compares the vendor value with the latest stored revision; a change creates a new revision with `restatement_flag = true` and supersedes the old one. Pre-ingestion restatements cannot be reconstructed from this vendor (documented limitation). """ from __future__ import annotations import json from collections.abc import Iterator from datetime import UTC, date, datetime, time, timedelta from typing import Any from zoneinfo import ZoneInfo import httpx from findata_core.config import get_settings from findata_core.redact import redact BASE = "https://firstratedata.com/api-f" SOURCE_ID = "firstrate_fundamentals" ET = ZoneInfo("America/New_York") DEFAULT_TRANSPORT: httpx.BaseTransport | None = None # metric -> (unit, kind) kind: flow|stock|per_share|ratio|count METRICS: dict[str, tuple[str, str]] = { "rev": ("USD", "flow"), "cogs": ("USD", "flow"), "gp": ("USD", "flow"), "sgna": ("USD", "flow"), "research_dev": ("USD", "flow"), "op_ex": ("USD", "flow"), "op_inc": ("USD", "flow"), "interest_exp": ("USD", "flow"), "ebt": ("USD", "flow"), "tax_expense": ("USD", "flow"), "net_inc": ("USD", "flow"), "net_inc_com": ("USD", "flow"), "net_inc_non_control": ("USD", "flow"), "net_loss_discontinued": ("USD", "flow"), "cos_inc": ("USD", "flow"), "ebit": ("USD", "flow"), "ebitda": ("USD", "flow"), "depr": ("USD", "flow"), "share_comp": ("USD", "flow"), "pref_divs": ("USD", "flow"), "eps": ("USD/share", "per_share"), "eps_d": ("USD/share", "per_share"), "dps": ("USD/share", "per_share"), "shares": ("shares", "count"), "shares_av": ("shares", "count"), "shares_av_diluted": ("shares", "count"), "assets_tot": ("USD", "stock"), "assets_c": ("USD", "stock"), "assets_nc": ("USD", "stock"), "cash": ("USD", "stock"), "receivables": ("USD", "stock"), "inventory": ("USD", "stock"), "inv_c": ("USD", "stock"), "inv_nc": ("USD", "stock"), "investments": ("USD", "stock"), "ppe": ("USD", "stock"), "intangibles": ("USD", "stock"), "tax_asset": ("USD", "stock"), "liab_tot": ("USD", "stock"), "liab_c": ("USD", "stock"), "liab_nc": ("USD", "stock"), "payables": ("USD", "stock"), "def_rev": ("USD", "stock"), "debt": ("USD", "stock"), "debt_c": ("USD", "stock"), "debt_nc": ("USD", "stock"), "tax_liability": ("USD", "stock"), "equity": ("USD", "stock"), "ret_earnings": ("USD", "stock"), "acc_oci": ("USD", "stock"), "cash_op": ("USD", "flow"), "cash_inv": ("USD", "flow"), "cash_fin": ("USD", "flow"), "capex": ("USD", "flow"), "cash_div": ("USD", "flow"), "cash_change": ("USD", "flow"), "cash_business_acq": ("USD", "flow"), "cash_inv_acq": ("USD", "flow"), "cash_debt_sec": ("USD", "flow"), "cash_equity": ("USD", "flow"), "fx_impact_cash": ("USD", "flow"), "roa": ("ratio", "ratio"), "roe": ("ratio", "ratio"), "roic": ("ratio", "ratio"), "gm": ("ratio", "ratio"), "om": ("ratio", "ratio"), "intr_cover": ("ratio", "ratio"), } NON_METRIC_FIELDS = {"period_end", "period_type", "filing_date", "fiscal_year", "fiscal_period", "actual_year", "actual_period"} def available_at_policy(filing_date: date | None, period_end: date) -> datetime: """PIT availability. Falls back to period_end + 90 days when the filing date is missing (flagged upstream).""" base = filing_date if filing_date else period_end + timedelta(days=90) return datetime.combine(base + timedelta(days=1), time(0, 0), tzinfo=ET).astimezone(UTC) def facts_from_financials(payload: dict[str, Any], observed_at: datetime, ticker: str | None = None) -> Iterator[dict[str, Any]]: """Explode one `/v1/financials/all` or `/single` JSON payload into long-format facts (dicts).""" data = payload.get("data") or {} tkr = (data.get("ticker") or ticker or "").upper() records = data.get("financial_data") or ([data] if "period_end" in data else []) for rec in records: pe = date.fromisoformat(rec["period_end"][:10]) fd = date.fromisoformat(rec["filing_date"][:10]) if rec.get("filing_date") else None ptype = "a" if str(rec.get("period_type", "")).lower().startswith("a") or str(rec.get("fiscal_period", "")).lower() == "fy" else "q" avail = available_at_policy(fd, pe) start = None if ptype == "q": start = (pe.replace(day=1) - timedelta(days=60)).replace(day=1) # approx. quarter start (vendor gives no start) else: start = (pe - timedelta(days=364)).replace(day=1) for k, v in rec.items(): if k in NON_METRIC_FIELDS or k not in METRICS: continue unit, _kind = METRICS[k] yield { "entity_id": f"frd:{tkr}", "security_id": f"firstrate:stock:{tkr}", "metric": k, "value": None if v is None else float(v), "unit": unit, "currency": "USD" if unit.startswith("USD") else None, "fiscal_period_start": start, "fiscal_period_end": pe, "fiscal_period_type": ptype, "fiscal_year": rec.get("fiscal_year"), "fiscal_period": str(rec.get("fiscal_period") or "").lower() or None, "filing_date": fd, "published_at": None, "available_at": avail, "observed_at": observed_at, "source_document": f"SEC {'10-K' if ptype == 'a' else '10-Q'} via FirstRate (filing_date={fd})", "source_id": SOURCE_ID, "pit_fallback": fd is None, } class FirstRateFundamentalsConnector: source_id = SOURCE_ID def __init__(self, userid: str | None = None, transport: httpx.BaseTransport | None = None): self._userid = userid if userid is not None else get_settings().frd_userid.get_secret_value() self._client = httpx.Client(timeout=60.0, transport=transport or DEFAULT_TRANSPORT, headers={"User-Agent": "FinDataOS/0.1 (+private research lakehouse)"}) def _get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]: if not self._userid: raise RuntimeError("FRD_USERID not configured") q = dict(params or {}) q["type"] = "json" q["userid"] = self._userid r = self._client.get(f"{BASE}{path}", params=q) if r.status_code != 200: raise RuntimeError(redact(f"fundamentals HTTP {r.status_code} on {path}: {r.text[:200]}", [self._userid])) try: payload = r.json() except json.JSONDecodeError as e: raise RuntimeError(redact(f"fundamentals: non-JSON response on {path}", [self._userid])) from e if payload.get("status") not in (None, "success"): raise RuntimeError(redact(f"fundamentals: status={payload.get('status')} on {path}", [self._userid])) return payload def discover(self) -> dict[str, Any]: p = self._get("/v1/financials/all/AAPL/annual") return {"source_id": SOURCE_ID, "probe": "financials/all/AAPL/annual", "count": p.get("data", {}).get("count"), "fields": sorted((p.get("data", {}).get("financial_data") or [{}])[0].keys())} def financials_all(self, ticker: str, period: str = "quarterly") -> dict[str, Any]: assert period in ("annual", "quarterly") return self._get(f"/v1/financials/all/{ticker.upper()}/{period}") def financials_single(self, ticker: str, period: str, year: int) -> dict[str, Any]: return self._get(f"/v1/financials/single/{ticker.upper()}", {"period": period, "year": year}) def earnings(self, ticker: str) -> list[dict[str, Any]]: p = self._get(f"/v1/earnings_release/t/{ticker.upper()}") return p.get("data", {}).get("earnings_release_data") or [] def metrics(self, ticker: str, start: date, end: date) -> list[dict[str, Any]]: p = self._get(f"/v1/metrics/t/{ticker.upper()}/{start.isoformat()}/{end.isoformat()}") return p.get("data", {}).get("metrics_data") or [] def close(self) -> None: self._client.close()