1"""Reference data: ticker listings, delisted log, splits and dividends → instruments / corporate_actions."""2from __future__ import annotations34from typing import Any56from findata_core.db import transaction7from findata_market_firstrate import FirstRateMarketConnector89from findata_ingest.lake import Lake10from findata_ingest.tasks.common import record_raw_archive111213def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:14 params = job.get("params") or {}15 types = params.get("types") or ["stock", "etf"]16 conn_fr = FirstRateMarketConnector()17 stats: dict[str, Any] = {}18 for t in types:19 rows = conn_fr.ticker_listing(t)20 with transaction() as conn:21 for r in rows:22 iid = f"firstrate:{t}:{r['symbol']}"23 conn.execute(24 "INSERT INTO instruments(instrument_id, symbol, name, asset_class, venue, currency, first_date, last_date, status, source_id) "25 "VALUES (%s,%s,%s,%s,NULL,'USD',%s,%s,%s,'firstrate') ON CONFLICT (instrument_id) DO UPDATE SET name=EXCLUDED.name, "26 "first_date=EXCLUDED.first_date, last_date=EXCLUDED.last_date, status=EXCLUDED.status, updated_at=now()",27 (iid, r["symbol"], r["name"], t, r["first_date"], r["last_date"], r["status"]))28 if r["symbol"].endswith("-DELISTED"):29 base = r["symbol"][: -len("-DELISTED")]30 conn.execute("INSERT INTO instrument_aliases(instrument_id, symbol, valid_from, valid_to, source_id, note) VALUES (%s,%s,%s,%s,'firstrate','vendor -DELISTED suffix') "31 "ON CONFLICT DO NOTHING", (iid, base, r["first_date"], r["last_date"]))32 stats[f"instruments_{t}"] = len(rows)33 log.info("metadata.listing", type=t, rows=len(rows))34 if "stock" in types:35 dl = conn_fr.delisted_log()36 with transaction() as conn:37 for r in dl:38 conn.execute("UPDATE instruments SET status='delisted', last_date=COALESCE(%s, last_date), updated_at=now() WHERE instrument_id=%s",39 (r["last_date"], f"firstrate:stock:{r['symbol']}"))40 stats["delisted_log"] = len(dl)41 for kind in ("splits", "dividends"):42 res, rows = conn_fr.meta_actions("stock", kind, lake.raw_dir("firstrate", "corporate_actions.stock"))43 with transaction() as conn:44 record_raw_archive(conn, "firstrate", "corporate_actions.stock", {"endpoint": "meta_file", "type": "stock", "metafile_type": kind},45 str(res.path.relative_to(lake.root)), res.sha256, res.bytes, res.file_count, res.vendor_last_update, job.get("run_id"))46 n = 047 for r in rows:48 conn.execute("INSERT INTO corporate_actions(instrument_id, action_type, ex_date, value, source_id, ingest_run_id) VALUES (%s,%s,%s,%s,'firstrate',%s) ON CONFLICT DO NOTHING",49 (f"firstrate:stock:{r['symbol']}", r["action_type"], r["ex_date"], r["value"], job.get("run_id")))50 n += 151 stats[kind] = n52 log.info("metadata.actions", kind=kind, rows=n)53 return stats54