#!/usr/bin/env python
"""Phase 2a: universe audit — fundamentals entities x 1-minute coverage x SEC crosswalk; exclusions documented.

Inputs: node extracts produced by scripts/01a_extract_node_tables.sh (entities, earnings_events, corporate_actions —
read-only Postgres copies, documented exception), FinData /coverage/symbols and /instruments, SEC public files.
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

import polars as pl

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from wp39.api_client import FinDataClient  # noqa: E402
from wp39.instruments import build_crosswalk, fetch_sec_submissions, fetch_sec_tickers  # noqa: E402
from wp39.pipeline import Paths, get_logger, load_config  # noqa: E402
from wp39.provenance import now_iso, sha256_file, write_json  # noqa: E402


def main() -> int:
    P = Paths.default()
    log = get_logger("01_audit", P.logs / "01_audit.log")
    cfg = load_config("research"); contract = load_config("data_contract")
    ext = P.cache / "external"
    ents = pl.read_csv(ext / "entities.csv")
    ents = ents.with_columns(pl.col("entity_id").str.replace("^frd:", "").alias("ticker"))
    log.info("fundamentals entities: %d", ents.height)

    client = FinDataClient(log_path=P.logs / "api_calls.jsonl")
    cov = pl.DataFrame(list(client.coverage_symbols(contract["findata"]["datasets"]["bars_1m"], min_sessions=1)))
    cov = cov.rename({"symbol": "ticker", "sessions": "n_sessions_1m", "d0": "first_session_1m", "d1": "last_session_1m", "bars": "n_bars_1m"})
    cov = cov.with_columns([pl.col("first_session_1m").str.to_date(), pl.col("last_session_1m").str.to_date()])
    log.info("1m coverage rows: %d (incl. %d delisted)", cov.height, cov.filter(pl.col("ticker").str.ends_with("-DELISTED")).height)

    sec = fetch_sec_tickers(ext)
    u = ents.join(cov, on="ticker", how="left")
    u = u.join(sec.unique(subset=["ticker"]), on="ticker", how="left")
    subs = fetch_sec_submissions(sorted(u["cik"].drop_nulls().unique().to_list()), ext, log=log)
    u = u.join(subs, on="cik", how="left")
    u = u.with_columns([
        pl.col("first_q").str.to_date(), pl.col("last_q").str.to_date(), pl.col("first_filing").str.to_date(),
        pl.col("cik").is_not_null().alias("has_cik"),
        pl.col("sic_division").fill_null("unknown"),
    ])
    # eligibility (frozen): >= min quarters, 1m coverage overlapping 2012+, SIC not public administration
    u = u.with_columns([
        (pl.col("n_q") >= cfg["sample"]["min_quarters_per_firm"]).alias("ok_quarters"),
        (pl.col("n_sessions_1m").fill_null(0) >= 250).alias("ok_coverage"),
        (pl.col("last_session_1m") >= pl.date(2012, 6, 30)).fill_null(False).alias("ok_period"),
        (pl.col("sic_division") != "J_public_admin").alias("ok_sic"),
    ])
    u = u.with_columns((pl.col("ok_quarters") & pl.col("ok_coverage") & pl.col("ok_period") & pl.col("ok_sic")).alias("eligible"))
    u.write_parquet(P.processed / "universe.parquet")
    cascade = [
        {"step": "fundamentals_entities", "n": u.height},
        {"step": ">= 8 quarterly periods", "n": u.filter(pl.col("ok_quarters")).height},
        {"step": "+ 1m bars (>= 250 sessions)", "n": u.filter(pl.col("ok_quarters") & pl.col("ok_coverage")).height},
        {"step": "+ bars after 2012-06", "n": u.filter(pl.col("ok_quarters") & pl.col("ok_coverage") & pl.col("ok_period")).height},
        {"step": "+ not public administration", "n": u.filter(pl.col("eligible")).height},
    ]
    rep = {
        "built_at": now_iso(), "entities": u.height, "eligible": u.filter(pl.col("eligible")).height,
        "with_cik": u.filter(pl.col("has_cik")).height, "with_sic": u.filter(pl.col("sic").is_not_null()).height,
        "sic_division_counts": u.filter(pl.col("eligible")).group_by("sic_division").len().sort("len", descending=True).to_dicts(),
        "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(),
        "exchanges": u.filter(pl.col("eligible")).explode("exchanges").group_by("exchanges").len().sort("len", descending=True).head(8).to_dicts(),
        "delisted_symbols_in_1m_lake": int(cov.filter(pl.col("ticker").str.ends_with("-DELISTED")).height),
        "delisted_entities_with_fundamentals": 0,
        "cascade": cascade,
        "n_former_names_gt0": u.filter(pl.col("eligible") & (pl.col("n_former_names") > 0)).height,
        "sec_tickers_snapshot_sha256": sha256_file(ext / "sec_company_tickers.json"),
    }
    write_json(P.results / "universe_audit.json", rep)
    pl.DataFrame(cascade).write_csv(P.results / "table_universe_cascade.csv")
    log.info("eligible firms: %d / %d; with CIK %d; with SIC %d", rep["eligible"], rep["entities"], rep["with_cik"], rep["with_sic"])
    print(json.dumps({k: rep[k] for k in ("entities", "eligible", "with_cik", "with_sic", "cascade")}, indent=1, default=str))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())