1"""Point-in-time fundamentals: extraction, wide quarterly panel, and as-of joins (docs/PIT_PROTOCOL.md).23The API's `as_of` semantics are replicated locally on the full extract (available_at <= as_of and revision 1) so that4one call per entity suffices; `tests/test_api_contract.py::test_asof_equivalence` checks the equivalence live.5"""6from __future__ import annotations78import datetime as dt9from pathlib import Path1011import polars as pl1213from .api_client import FinDataClient14from .provenance import append_jsonl, now_iso, sha256_file1516METRICS = ["rev", "cogs", "gp", "op_inc", "op_ex", "ebit", "ebitda", "net_inc", "net_inc_com", "eps", "eps_d",17 "cash_op", "capex", "cash_inv", "cash_fin", "cash", "assets_c", "assets_tot", "liab_c", "liab_tot", "debt",18 "debt_c", "debt_nc", "equity", "receivables", "inventory", "payables", "ppe", "intangibles", "research_dev",19 "sgna", "interest_exp", "intr_cover", "depr", "shares", "shares_av", "shares_av_diluted", "roa", "roe",20 "roic", "gm", "om", "ret_earnings", "cash_div", "dps", "share_comp", "tax_expense"]2122FACT_COLS = ["entity_id", "security_id", "metric", "value", "unit", "fiscal_period_start", "fiscal_period_end",23 "fiscal_period_type", "fiscal_year", "fiscal_period", "filing_date", "available_at", "observed_at",24 "revision_id", "restatement_flag", "superseded_at", "source_document"]252627def fetch_entity_facts(client: FinDataClient, entity_id: str, dest: Path, manifest: Path) -> pl.DataFrame:28 """All quarterly facts of one entity (no as_of → full history, current revision). Cached as Parquet."""29 if dest.exists():30 return pl.read_parquet(dest)31 d = client.fundamentals(entity_id, as_of=None, period_type="q", limit=20000)32 rows = d.get("data", [])33 if len(rows) >= 20000:34 raise RuntimeError(f"{entity_id}: fact count hit the API limit; narrow the metrics")35 df = pl.DataFrame([{k: r.get(k) for k in FACT_COLS} for r in rows]) if rows else pl.DataFrame(schema={c: pl.Utf8 for c in FACT_COLS})36 if rows:37 df = df.with_columns([38 pl.col("value").cast(pl.Float64), pl.col("fiscal_year").cast(pl.Int32), pl.col("revision_id").cast(pl.Int32),39 pl.col("fiscal_period_end").str.to_date(), pl.col("fiscal_period_start").str.to_date(),40 pl.col("filing_date").str.to_date(),41 pl.col("available_at").str.to_datetime(time_zone="UTC", strict=False),42 pl.col("observed_at").str.to_datetime(time_zone="UTC", strict=False),43 ])44 dest.parent.mkdir(parents=True, exist_ok=True)45 df.write_parquet(dest)46 append_jsonl(manifest, {"entity_id": entity_id, "rows": df.height, "sha256": sha256_file(dest), "fetched_at": now_iso(),47 "pit_policy": d.get("meta", {}).get("pit_policy"), "license": d.get("meta", {}).get("license"),48 "path": str(dest)})49 return df505152def wide_quarterly(facts: pl.DataFrame) -> pl.DataFrame:53 """One row per (entity, fiscal_period_end) with metric columns, filing_date, available_at (ET date of availability)."""54 f = facts.filter((pl.col("fiscal_period_type") == "q") & (pl.col("revision_id") == 1) & pl.col("metric").is_in(METRICS))55 if f.height == 0:56 return pl.DataFrame()57 wide = f.pivot(on="metric", index=["entity_id", "security_id", "fiscal_period_end", "fiscal_year", "fiscal_period",58 "filing_date", "available_at"], values="value", aggregate_function="first")59 for m in METRICS:60 if m not in wide.columns:61 wide = wide.with_columns(pl.lit(None, dtype=pl.Float64).alias(m))62 wide = wide.with_columns(pl.col("available_at").dt.convert_time_zone("America/New_York").dt.date().alias("available_date"))63 return wide.sort(["entity_id", "fiscal_period_end"])646566def add_quarter_index(wide: pl.DataFrame) -> pl.DataFrame:67 """Sequential quarter index per entity + gap flags (irregular fiscal calendars)."""68 w = wide.sort(["entity_id", "fiscal_period_end"]).with_columns([69 pl.col("fiscal_period_end").diff().over("entity_id").dt.total_days().alias("days_since_prev_q"),70 ])71 w = w.with_columns(pl.int_range(pl.len()).over("entity_id").alias("qidx"))72 return w737475def asof_latest_available(wide: pl.DataFrame, cutoffs: pl.DataFrame) -> pl.DataFrame:76 """For each (entity_id, cutoff) return the latest quarter with available_date <= cutoff (strict PIT).7778 cutoffs: columns entity_id, cutoff (date). Output keeps all cutoff columns and appends the fundamentals row (suffix _pit).79 """80 cols = [m for m in METRICS if m in wide.columns]81 w = wide.select(["entity_id", "fiscal_period_end", "available_date", "qidx"] + cols).rename(82 {"fiscal_period_end": "pit_period_end", "available_date": "pit_available_date", "qidx": "pit_qidx"})83 w = w.sort(["entity_id", "pit_available_date"])84 c = cutoffs.sort(["entity_id", "cutoff"])85 joined = c.join_asof(w, left_on="cutoff", right_on="pit_available_date", by="entity_id", strategy="backward")86 # join_asof with strategy backward is <=; make the inequality explicit for the leakage test87 joined = joined.with_columns((pl.col("pit_available_date") <= pl.col("cutoff")).alias("pit_ok"))88 return joined899091def pit_feature_block(df: pl.DataFrame) -> pl.DataFrame:92 """Fundamental ratios from the latest available quarter (columns from asof_latest_available) + lagged growth.9394 Growth rates need the same-quarter-last-year value *as of the cutoff*; they are supplied by `samples.py` through the95 wide panel (quarters with available_date <= cutoff only).96 """97 eps = 1e-998 ta = pl.col("assets_tot").abs() + eps99 rev = pl.col("rev").abs() + eps100 return df.with_columns([101 (pl.col("gp") / rev).alias("f_gross_margin"),102 (pl.col("op_inc") / rev).alias("f_op_margin"),103 (pl.col("net_inc") / ta).alias("f_ni_assets"),104 (pl.col("cash_op") / ta).alias("f_ocf_assets"),105 ((pl.col("net_inc") - pl.col("cash_op")) / ta).alias("f_accruals"),106 (pl.col("cash") / ta).alias("f_cash_assets"),107 (pl.col("debt") / ta).alias("f_leverage"),108 (pl.col("assets_c") / (pl.col("liab_c").abs() + eps)).alias("f_current_ratio"),109 (pl.col("capex") / ta).alias("f_capex_assets"),110 (pl.col("research_dev") / rev).alias("f_rd_rev"),111 pl.col("intr_cover").clip(-50, 50).alias("f_int_cover"),112 (pl.col("net_inc") < 0).cast(pl.Int8).alias("f_loss"),113 pl.col("assets_tot").abs().log1p().alias("f_log_assets"),114 pl.col("roa").alias("f_roa"),115 ])116