1#!/usr/bin/env python2"""Phase 2b: fundamentals extraction per eligible entity (API, cached), wide quarterly panel, reveal dates, labels.3Also freezes the research design (hash of docs/RESEARCH_DESIGN.md + configs) before any model is fitted."""4from __future__ import annotations56import json7import sys8from concurrent.futures import ThreadPoolExecutor9from pathlib import Path1011import polars as pl1213sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))14from wp39.api_client import FinDataClient # noqa: E40215from wp39.labels import build_quarter_table, finalize_labels, match_release_dates # noqa: E40216from wp39.pipeline import Paths, get_logger, load_config # noqa: E40217from wp39.pit_fundamentals import add_quarter_index, fetch_entity_facts, wide_quarterly # noqa: E40218from wp39.provenance import git_sha, now_iso, sha256_file, write_json # noqa: E402192021def main() -> int:22 P = Paths.default()23 log = get_logger("02_pit", P.logs / "02_pit.log")24 cfg = load_config("research")25 # design freeze26 freeze = {"frozen_at": now_iso(), "code_sha": git_sha(P.root),27 "files": {f: sha256_file(P.root / f) for f in ("docs/RESEARCH_DESIGN.md", "docs/PIT_PROTOCOL.md",28 "configs/research.yaml", "configs/models.yaml")}}29 fz = P.manifests / "design_freeze.json"30 if not fz.exists():31 write_json(fz, freeze)32 uni = pl.read_parquet(P.processed / "universe.parquet").filter(pl.col("eligible"))33 ents = uni["entity_id"].to_list()34 log.info("fetching quarterly facts for %d entities", len(ents))35 man = P.manifests / "fundamentals_manifest.jsonl"3637 def one(e: str):38 c = FinDataClient(log_path=P.logs / "api_calls.jsonl")39 try:40 return fetch_entity_facts(c, e, P.cache / "fundamentals" / f"{e.replace(':', '_')}.parquet", man)41 finally:42 c.close()4344 frames = []45 with ThreadPoolExecutor(max_workers=6) as ex:46 for i, df in enumerate(ex.map(one, ents)):47 frames.append(df)48 if i % 500 == 0:49 log.info("fundamentals %d/%d", i, len(ents))50 facts = pl.concat([f for f in frames if f.height], how="diagonal_relaxed")51 log.info("facts: %d rows, revisions>1: %d", facts.height, facts.filter(pl.col("revision_id") > 1).height)52 wide = add_quarter_index(wide_quarterly(facts))53 events = pl.read_csv(P.cache / "external" / "earnings_events.csv", try_parse_dates=True)54 wide = match_release_dates(wide, events)55 q = finalize_labels(build_quarter_table(wide), threshold=cfg["targets"]["primary"]["threshold"])56 q.write_parquet(P.processed / "quarters.parquet")57 rep = {58 "built_at": now_iso(), "entities": q["entity_id"].n_unique(), "firm_quarters": q.height,59 "release_match": q.group_by("release_match").len().sort("len", descending=True).to_dicts(),60 "release_time": q.group_by("release_time").len().sort("len", descending=True).to_dicts(),61 "filing_minus_release_days_quantiles": {str(p): q["filing_minus_release_days"].quantile(p) for p in (0.1, 0.5, 0.9)},62 "label_base_rates_all": {y: float(q[y].mean()) for y in ("y_adverse_opinc", "y_adverse_netinc", "y_roa_decline",63 "y_ocf_negative", "y_liquidity_deterioration", "y_abs_surprise")},64 "eps_miss_rate_where_available": float(q["y_eps_miss"].drop_nulls().mean()) if q["y_eps_miss"].drop_nulls().len() else None,65 "eps_estimate_available_share": float(q["eps_estimate"].is_not_null().mean()),66 "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)},67 "regular_calendar_share": float((q["_gap_ok8"] == 8).mean()),68 }69 write_json(P.results / "pit_panel_audit.json", rep)70 print(json.dumps(rep, indent=1, default=str))71 return 0727374if __name__ == "__main__":75 raise SystemExit(main())76