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_financials1112from findata_ingest import queue13from findata_ingest.lake import Lake141516def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:17 p = job.get("params") or {}18 tickers = [t.upper() for t in (p.get("tickers") or [])]19 if not tickers:20 n = int(p.get("limit", 50))21 with connection() as conn:22 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()]23 fr = FirstRateFundamentalsConnector()24 observed = datetime.now(UTC)25 run_id = job.get("run_id")26 stats = {"tickers": 0, "facts_new": 0, "facts_unchanged": 0, "facts_restated": 0, "earnings": 0, "errors": 0, "pit_fallback": 0}27 ck = queue.get_checkpoint(f"job:{job['id']}", "state") or {}28 done = set(ck.get("done") or [])29 for tk in tickers:30 if tk in done:31 continue32 try:33 facts = []34 for period in ("quarterly", "annual"):35 payload = fr.financials_all(tk, period)36 facts.extend(facts_from_financials(payload, observed, tk))37 earnings = fr.earnings(tk)38 except Exception as e:39 stats["errors"] += 140 log.warning("fundamentals.error", ticker=tk, error=str(e)[:200])41 continue42 with transaction() as conn:43 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) "44 "ON CONFLICT (entity_id) DO UPDATE SET tickers=EXCLUDED.tickers, updated_at=now()", (f"frd:{tk}", tk, [tk]))45 for f in facts:46 if f.pop("pit_fallback", False):47 stats["pit_fallback"] += 148 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",49 (f["entity_id"], f["metric"], f["fiscal_period_end"], f["fiscal_period_type"], f["source_id"])).fetchone()50 if cur is not None:51 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"])))52 if same:53 stats["facts_unchanged"] += 154 continue55 conn.execute("UPDATE fundamentals_facts SET superseded_at=%s WHERE id=%s", (observed, cur["id"]))56 rev = int(cur["revision_id"]) + 157 restated = True58 stats["facts_restated"] += 159 else:60 rev, restated = 1, False61 stats["facts_new"] += 162 conn.execute(63 "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, "64 "filing_date, published_at, available_at, observed_at, source_document, source_id, revision_id, restatement_flag, ingest_run_id) "65 "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,"66 "%(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)",67 {**f, "rev": rev, "restated": restated, "run_id": run_id})68 for e in earnings:69 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') "70 "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()",71 (f"frd:{tk}", e["date"], e.get("eps_estimate"), e.get("eps_reported"), e.get("release_time")))72 stats["earnings"] += len(earnings)73 stats["tickers"] += 174 done.add(tk)75 queue.set_checkpoint(f"job:{job['id']}", "state", {"done": sorted(done)})76 queue.progress(job["id"], {**stats, "last": tk})77 fr.close()78 log.info("fundamentals.done", **stats)79 return stats80