#!/usr/bin/env python
"""Phase 2b: fundamentals extraction per eligible entity (API, cached), wide quarterly panel, reveal dates, labels.
Also freezes the research design (hash of docs/RESEARCH_DESIGN.md + configs) before any model is fitted."""
from __future__ import annotations

import json
import sys
from concurrent.futures import ThreadPoolExecutor
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.labels import build_quarter_table, finalize_labels, match_release_dates  # noqa: E402
from wp39.pipeline import Paths, get_logger, load_config  # noqa: E402
from wp39.pit_fundamentals import add_quarter_index, fetch_entity_facts, wide_quarterly  # noqa: E402
from wp39.provenance import git_sha, now_iso, sha256_file, write_json  # noqa: E402


def main() -> int:
    P = Paths.default()
    log = get_logger("02_pit", P.logs / "02_pit.log")
    cfg = load_config("research")
    # design freeze
    freeze = {"frozen_at": now_iso(), "code_sha": git_sha(P.root),
              "files": {f: sha256_file(P.root / f) for f in ("docs/RESEARCH_DESIGN.md", "docs/PIT_PROTOCOL.md",
                                                               "configs/research.yaml", "configs/models.yaml")}}
    fz = P.manifests / "design_freeze.json"
    if not fz.exists():
        write_json(fz, freeze)
    uni = pl.read_parquet(P.processed / "universe.parquet").filter(pl.col("eligible"))
    ents = uni["entity_id"].to_list()
    log.info("fetching quarterly facts for %d entities", len(ents))
    man = P.manifests / "fundamentals_manifest.jsonl"

    def one(e: str):
        c = FinDataClient(log_path=P.logs / "api_calls.jsonl")
        try:
            return fetch_entity_facts(c, e, P.cache / "fundamentals" / f"{e.replace(':', '_')}.parquet", man)
        finally:
            c.close()

    frames = []
    with ThreadPoolExecutor(max_workers=6) as ex:
        for i, df in enumerate(ex.map(one, ents)):
            frames.append(df)
            if i % 500 == 0:
                log.info("fundamentals %d/%d", i, len(ents))
    facts = pl.concat([f for f in frames if f.height], how="diagonal_relaxed")
    log.info("facts: %d rows, revisions>1: %d", facts.height, facts.filter(pl.col("revision_id") > 1).height)
    wide = add_quarter_index(wide_quarterly(facts))
    events = pl.read_csv(P.cache / "external" / "earnings_events.csv", try_parse_dates=True)
    wide = match_release_dates(wide, events)
    q = finalize_labels(build_quarter_table(wide), threshold=cfg["targets"]["primary"]["threshold"])
    q.write_parquet(P.processed / "quarters.parquet")
    rep = {
        "built_at": now_iso(), "entities": q["entity_id"].n_unique(), "firm_quarters": q.height,
        "release_match": q.group_by("release_match").len().sort("len", descending=True).to_dicts(),
        "release_time": q.group_by("release_time").len().sort("len", descending=True).to_dicts(),
        "filing_minus_release_days_quantiles": {str(p): q["filing_minus_release_days"].quantile(p) for p in (0.1, 0.5, 0.9)},
        "label_base_rates_all": {y: float(q[y].mean()) for y in ("y_adverse_opinc", "y_adverse_netinc", "y_roa_decline",
                                                                  "y_ocf_negative", "y_liquidity_deterioration", "y_abs_surprise")},
        "eps_miss_rate_where_available": float(q["y_eps_miss"].drop_nulls().mean()) if q["y_eps_miss"].drop_nulls().len() else None,
        "eps_estimate_available_share": float(q["eps_estimate"].is_not_null().mean()),
        "S_opinc_quantiles": {str(p): q["S_opinc"].drop_nulls().quantile(p) for p in (0.01, 0.1, 0.25, 0.5, 0.75, 0.9, 0.99)},
        "regular_calendar_share": float((q["_gap_ok8"] == 8).mean()),
    }
    write_json(P.results / "pit_panel_audit.json", rep)
    print(json.dumps(rep, indent=1, default=str))
    return 0


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