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 with connection() as conn:21 if p.get("all_active"):22 tickers = [r["symbol"] for r in conn.execute("SELECT symbol FROM instruments WHERE asset_class='stock' AND status='active' ORDER BY symbol").fetchall()]23 elif p.get("tracked"):24 row = conn.execute("SELECT value FROM settings_kv WHERE key='fundamentals_tracked_tickers'").fetchone()25 tickers = list(row["value"]) if row and row["value"] else [r["symbol"] for r in conn.execute("SELECT DISTINCT split_part(entity_id, ':', 2) symbol FROM fundamentals_facts WHERE entity_id LIKE 'frd:%'").fetchall()]26 else:27 n = int(p.get("limit", 50))28 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()]29 fr = FirstRateFundamentalsConnector()30 observed = datetime.now(UTC)31 run_id = job.get("run_id")32 stats = {"tickers": 0, "facts_new": 0, "facts_unchanged": 0, "facts_restated": 0, "earnings": 0, "errors": 0, "pit_fallback": 0}33 ck = queue.get_checkpoint(f"job:{job['id']}", "state") or {}34 done = set(ck.get("done") or [])35 no_data = set(ck.get("no_data") or [])36 for tk in tickers:37 if tk in done:38 continue39 try:40 facts = []41 for period in ("quarterly", "annual"):42 payload = fr.financials_all(tk, period)43 facts.extend(facts_from_financials(payload, observed, tk))44 earnings = fr.earnings(tk)45 except Exception as e:46 stats["errors"] += 147 log.warning("fundamentals.error", ticker=tk, error=str(e)[:200])48 continue49 with transaction() as conn:50 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) "51 "ON CONFLICT (entity_id) DO UPDATE SET tickers=EXCLUDED.tickers, updated_at=now()", (f"frd:{tk}", tk, [tk]))52 for f in facts:53 if f.pop("pit_fallback", False):54 stats["pit_fallback"] += 155 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",56 (f["entity_id"], f["metric"], f["fiscal_period_end"], f["fiscal_period_type"], f["source_id"])).fetchone()57 if cur is not None:58 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"])))59 if same:60 stats["facts_unchanged"] += 161 continue62 conn.execute("UPDATE fundamentals_facts SET superseded_at=%s WHERE id=%s", (observed, cur["id"]))63 rev = int(cur["revision_id"]) + 164 restated = True65 stats["facts_restated"] += 166 else:67 rev, restated = 1, False68 stats["facts_new"] += 169 conn.execute(70 "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, "71 "filing_date, published_at, available_at, observed_at, source_document, source_id, revision_id, restatement_flag, ingest_run_id) "72 "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,"73 "%(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)",74 {**f, "rev": rev, "restated": restated, "run_id": run_id})75 for e in earnings:76 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') "77 "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()",78 (f"frd:{tk}", e["date"], e.get("eps_estimate"), e.get("eps_reported"), e.get("release_time")))79 stats["earnings"] += len(earnings)80 stats["tickers"] += 181 done.add(tk)82 queue.set_checkpoint(f"job:{job['id']}", "state", {"done": sorted(done), "no_data": sorted(no_data)})83 queue.progress(job["id"], {**stats, "last": tk})84 fr.close()85 stats["no_statements_404"] = len(no_data)86 log.info("fundamentals.done", **stats)87 return stats88