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 for tk in tickers:36 if tk in done:37 continue38 try:39 facts = []40 for period in ("quarterly", "annual"):41 payload = fr.financials_all(tk, period)42 facts.extend(facts_from_financials(payload, observed, tk))43 earnings = fr.earnings(tk)44 except Exception as e:45 stats["errors"] += 146 log.warning("fundamentals.error", ticker=tk, error=str(e)[:200])47 continue48 with transaction() as conn:49 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) "50 "ON CONFLICT (entity_id) DO UPDATE SET tickers=EXCLUDED.tickers, updated_at=now()", (f"frd:{tk}", tk, [tk]))51 for f in facts:52 if f.pop("pit_fallback", False):53 stats["pit_fallback"] += 154 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",55 (f["entity_id"], f["metric"], f["fiscal_period_end"], f["fiscal_period_type"], f["source_id"])).fetchone()56 if cur is not None:57 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"])))58 if same:59 stats["facts_unchanged"] += 160 continue61 conn.execute("UPDATE fundamentals_facts SET superseded_at=%s WHERE id=%s", (observed, cur["id"]))62 rev = int(cur["revision_id"]) + 163 restated = True64 stats["facts_restated"] += 165 else:66 rev, restated = 1, False67 stats["facts_new"] += 168 conn.execute(69 "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, "70 "filing_date, published_at, available_at, observed_at, source_document, source_id, revision_id, restatement_flag, ingest_run_id) "71 "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,"72 "%(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)",73 {**f, "rev": rev, "restated": restated, "run_id": run_id})74 for e in earnings:75 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') "76 "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()",77 (f"frd:{tk}", e["date"], e.get("eps_estimate"), e.get("eps_reported"), e.get("release_time")))78 stats["earnings"] += len(earnings)79 stats["tickers"] += 180 done.add(tk)81 queue.set_checkpoint(f"job:{job['id']}", "state", {"done": sorted(done)})82 queue.progress(job["id"], {**stats, "last": tk})83 fr.close()84 log.info("fundamentals.done", **stats)85 return stats86