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_ingest.lake import Lake8from findata_ingest.tasks.common import record_raw_archive9from findata_market_firstrate import FirstRateMarketConnector101112def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:13 params = job.get("params") or {}14 types = params.get("types") or ["stock", "etf"]15 conn_fr = FirstRateMarketConnector()16 stats: dict[str, Any] = {}17 for t in types:18 rows = conn_fr.ticker_listing(t)19 with transaction() as conn:20 for r in rows:21 iid = f"firstrate:{t}:{r['symbol']}"22 conn.execute(23 "INSERT INTO instruments(instrument_id, symbol, name, asset_class, venue, currency, first_date, last_date, status, source_id) "24 "VALUES (%s,%s,%s,%s,NULL,'USD',%s,%s,%s,'firstrate') ON CONFLICT (instrument_id) DO UPDATE SET name=EXCLUDED.name, "25 "first_date=EXCLUDED.first_date, last_date=EXCLUDED.last_date, status=EXCLUDED.status, updated_at=now()",26 (iid, r["symbol"], r["name"], t, r["first_date"], r["last_date"], r["status"]))27 if r["symbol"].endswith("-DELISTED"):28 base = r["symbol"][: -len("-DELISTED")]29 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') "30 "ON CONFLICT DO NOTHING", (iid, base, r["first_date"], r["last_date"]))31 stats[f"instruments_{t}"] = len(rows)32 log.info("metadata.listing", type=t, rows=len(rows))33 if "stock" in types:34 dl = conn_fr.delisted_log()35 with transaction() as conn:36 for r in dl:37 conn.execute("UPDATE instruments SET status='delisted', last_date=COALESCE(%s, last_date), updated_at=now() WHERE instrument_id=%s",38 (r["last_date"], f"firstrate:stock:{r['symbol']}"))39 stats["delisted_log"] = len(dl)40 for kind in ("splits", "dividends"):41 res, rows = conn_fr.meta_actions("stock", kind, lake.raw_dir("firstrate", "corporate_actions.stock"))42 with transaction() as conn:43 record_raw_archive(conn, "firstrate", "corporate_actions.stock", {"endpoint": "meta_file", "type": "stock", "metafile_type": kind},44 str(res.path.relative_to(lake.root)), res.sha256, res.bytes, res.file_count, res.vendor_last_update, job.get("run_id"))45 n = 046 for r in rows:47 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",48 (f"firstrate:stock:{r['symbol']}", r["action_type"], r["ex_date"], r["value"], job.get("run_id")))49 n += 150 stats[kind] = n51 log.info("metadata.actions", kind=kind, rows=n)52 return stats53