#!/usr/bin/env python """Phase 6: generate LaTeX tables, figures and a results-macro file from saved artifacts (never from memory).""" from __future__ import annotations import json import sys from pathlib import Path import numpy as np import pandas as pd import polars as pl sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from wp39 import figures as F # noqa: E402 from wp39.pipeline import Paths, get_logger, load_config # noqa: E402 T = {} # macros def macro(name: str, value, fmt: str = "{}"): T[name] = fmt.format(value) def fnum(x, d=4): return "" if x is None or (isinstance(x, float) and np.isnan(x)) else f"{x:.{d}f}" def stars(p): return "***" if p < 0.01 else "**" if p < 0.05 else "*" if p < 0.1 else "" def tex_table(df: pd.DataFrame, path: Path, cols: list[tuple[str, str, str]], note: str = "", group: str | None = None): """cols: (column, header, fmt) — fmt is a python format or 'int'/'str'/'tex'. `group`: column whose change inserts a rule.""" lines = ["\\begin{tabular}{l" + "r" * (len(cols) - 1) + "}", "\\toprule", " & ".join(h for _, h, _ in cols) + " \\\\", "\\midrule"] prev = None for _, r in df.iterrows(): if group is not None: g = r[group] if group in df.columns else None if prev is not None and g != prev: lines.append("\\addlinespace[2pt]\\midrule") prev = g cells = [] for c, _, fmt in cols: v = r[c] if fmt == "int": cells.append(f"{int(v):,}" if pd.notna(v) else "") elif fmt == "str": cells.append(str(v).replace("_", " ").replace("%", "\\%")) elif fmt == "tex": cells.append(str(v)) else: cells.append(fmt.format(v) if pd.notna(v) else "") lines.append(" & ".join(cells) + " \\\\") lines += ["\\bottomrule", "\\end{tabular}"] path.write_text("\n".join(lines) + "\n") def main() -> int: P = Paths.default() log = get_logger("09_manuscript", P.logs / "09_manuscript.log") cfg = load_config("research"); hp = cfg["cutoffs"]["primary_horizon_td"] E, R, TB, FG = P.evaluation, P.results, P.paper_tables, P.paper_figures ua = json.load(open(R / "universe_audit.json")); pa = json.load(open(R / "pit_panel_audit.json")); sa = json.load(open(R / "sample_audit.json")) ev = pd.read_parquet(P.processed / "sample_event.parquet", columns=["entity_id", "fiscal_period_end", "h", "cutoff", "A_q", "y_adverse_opinc", "S_opinc", "f_log_mcap", "wd_amihud60", "f_leverage", "release_time", "sic_division"]) e20 = ev[ev.h == hp] # ---------------- macros: sample macro("nEntities", ua["entities"], "{:,}"); macro("nEligible", ua["eligible"], "{:,}"); macro("nWithCik", ua["with_cik"], "{:,}") macro("nFirmQuarters", pa["firm_quarters"], "{:,}"); macro("nFirmsPanel", pa["entities"], "{:,}") macro("nSampleEvent", sa["event"]["rows"], "{:,}"); macro("nSampleFirms", sa["event"]["firms"], "{:,}") macro("nSampleQuarters", sa["event"]["quarters"], "{:,}"); macro("nSampleCalendar", sa["calendar"]["rows"], "{:,}") macro("nObsPrimary", len(e20), "{:,}"); macro("baseRatePrimary", 100 * e20.y_adverse_opinc.mean(), "{:.1f}") macro("firstCutoff", str(e20.cutoff.min())[:10]); macro("lastCutoff", str(e20.cutoff.max())[:10]) macro("releaseMatchedShare", 100 * next(r["len"] for r in pa["release_match"] if r["release_match"] == "ok") / pa["firm_quarters"], "{:.1f}") macro("filingLagMedian", pa["filing_minus_release_days_quantiles"]["0.5"], "{:.0f}") macro("epsEstimateShare", 100 * pa["eps_estimate_available_share"], "{:.0f}") # ---------------- Table 2: universe cascade tex_table(pd.read_csv(R / "table_universe_cascade.csv"), TB / "tab02_universe.tex", [("step", "Step", "str"), ("n", "Firms", "int")]) casc = pd.read_csv(R / "table_sample_cascade_event.csv") tex_table(casc, TB / "tab02b_sample_cascade.tex", [("step", "Exclusion step", "str"), ("remaining", "Remaining", "int"), ("dropped", "Dropped", "int")]) # ---------------- Table 3: summary statistics (h = 20 sample) feat = pd.read_parquet(P.processed / "sample_event.parquet") feat = feat[feat.h == hp] desc_cols = {"S_opinc": "Scaled operating-income surprise $S_q$", "y_adverse_opinc": "Adverse surprise ($S_q \\le -0.5\\%$)", "f_roa": "ROA (latest available)", "f_leverage": "Debt / assets", "f_log_mcap": "log market cap", "wd_ret20": "20-day return", "wd_vol20": "20-day daily volatility", "wd_amihud60": "Amihud illiquidity (60d, $\\times 10^6$)", "w1m_rv": "1-min realized variance (daily mean)", "w1m_rj": "1-min relative jump", "w1m_rsneg_share": "1-min negative semivariance share", "w1m_down_vol_share": "1-min down-bar volume share", "w1m_present_share": "1-min bars present (share)", "w30m_rv": "30-min realized variance", "w5m_rv": "5-min realized variance", "info_age_td": "Days since last release"} rows = [] for c, lab in desc_cols.items(): s = pd.to_numeric(feat[c], errors="coerce").replace([np.inf, -np.inf], np.nan).dropna() rows.append({"var": lab, "n": len(s), "mean": s.mean(), "sd": s.std(), "p10": s.quantile(0.1), "p50": s.median(), "p90": s.quantile(0.9)}) tex_table(pd.DataFrame(rows), TB / "tab03_summary.tex", [("var", "Variable", "tex"), ("n", "N", "int"), ("mean", "Mean", "{:.4g}"), ("sd", "SD", "{:.4g}"), ("p10", "P10", "{:.4g}"), ("p50", "Median", "{:.4g}"), ("p90", "P90", "{:.4g}")]) # ---------------- Table 4: information sets fc = json.load(open(E / "feature_columns_event.json")) info = pd.DataFrame([{"set": "B1", "desc": "PIT fundamentals + history + size", "k": len(fc["fundamentals"])}, {"set": "B2", "desc": "B1 + daily OHLCV (20/60-day)", "k": len(fc["fundamentals"]) + len(fc["daily"])}, {"set": "B3", "desc": "B2 + 30-minute families", "k": len(fc["fundamentals"]) + len(fc["daily"]) + len(fc["intraday_30m"])}, {"set": "B4", "desc": "B2 + 5-minute families", "k": len(fc["fundamentals"]) + len(fc["daily"]) + len(fc["intraday_5m"])}, {"set": "B5", "desc": "B2 + 1-minute families", "k": len(fc["fundamentals"]) + len(fc["daily"]) + len(fc["intraday_1m"])}, {"set": "B5all", "desc": "B2 + 30m + 5m + 1m", "k": sum(len(v) for v in fc.values())}]) tex_table(info, TB / "tab04_info_sets.tex", [("set", "Set", "str"), ("desc", "Information set", "str"), ("k", "Features", "int")]) macro("kFund", len(fc["fundamentals"])); macro("kDaily", len(fc["daily"])); macro("kIntra", len(fc["intraday_1m"])) # ---------------- Table 5: headline OOS metrics (h=20, chrono test) met = pd.read_csv(E / "metrics_all.csv") m20 = met[(met.h == hp) & (met.target == "y_adverse_opinc") & (met.part == "test") & (met.tag == "chrono")].copy() order = ["B0", "B0b", "B1", "B2", "B3", "B4", "B5", "B5all", "LB1", "LB2", "LB3", "LB4", "LB5", "LB5all"] m20["o"] = m20.model.map({m: i for i, m in enumerate(order)}); m20 = m20.sort_values("o") m20["fam"] = np.where(m20.model.str.startswith("L"), "lightgbm", np.where(m20.model.str.startswith("B0"), "base", "logistic")) tex_table(m20, TB / "tab05_headline.tex", group="fam", cols= [("model", "Model", "str"), ("n", "N", "int"), ("logloss", "Log loss", "{:.4f}"), ("brier", "Brier", "{:.4f}"), ("auc", "AUC", "{:.3f}"), ("pr_auc", "PR-AUC", "{:.3f}"), ("cal_slope", "Cal. slope", "{:.2f}"), ("cal_intercept", "Cal. int.", "{:.2f}")]) for _, r in m20.iterrows(): macro(f"ll{r.model}", r.logloss, "{:.4f}"); macro(f"auc{r.model}", r.auc, "{:.3f}") # ---------------- Table 5b: paired contrasts con = pd.read_csv(E / "contrasts.csv") con["ci"] = con.apply(lambda r: f"[{r.ci_lo:+.4f}, {r.ci_hi:+.4f}]", axis=1) con["gwp"] = con.apply(lambda r: f"{r.gw_p:.3f}{stars(r.gw_p)}", axis=1) con["holm"] = con["gw_p_holm"].map(lambda v: "" if pd.isna(v) else f"{v:.3f}") tex_table(con, TB / "tab05b_contrasts.tex", group="family", cols=[("family", "Family", "str"), ("a", "A", "str"), ("b", "B", "str"), ("n", "N", "int"), ("d_logloss", "$\\Delta$ log loss", "{:+.4f}"), ("ci", "95\\% CI", "str"), ("gwp", "GW $p$", "str"), ("holm", "Holm", "str"), ("d_brier", "$\\Delta$ Brier", "{:+.5f}")]) for _, r in con.iterrows(): k = f"{r.family[:3]}{r.a}vs{r.b}".replace("B5all", "Ball") macro(f"d{k}", r.d_logloss, "{:+.4f}"); macro(f"lo{k}", r.ci_lo, "{:+.4f}"); macro(f"hi{k}", r.ci_hi, "{:+.4f}"); macro(f"p{k}", r.gw_p, "{:.3f}") macro(f"n{k}", r.n, "{:,}") # ---------------- Table 7: horizon ladder, secondary targets, firm holdout lad = pd.read_csv(E / "horizon_ladder.csv") lad["ci"] = lad.apply(lambda r: f"[{r.ci_lo:+.4f}, {r.ci_hi:+.4f}]", axis=1) tex_table(lad.sort_values(["family", "h"], ascending=[True, False]), TB / "tab07_horizons.tex", group="family", cols=[("family", "Family", "str"), ("h", "$h$", "int"), ("n", "N", "int"), ("auc_b", "AUC B2", "{:.3f}"), ("auc_a", "AUC B5", "{:.3f}"), ("d_logloss", "$\\Delta$ log loss", "{:+.4f}"), ("ci", "95\\% CI", "str"), ("gw_p", "GW $p$", "{:.3f}")]) h4 = json.load(open(E / "h4_timing.json")) for fam, v in h4.items(): macro(f"h4rho{fam}", v["spearman_gain_vs_minus_h"], "{:+.2f}"); macro(f"h4p{fam}", v["p"], "{:.2f}") sec = pd.read_csv(E / "secondary_targets.csv") sec["ci"] = sec.apply(lambda r: f"[{r.ci_lo:+.4f}, {r.ci_hi:+.4f}]", axis=1) tex_table(sec, TB / "tab10_targets.tex", group="target", cols=[("target", "Target", "str"), ("family", "Family", "str"), ("n", "N", "int"), ("base_rate", "Base rate", "{:.3f}"), ("auc_b", "AUC B2", "{:.3f}"), ("auc_a", "AUC B5", "{:.3f}"), ("d_logloss", "$\\Delta$ log loss", "{:+.4f}"), ("ci", "95\\% CI", "str"), ("gw_p", "GW $p$", "{:.3f}")]) fh = pd.read_csv(E / "firm_holdout.csv"); fh["ci"] = fh.apply(lambda r: f"[{r.ci_lo:+.4f}, {r.ci_hi:+.4f}]", axis=1) for _, r in fh.iterrows(): macro(f"d{r.family[:3]}{r.tag.replace('_', '')}", r.d_logloss, "{:+.4f}"); macro(f"lo{r.family[:3]}{r.tag.replace('_', '')}", r.ci_lo, "{:+.4f}"); macro(f"hi{r.family[:3]}{r.tag.replace('_', '')}", r.ci_hi, "{:+.4f}") tex_table(fh, TB / "tab09b_firmholdout.tex", [("tag", "Firms", "str"), ("family", "Family", "str"), ("n", "N", "int"), ("d_logloss", "$\\Delta$ log loss", "{:+.4f}"), ("ci", "95\\% CI", "str"), ("gw_p", "GW $p$", "{:.3f}")]) # ---------------- Table 6: panel regressions pr = pd.read_csv(E / "panel_regressions.csv") pr = pr[pr["var"].isin(["lfds5", "lfds2", "d_lfds", "dlfds_x_size0", "dlfds_x_size2", "dlfds_x_illiq0", "dlfds_x_illiq2", "dlfds_x_lev0", "dlfds_x_lev2"])].copy() pr["coef_s"] = pr.apply(lambda r: f"{r.coef:.4f}{stars(r.p)}", axis=1); pr["se_s"] = pr.se.map(lambda v: f"({v:.4f})") tex_table(pr, TB / "tab06_panel.tex", group="spec", cols=[("spec", "Specification", "str"), ("var", "Regressor", "str"), ("coef_s", "Coef.", "str"), ("se_s", "SE", "str"), ("n", "N", "int"), ("n_firms", "Firms", "int"), ("n_quarters", "Quarters", "int")]) for _, r in pr.iterrows(): macro(f"pan{r.spec.replace('_', '')}{r['var'].replace('_', '')}", r.coef, "{:+.4f}"); macro(f"pant{r.spec.replace('_', '')}{r['var'].replace('_', '')}", r.t, "{:.2f}") # ---------------- Table 7b: mechanisms / subgroups; Table 8: placebos sub = pd.read_csv(E / "subgroups.csv"); sub["ci"] = sub.apply(lambda r: f"[{r.ci_lo:+.4f}, {r.ci_hi:+.4f}]", axis=1) tex_table(sub[sub.family == "lightgbm"], TB / "tab07b_subgroups.tex", [("subgroup", "Subgroup", "str"), ("n", "N", "int"), ("base_rate", "Base rate", "{:.3f}"), ("auc_b", "AUC LB2", "{:.3f}"), ("auc_a", "AUC LB5", "{:.3f}"), ("mean", "$\\Delta$ log loss", "{:+.4f}"), ("ci", "95\\% CI", "str")]) plc = pd.read_csv(E / "placebo_shuffled.csv"); plc["ci"] = plc.apply(lambda r: f"[{r.ci_lo:+.4f}, {r.ci_hi:+.4f}]", axis=1) tex_table(plc, TB / "tab08_placebo.tex", [("placebo", "Placebo", "str"), ("draw", "Draw", "int"), ("n", "N", "int"), ("auc_a", "AUC LB5", "{:.3f}"), ("auc_b", "AUC LB2", "{:.3f}"), ("mean", "$\\Delta$ log loss", "{:+.4f}"), ("ci", "95\\% CI", "str")]) macro("placeboMeanD", plc["mean"].mean(), "{:+.4f}"); macro("placeboAUC", plc["auc_a"].mean(), "{:.3f}") # ---------------- Table 11: economic value h5 = json.load(open(E / "h5_portfolios.json")); car = pd.read_csv(E / "h5_car_regressions.csv") rows = [] for m, v in h5.items(): for k in ("gross", "net20bps"): a = v[k]; rows.append({"model": m, "variant": k, "alpha": 100 * a["alpha_monthly"], "t": a["alpha_t"], "mean_ls": 100 * a["mean_ls"], "sharpe": a["sharpe_ann"], "T": a["T"], "turnover": v["avg_turnover"]}) tex_table(pd.DataFrame(rows), TB / "tab11_economic.tex", group="model", cols=[("model", "Score", "str"), ("variant", "Variant", "str"), ("mean_ls", "Mean L/S (\\%/mo)", "{:.2f}"), ("alpha", "FF5+MOM $\\alpha$ (\\%/mo)", "{:+.2f}"), ("t", "$t$ (HAC)", "{:.2f}"), ("sharpe", "Sharpe (ann.)", "{:.2f}"), ("turnover", "Turnover", "{:.2f}"), ("T", "Months", "int")]) for m, v in h5.items(): macro(f"alpha{m}gross", 100 * v["gross"]["alpha_monthly"], "{:+.2f}"); macro(f"alphat{m}gross", v["gross"]["alpha_t"], "{:.2f}") macro(f"alpha{m}net", 100 * v["net20bps"]["alpha_monthly"], "{:+.2f}"); macro(f"alphat{m}net", v["net20bps"]["alpha_t"], "{:.2f}") carr = car[car["var"].isin(["lfds5", "lfds2", "d_lfds"])].copy(); carr["coef_s"] = carr.apply(lambda r: f"{r.coef:.4f}{stars(r.p)}", axis=1); carr["se_s"] = carr.se.map(lambda v: f"({v:.4f})") tex_table(carr, TB / "tab11b_car.tex", group="spec", cols=[("spec", "Specification", "str"), ("var", "Regressor", "str"), ("coef_s", "Coef.", "str"), ("se_s", "SE", "str"), ("n", "N", "int")]) for _, r in carr.iterrows(): macro(f"car{r.spec.replace('_', '')}{r['var'].replace('_', '')}", r.coef, "{:+.4f}"); macro(f"cart{r.spec.replace('_', '')}{r['var'].replace('_', '')}", r.t, "{:.2f}") # continuous + power cont = pd.read_csv(E / "continuous.csv") tex_table(cont[cont.model.isin(["B2", "B5"])], TB / "tab10b_continuous.tex", [("model", "Model", "str"), ("n", "N", "int"), ("mae", "MAE", "{:.5f}"), ("rmse", "RMSE", "{:.5f}"), ("oos_r2", "OOS $R^2$", "{:.4f}")]) for _, r in cont.iterrows(): if r.model in ("B2", "B5"): macro(f"r2{r.model}", r.oos_r2, "{:.4f}") pw = json.load(open(E / "power.json")); macro("mde", pw.get("mde_logloss_80pct_power", float("nan")), "{:.4f}"); macro("valQuarters", pw.get("validation_quarters", 0)) # ---------------- figures F.fig_timeline(FG) yrs = e20.assign(year=pd.to_datetime(e20.cutoff).dt.year).groupby("year").agg(firms=("entity_id", "nunique"), firm_quarters=("entity_id", "size"), adverse_rate=("y_adverse_opinc", "mean")).reset_index() F.fig_coverage(FG, yrs) te = e20[pd.to_datetime(e20.cutoff) >= pd.Timestamp(str(cfg["splits"]["test"][0]))].copy() prev_rows = [{"label": "All firm-quarters", "sel": np.ones(len(te), bool)}] for name, col, lab in (("size", "f_log_mcap", "Market cap"), ("illiq", "wd_amihud60", "Illiquidity"), ("lev", "f_leverage", "Leverage")): q = te[col].rank(pct=True) prev_rows += [{"label": f"{lab}: bottom tercile", "sel": (q <= 1 / 3).values}, {"label": f"{lab}: top tercile", "sel": (q > 2 / 3).values}] for rt in ("before_open", "after_close"): prev_rows.append({"label": f"Release {rt.replace('_', ' ')}", "sel": (te.release_time == rt).values}) pr_ = [] for r in prev_rows: y = te.y_adverse_opinc.values[r["sel"]]; n = len(y); p = y.mean() if n else np.nan; se = np.sqrt(p * (1 - p) / n) if n else np.nan pr_.append({"label": r["label"], "rate": p, "lo": p - 1.96 * se, "hi": p + 1.96 * se}) F.fig_prevalence(FG, pd.DataFrame(pr_)) F.fig_frequency_gains(FG, con) F.fig_reliability(FG, pd.read_csv(E / "reliability.csv")) F.fig_signal_evolution(FG, lad) F.fig_heterogeneity(FG, sub, ["all", "size_small", "size_large", "illiq_high", "illiq_low", "lev_high", "lev_low", "loss_firms", "profit_firms", "covid_2020", "ex_covid", "window_clean", "window_straddles_prev_release", "dense_bars", "sparse_bars", "sector_finance", "sector_nonfinance"]) F.fig_rolling(FG, pd.read_csv(E / "walkforward_by_year.csv"), pd.read_csv(E / "walkforward_contrasts.csv")) if "LB5" in h5: F.fig_economic(FG, pd.read_csv(E / "portfolios_LB5.csv"), h5["LB5"]) # ---------------- macros file lines = ["% Auto-generated by scripts/09_make_manuscript.py — do not edit"] for k, v in T.items(): words = {"0": "Zero", "1": "One", "2": "Two", "3": "Three", "4": "Four", "5": "Five", "6": "Six", "7": "Seven", "8": "Eight", "9": "Nine"} kk = "".join(words.get(ch, ch) if ch.isdigit() else ch for ch in k if ch.isalnum()) lines.append(f"\\newcommand{{\\{kk}}}{{{v}}}") (P.root / "paper" / "results_macros.tex").write_text("\n".join(lines) + "\n") log.info("wrote %d macros, %d tables, figures to %s", len(T), len(list(TB.glob('*.tex'))), FG) return 0 if __name__ == "__main__": raise SystemExit(main())