1"""Fundamentals ingestion (FirstRate /api-f) with revision tracking. Params: tickers[] or universe (active stocks,2limit n). Each run: fetch quarterly + annual financials and earnings; compare with latest stored revision per3(entity, metric, period_end, period_type); unchanged → nothing; changed → new revision, restatement_flag, supersede."""4from __future__ import annotations56from datetime import UTC, datetime7from typing import Any89from findata_core.db import connection, transaction10from findata_fundamentals import FirstRateFundamentalsConnector, facts_from_financials11from findata_ingest import queue12from findata_ingest.lake import Lake131415def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:16 p = job.get("params") or {}17 tickers = [t.upper() for t in (p.get("tickers") or [])]18 if not tickers:19 n = int(p.get("limit", 50))20 with connection() as conn:21 tickers = [r["symbol"] for r in conn.execute("SELECT symbol FROM instruments WHERE asset_class='stock' AND status='active' ORDER BY symbol LIMIT %s", (n,)).fetchall()]22 fr = FirstRateFundamentalsConnector()23 observed = datetime.now(UTC)24 run_id = job.get("run_id")25 stats = {"tickers": 0, "facts_new": 0, "facts_unchanged": 0, "facts_restated": 0, "earnings": 0, "errors": 0, "pit_fallback": 0}26 ck = queue.get_checkpoint(f"job:{job['id']}", "state") or {}27 done = set(ck.get("done") or [])28 for tk in tickers:29 if tk in done:30 continue31 try:32 facts = []33 for period in ("quarterly", "annual"):34 payload = fr.financials_all(tk, period)35 facts.extend(facts_from_financials(payload, observed, tk))36 earnings = fr.earnings(tk)37 except Exception as e: # noqa: BLE00138 stats["errors"] += 139 log.warning("fundamentals.error", ticker=tk, error=str(e)[:200])40 continue41 with transaction() as conn:42 conn.execute("INSERT INTO entities(entity_id, name, tickers) VALUES (%s, (SELECT name FROM instruments WHERE symbol=%s AND asset_class='stock' LIMIT 1), %s) "43 "ON CONFLICT (entity_id) DO UPDATE SET tickers=EXCLUDED.tickers, updated_at=now()", (f"frd:{tk}", tk, [tk]))44 for f in facts:45 if f.pop("pit_fallback", False):46 stats["pit_fallback"] += 147 cur = conn.execute("SELECT id, value, revision_id FROM fundamentals_facts WHERE entity_id=%s AND metric=%s AND fiscal_period_end=%s AND fiscal_period_type=%s AND source_id=%s AND superseded_at IS NULL",48 (f["entity_id"], f["metric"], f["fiscal_period_end"], f["fiscal_period_type"], f["source_id"])).fetchone()49 if cur is not None:50 same = (cur["value"] is None and f["value"] is None) or (cur["value"] is not None and f["value"] is not None and abs(float(cur["value"]) - f["value"]) <= 1e-9 * max(1.0, abs(f["value"])))51 if same:52 stats["facts_unchanged"] += 153 continue54 conn.execute("UPDATE fundamentals_facts SET superseded_at=%s WHERE id=%s", (observed, cur["id"]))55 rev = int(cur["revision_id"]) + 156 restated = True57 stats["facts_restated"] += 158 else:59 rev, restated = 1, False60 stats["facts_new"] += 161 conn.execute(62 "INSERT INTO fundamentals_facts(entity_id, security_id, metric, value, unit, currency, fiscal_period_start, fiscal_period_end, fiscal_period_type, fiscal_year, fiscal_period, "63 "filing_date, published_at, available_at, observed_at, source_document, source_id, revision_id, restatement_flag, ingest_run_id) "64 "VALUES (%(entity_id)s,%(security_id)s,%(metric)s,%(value)s,%(unit)s,%(currency)s,%(fiscal_period_start)s,%(fiscal_period_end)s,%(fiscal_period_type)s,%(fiscal_year)s,%(fiscal_period)s,"65 "%(filing_date)s,%(published_at)s,%(available_at)s,%(observed_at)s,%(source_document)s,%(source_id)s,%(rev)s,%(restated)s,%(run_id)s)",66 {**f, "rev": rev, "restated": restated, "run_id": run_id})67 for e in earnings:68 conn.execute("INSERT INTO earnings_events(entity_id, event_date, eps_estimate, eps_reported, release_time, source_id) VALUES (%s,%s,%s,%s,%s,'firstrate_fundamentals') "69 "ON CONFLICT (entity_id, event_date, source_id) DO UPDATE SET eps_estimate=EXCLUDED.eps_estimate, eps_reported=EXCLUDED.eps_reported, release_time=EXCLUDED.release_time, observed_at=now()",70 (f"frd:{tk}", e["date"], e.get("eps_estimate"), e.get("eps_reported"), e.get("release_time")))71 stats["earnings"] += len(earnings)72 stats["tickers"] += 173 done.add(tk)74 queue.set_checkpoint(f"job:{job['id']}", "state", {"done": sorted(done)})75 queue.progress(job["id"], {**stats, "last": tk})76 fr.close()77 log.info("fundamentals.done", **stats)78 return stats79