1#!/usr/bin/env python2"""Phase 2a: universe audit — fundamentals entities x 1-minute coverage x SEC crosswalk; exclusions documented.34Inputs: node extracts produced by scripts/01a_extract_node_tables.sh (entities, earnings_events, corporate_actions —5read-only Postgres copies, documented exception), FinData /coverage/symbols and /instruments, SEC public files.6"""7from __future__ import annotations89import json10import sys11from pathlib import Path1213import polars as pl1415sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))16from wp39.api_client import FinDataClient # noqa: E40217from wp39.instruments import build_crosswalk, fetch_sec_submissions, fetch_sec_tickers # noqa: E40218from wp39.pipeline import Paths, get_logger, load_config # noqa: E40219from wp39.provenance import now_iso, sha256_file, write_json # noqa: E402202122def main() -> int:23 P = Paths.default()24 log = get_logger("01_audit", P.logs / "01_audit.log")25 cfg = load_config("research"); contract = load_config("data_contract")26 ext = P.cache / "external"27 ents = pl.read_csv(ext / "entities.csv")28 ents = ents.with_columns(pl.col("entity_id").str.replace("^frd:", "").alias("ticker"))29 log.info("fundamentals entities: %d", ents.height)3031 client = FinDataClient(log_path=P.logs / "api_calls.jsonl")32 cov = pl.DataFrame(list(client.coverage_symbols(contract["findata"]["datasets"]["bars_1m"], min_sessions=1)))33 cov = cov.rename({"symbol": "ticker", "sessions": "n_sessions_1m", "d0": "first_session_1m", "d1": "last_session_1m", "bars": "n_bars_1m"})34 cov = cov.with_columns([pl.col("first_session_1m").str.to_date(), pl.col("last_session_1m").str.to_date()])35 log.info("1m coverage rows: %d (incl. %d delisted)", cov.height, cov.filter(pl.col("ticker").str.ends_with("-DELISTED")).height)3637 sec = fetch_sec_tickers(ext)38 u = ents.join(cov, on="ticker", how="left")39 u = u.join(sec.unique(subset=["ticker"]), on="ticker", how="left")40 subs = fetch_sec_submissions(sorted(u["cik"].drop_nulls().unique().to_list()), ext, log=log)41 u = u.join(subs, on="cik", how="left")42 u = u.with_columns([43 pl.col("first_q").str.to_date(), pl.col("last_q").str.to_date(), pl.col("first_filing").str.to_date(),44 pl.col("cik").is_not_null().alias("has_cik"),45 pl.col("sic_division").fill_null("unknown"),46 ])47 # eligibility (frozen): >= min quarters, 1m coverage overlapping 2012+, SIC not public administration48 u = u.with_columns([49 (pl.col("n_q") >= cfg["sample"]["min_quarters_per_firm"]).alias("ok_quarters"),50 (pl.col("n_sessions_1m").fill_null(0) >= 250).alias("ok_coverage"),51 (pl.col("last_session_1m") >= pl.date(2012, 6, 30)).fill_null(False).alias("ok_period"),52 (pl.col("sic_division") != "J_public_admin").alias("ok_sic"),53 ])54 u = u.with_columns((pl.col("ok_quarters") & pl.col("ok_coverage") & pl.col("ok_period") & pl.col("ok_sic")).alias("eligible"))55 u.write_parquet(P.processed / "universe.parquet")56 cascade = [57 {"step": "fundamentals_entities", "n": u.height},58 {"step": ">= 8 quarterly periods", "n": u.filter(pl.col("ok_quarters")).height},59 {"step": "+ 1m bars (>= 250 sessions)", "n": u.filter(pl.col("ok_quarters") & pl.col("ok_coverage")).height},60 {"step": "+ bars after 2012-06", "n": u.filter(pl.col("ok_quarters") & pl.col("ok_coverage") & pl.col("ok_period")).height},61 {"step": "+ not public administration", "n": u.filter(pl.col("eligible")).height},62 ]63 rep = {64 "built_at": now_iso(), "entities": u.height, "eligible": u.filter(pl.col("eligible")).height,65 "with_cik": u.filter(pl.col("has_cik")).height, "with_sic": u.filter(pl.col("sic").is_not_null()).height,66 "sic_division_counts": u.filter(pl.col("eligible")).group_by("sic_division").len().sort("len", descending=True).to_dicts(),67 "first_q_year_counts": u.filter(pl.col("eligible")).with_columns(pl.col("first_q").dt.year().alias("y")).group_by("y").len().sort("y").to_dicts(),68 "exchanges": u.filter(pl.col("eligible")).explode("exchanges").group_by("exchanges").len().sort("len", descending=True).head(8).to_dicts(),69 "delisted_symbols_in_1m_lake": int(cov.filter(pl.col("ticker").str.ends_with("-DELISTED")).height),70 "delisted_entities_with_fundamentals": 0,71 "cascade": cascade,72 "n_former_names_gt0": u.filter(pl.col("eligible") & (pl.col("n_former_names") > 0)).height,73 "sec_tickers_snapshot_sha256": sha256_file(ext / "sec_company_tickers.json"),74 }75 write_json(P.results / "universe_audit.json", rep)76 pl.DataFrame(cascade).write_csv(P.results / "table_universe_cascade.csv")77 log.info("eligible firms: %d / %d; with CIK %d; with SIC %d", rep["eligible"], rep["entities"], rep["with_cik"], rep["with_sic"])78 print(json.dumps({k: rep[k] for k in ("entities", "eligible", "with_cik", "with_sic", "cascade")}, indent=1, default=str))79 return 0808182if __name__ == "__main__":83 raise SystemExit(main())84