1#!/usr/bin/env python2"""Phase 4: LightGBM (B6) with an identical random-search budget for every information set on the chronological split3(primary horizon, primary target + secondary targets), horizon ladder and walk-forward refits with a fixed configuration4(no tuning inside folds, so earlier folds never see later data). No neural/text models (DECISIONS D-006)."""5from __future__ import annotations67import json8import sys9from pathlib import Path1011import numpy as np12import polars as pl1314sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))15from wp39.models import INFO_SETS, Preprocessor, columns_for, fit_lightgbm, predict_proba # noqa: E40216from wp39.pipeline import Paths, get_logger, load_config # noqa: E40217from wp39.provenance import now_iso, write_json # noqa: E40218from wp39.samples import feature_columns # noqa: E40219from wp39.splits import chronological, firm_holdout_mask, walk_forward # noqa: E4022021KEY = ["entity_id", "symbol", "fiscal_period_end", "h", "cutoff", "A_q", "sic_division", "release_time"]22FIXED = {"num_leaves": 15, "learning_rate": 0.03, "min_child_samples": 100, "feature_fraction": 0.7, "bagging_fraction": 0.8,23 "bagging_freq": 1, "lambda_l2": 1.0, "n_estimators": 600, "verbose": -1, "n_jobs": 12}242526def save(P, df, p, design, h, target, model, part, tag, extra=None):27 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")])28 if extra is not None:29 out = out.with_columns(extra)30 out.write_parquet(P.evaluation / "preds" / f"{design}_h{h}_{target}_{model}_{part}_{tag}.parquet")313233def tuned(P, cfg, mcfg, fc, design, h, target, sp, tag, sets, log, record):34 ytr, yva = sp["train"][target].to_numpy().astype(int), sp["validation"][target].to_numpy().astype(int)35 for name in sets:36 cols = columns_for(name, fc); pp = Preprocessor(cols).fit(sp["train"])37 m, info = fit_lightgbm(pp.transform(sp["train"]), ytr, pp.transform(sp["validation"]), yva, mcfg["lightgbm"], cfg["seed"], n_jobs=12)38 save(P, sp["validation"], predict_proba(m, pp.transform(sp["validation"])), design, h, target, "L" + name, "validation", tag)39 save(P, sp["test"], predict_proba(m, pp.transform(sp["test"])), design, h, target, "L" + name, "test", tag)40 imp = dict(zip(cols, (m.booster_.feature_importance("gain") / max(m.booster_.feature_importance("gain").sum(), 1)).round(4).tolist()))41 record.append({"design": design, "h": h, "target": target, "model": "L" + name, "tag": tag, "n_features": len(cols),42 "val_score": info["val_score"], "params": info["params"], "trials": len(info["trials"]), "importance_gain": imp})43 log.info("%s h=%d %s L%s %s: val_ll=%.4f params=%s", design, h, target, name, tag, info["val_score"], info["params"])444546def fixed_fit(train, test, cols, target, seed):47 import lightgbm as lgb48 pp = Preprocessor(cols).fit(train)49 # early stopping on the last 4 calendar quarters of the training window (purged by construction: still before test)50 cut = train["cutoff"].quantile(0.85)51 tr, va = train.filter(pl.col("cutoff") < cut), train.filter(pl.col("cutoff") >= cut)52 m = lgb.LGBMClassifier(random_state=seed, **FIXED)53 m.fit(pp.transform(tr), tr[target].to_numpy().astype(int), eval_set=[(pp.transform(va), va[target].to_numpy().astype(int))],54 callbacks=[lgb.early_stopping(50, verbose=False)])55 return predict_proba(m, pp.transform(test)), int(m.best_iteration_ or 0)565758def main() -> int:59 P = Paths.default()60 log = get_logger("06_advanced", P.logs / "06_advanced.log")61 cfg, mcfg = load_config("research"), load_config("models")62 hp = cfg["cutoffs"]["primary_horizon_td"]63 record = []64 df = pl.read_parquet(P.processed / "sample_event.parquet")65 fc = feature_columns(df)66 # 1) primary: tuned LightGBM, all information sets, h = 20, primary target67 d = df.filter((pl.col("h") == hp) & pl.col("y_adverse_opinc").is_not_null())68 sp = chronological(d, cfg["splits"])69 tuned(P, cfg, mcfg, fc, "event", hp, "y_adverse_opinc", sp, "chrono", list(INFO_SETS), log, record)70 # 2) secondary targets (B2 vs B5 only), tuned71 for target in ("y_adverse_netinc", "y_roa_decline", "y_ocf_negative", "y_liquidity_deterioration", "y_abs_surprise"):72 dt_ = df.filter((pl.col("h") == hp) & pl.col(target).is_not_null())73 tuned(P, cfg, mcfg, fc, "event", hp, target, chronological(dt_, cfg["splits"]), "chrono", ["B2", "B5"], log, record)74 # 3) horizon ladder (fixed config): B2 vs B5 at every h on the chronological split75 for h in sorted(df["h"].unique().to_list()):76 dh = df.filter((pl.col("h") == h) & pl.col("y_adverse_opinc").is_not_null())77 sph = chronological(dh, cfg["splits"])78 for name in ("B2", "B5"):79 p, it = fixed_fit(pl.concat([sph["train"], sph["validation"]]), sph["test"], columns_for(name, fc), "y_adverse_opinc", cfg["seed"])80 save(P, sph["test"], p, "event", h, "y_adverse_opinc", "L" + name, "test", "fixed")81 record.append({"design": "event", "h": h, "target": "y_adverse_opinc", "model": "L" + name, "tag": "fixed", "best_iteration": it})82 # 4) firm hold-out (fixed config)83 hold = firm_holdout_mask(d["entity_id"], cfg["splits"]["firm_holdout_fraction"])84 d2 = d.with_columns(pl.Series("_hold", hold))85 sp2 = chronological(d2.filter(~pl.col("_hold")), cfg["splits"])86 te_hold = d2.filter(pl.col("_hold") & (pl.col("cutoff") >= sp2["test"]["cutoff"].min()))87 for name in ("B2", "B5"):88 trv = pl.concat([sp2["train"], sp2["validation"]])89 p, _ = fixed_fit(trv, te_hold, columns_for(name, fc), "y_adverse_opinc", cfg["seed"]); save(P, te_hold, p, "event", hp, "y_adverse_opinc", "L" + name, "test", "firmhold_heldout")90 p, _ = fixed_fit(trv, sp2["test"], columns_for(name, fc), "y_adverse_opinc", cfg["seed"]); save(P, sp2["test"], p, "event", hp, "y_adverse_opinc", "L" + name, "test", "firmhold_seen")91 # 5) walk-forward out-of-fold (fixed config) for B1..B5 at h = 20 -> LFDS for the panel regressions / H4 / H592 for name in ("B1", "B2", "B3", "B4", "B5"):93 parts = []94 for y, tr, te in walk_forward(d, cfg["splits"], 2026):95 p, it = fixed_fit(tr, te, columns_for(name, fc), "y_adverse_opinc", cfg["seed"])96 parts.append(te.select(KEY + ["y_adverse_opinc"]).with_columns([pl.Series("p", p), pl.lit(y).alias("fold")]))97 oof = pl.concat(parts).with_columns([pl.lit("L" + name).alias("model"), pl.lit("oof").alias("part"), pl.lit("walkforward").alias("tag")])98 oof.write_parquet(P.evaluation / "preds" / f"event_h{hp}_y_adverse_opinc_L{name}_oof_walkforward.parquet")99 log.info("walk-forward L%s: %d oof rows", name, oof.height)100 # 6) calendar design: walk-forward OOF for B2/B5 (scores for H5 portfolio sorts)101 dc = pl.read_parquet(P.processed / "sample_calendar.parquet").with_columns(pl.lit(0).alias("h"))102 fcc = feature_columns(dc)103 for name in ("B2", "B5"):104 parts = []105 for y, tr, te in walk_forward(dc, cfg["splits"], 2026):106 p, _ = fixed_fit(tr, te, columns_for(name, fcc), "y_adverse_opinc", cfg["seed"])107 parts.append(te.select(KEY + ["y_adverse_opinc"]).with_columns([pl.Series("p", p), pl.lit(y).alias("fold")]))108 oof = pl.concat(parts).with_columns([pl.lit("L" + name).alias("model"), pl.lit("oof").alias("part"), pl.lit("walkforward").alias("tag")])109 oof.write_parquet(P.evaluation / "preds" / f"calendar_h0_y_adverse_opinc_L{name}_oof_walkforward.parquet")110 write_json(P.evaluation / "advanced_fit_log.json", {"run_at": now_iso(), "fixed_config": FIXED, "fits": record})111 print(json.dumps({"fits": len(record)}, indent=1))112 return 0113114115if __name__ == "__main__":116 raise SystemExit(main())117