#!/usr/bin/env python """Phase 3: baseline ladder B0-B5 (base rates, logistic ridge per information set) on the chronological split, every horizon, primary + secondary targets; walk-forward out-of-fold predictions; firm hold-out variant. Predictions are saved per (design, h, target, model) so that 07_evaluate.py computes every metric from files.""" from __future__ import annotations import json import sys from pathlib import Path import numpy as np import polars as pl sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from wp39.models import INFO_SETS, Preprocessor, base_rate_predict, columns_for, fit_logistic, fit_ridge, firm_history_predict, predict_proba # noqa: E402 from wp39.pipeline import Paths, get_logger, load_config # noqa: E402 from wp39.provenance import now_iso, write_json # noqa: E402 from wp39.samples import feature_columns # noqa: E402 from wp39.splits import chronological, firm_holdout_mask, overlap_check, walk_forward # noqa: E402 KEY = ["entity_id", "symbol", "fiscal_period_end", "h", "cutoff", "A_q", "sic_division", "release_time"] TARGETS = ["y_adverse_opinc", "y_adverse_netinc", "y_roa_decline", "y_ocf_negative", "y_liquidity_deterioration", "y_abs_surprise"] def save_preds(P: Paths, df: pl.DataFrame, p: np.ndarray, tag: str, design: str, h: int, target: str, model: str, part: str) -> None: out = df.select(KEY + [target]).with_columns([pl.Series("p", p), pl.lit(model).alias("model"), pl.lit(part).alias("part"), pl.lit(tag).alias("tag")]) d = P.evaluation / "preds"; d.mkdir(exist_ok=True) out.write_parquet(d / f"{design}_h{h}_{target}_{model}_{part}_{tag}.parquet") def run_split(P, cfg, mcfg, fc, design, h, target, train, valid, test, tag, log, info_sets, record): ytr, yva, yte = (x[target].to_numpy().astype(int) for x in (train, valid, test)) # B0 save_preds(P, test, base_rate_predict(ytr, test.height), tag, design, h, target, "B0", "test") save_preds(P, valid, base_rate_predict(ytr, valid.height), tag, design, h, target, "B0", "validation") save_preds(P, test, firm_history_predict(train, test, target), tag, design, h, target, "B0b", "test") for name in info_sets: cols = columns_for(name, fc) pp = Preprocessor(cols).fit(train) Xtr, Xva, Xte = pp.transform(train), pp.transform(valid), pp.transform(test) m, info = fit_logistic(Xtr, ytr, Xva, yva, mcfg["logistic"]["C_grid"], cfg["seed"]) save_preds(P, valid, predict_proba(m, Xva), tag, design, h, target, name, "validation") save_preds(P, test, predict_proba(m, Xte), tag, design, h, target, name, "test") record.append({"design": design, "h": h, "target": target, "model": name, "tag": tag, "n_train": train.height, "n_valid": valid.height, "n_test": test.height, "n_features": len(cols), **info, "coef": dict(zip(cols, m.coef_[0].round(4).tolist()))}) log.info("%s h=%d %s %s %s: C=%s val_ll=%.4f (p=%d)", design, h, target, name, tag, info["C"], info["val_logloss"], len(cols)) def main() -> int: P = Paths.default() log = get_logger("05_baselines", P.logs / "05_baselines.log") cfg, mcfg = load_config("research"), load_config("models") record = [] for design in ("event", "calendar"): df = pl.read_parquet(P.processed / f"sample_{design}.parquet") if design == "calendar": df = df.with_columns(pl.lit(0).alias("h")) # calendar: h varies; pooled model fc = feature_columns(df) write_json(P.evaluation / f"feature_columns_{design}.json", fc) horizons = sorted(df["h"].unique().to_list()) if design == "event" else [0] for h in horizons: d = df.filter(pl.col("h") == h) if design == "event" else df targets = TARGETS if (design == "event" and h == cfg["cutoffs"]["primary_horizon_td"]) else ["y_adverse_opinc"] for target in targets: dd = d.filter(pl.col(target).is_not_null()) sp = chronological(dd, cfg["splits"]) chk = overlap_check(sp["train"], sp["test"]) assert chk["shared_observations"] == 0 and chk["train_outcomes_revealed_in_test"] == 0, chk sets = list(INFO_SETS) if target == "y_adverse_opinc" else ["B1", "B2", "B5"] run_split(P, cfg, mcfg, fc, design, h, target, sp["train"], sp["validation"], sp["test"], "chrono", log, sets, record) if design == "event" and h == cfg["cutoffs"]["primary_horizon_td"] and target == "y_adverse_opinc": # continuous target (ridge) on the same split for name in ("B2", "B5"): cols = columns_for(name, fc); pp = Preprocessor(cols).fit(sp["train"]) ytr = sp["train"]["S_opinc"].to_numpy(); lo, hi = np.quantile(ytr, [0.01, 0.99]); ytr = np.clip(ytr, lo, hi) m, info = fit_ridge(pp.transform(sp["train"]), ytr, pp.transform(sp["validation"]), np.clip(sp["validation"]["S_opinc"].to_numpy(), lo, hi)) save_preds(P, sp["test"], m.predict(pp.transform(sp["test"])), "chrono_cont", design, h, "S_opinc", name, "test") record.append({"design": design, "h": h, "target": "S_opinc", "model": name, "tag": "chrono_cont", "winsor": [float(lo), float(hi)], "train_mean": float(ytr.mean()), **info}) # firm hold-out: train/validate on non-held-out firms, test on both groups separately hold = firm_holdout_mask(dd["entity_id"], cfg["splits"]["firm_holdout_fraction"]) dd2 = dd.with_columns(pl.Series("_hold", hold)) sp2 = chronological(dd2.filter(~pl.col("_hold")), cfg["splits"]) te_hold = dd2.filter(pl.col("_hold") & (pl.col("cutoff") >= sp2["test"]["cutoff"].min())) run_split(P, cfg, mcfg, fc, design, h, target, sp2["train"], sp2["validation"], te_hold, "firmhold_heldout", log, ["B2", "B5"], record) run_split(P, cfg, mcfg, fc, design, h, target, sp2["train"], sp2["validation"], sp2["test"], "firmhold_seen", log, ["B2", "B5"], record) # walk-forward out-of-fold predictions (logistic, C fixed at the chronological choice per set) for name in ("B1", "B2", "B3", "B4", "B5"): cols = columns_for(name, fc) C = next(r["C"] for r in record if r["model"] == name and r["tag"] == "chrono" and r["h"] == h and r["target"] == target) parts = [] for y, tr, te in walk_forward(dd, cfg["splits"], 2026): pp = Preprocessor(cols).fit(tr) from sklearn.linear_model import LogisticRegression m = LogisticRegression(C=C, max_iter=2000, random_state=cfg["seed"]).fit(pp.transform(tr), tr[target].to_numpy().astype(int)) parts.append(te.select(KEY + [target]).with_columns([pl.Series("p", predict_proba(m, pp.transform(te))), pl.lit(y).alias("fold")])) oof = pl.concat(parts).with_columns([pl.lit(name).alias("model"), pl.lit("oof").alias("part"), pl.lit("walkforward").alias("tag")]) oof.write_parquet(P.evaluation / "preds" / f"{design}_h{h}_{target}_{name}_oof_walkforward.parquet") log.info("walk-forward %s: %d oof rows", name, oof.height) write_json(P.evaluation / "baselines_fit_log.json", {"run_at": now_iso(), "fits": record}) print(json.dumps({"fits": len(record)}, indent=1)) return 0 if __name__ == "__main__": raise SystemExit(main())