"""CTF second submission: raw-feature LightGBM, daily-covariance mean-variance portfolio. Authors: Paul Geertsema and Agents (Claude Opus 5, Grok 4.6) Self-contained per Rule 5. Defines `main(chars, features, daily_ret) -> DataFrame` with columns `id`, `eom`, `w` (Rules 11-12). `lightgbm` is declared in `requirements.txt` (Rule 8); everything else is pre-installed. Arrays are `float32` (Rule 9). Every constant below is a literal fixed in advance: nothing is selected at runtime (Rules 1, 18). Signal. LightGBM on the 402 raw characteristics as supplied. There is no ranking, winsorization, or other characteristic transformation. Missing stays missing: LightGBM routes NaNs to the loss-reducing side of each split. The target is arctan of the within-month z-score of the forward return: the learner sees the cross-section, not the market level, and extreme tails cannot dominate the loss. `min_child_samples=2000` keeps leaves from over-fitting to individual firm-months. Portfolio. Mean-variance against a Ledoit-Wolf covariance of the trailing 756 daily returns ending at the position date, ridge `lambda = kappa trace(Sigma) / n`. Tikhonov: the same Markowitz problem with `A = Sigma + lambda I` in place of Sigma. Let `a = A^-1 mu`, `b = A^-1 1`. These are intermediates, not portfolios. w_gmv = b / (1'b) GMV portfolio, sums to 1 w_tan = tangency portfolio under A sums to 1 w = (w_tan - w_gmv) scaled to unit gross sums to 0 The code forms `w` as `(a - (1'a / 1'b) b) / ||.||_1`, which is that difference. The submitted portfolio is `w`. Kappa is scale-free in universe size and volatility. Lookahead (Rule 1). The fit for month *t* uses rows strictly before *t*. `ret_exc_lead1m` at month *s* is realised over *s+1*, so the model has seen only targets realised at or before *t*. The covariance window ends at `eom[t]` inclusive; daily returns during the held month are never used. Every quantity at *t* is a function of data at or before *t* alone, which is what Rule 1's truncation test checks. """ from __future__ import annotations import time import numpy as np import pandas as pd import lightgbm as lgb # -------------------------------------------------------------------------------------- # Frozen configuration (Rules 1, 18). # -------------------------------------------------------------------------------------- SEED = 42 N_THREADS = 32 # Rule 9 allocates 32 cores REFIT_EVERY = 12 # annual refit of the expanding-window tree MIN_TRAIN_MONTHS = 36 # must still run on Rule 14's 123-month panel # `deterministic` / `force_row_wise`: Rule 18, histogram reduction order vs thread count. LGBM_PARAMS = dict( n_estimators=1000, # Default = 100, so 10 times larger learning_rate=0.01, # Default = 0.1, so 1/10 times smaller num_leaves=255, # Default = 31, so 8 times larger: combined -> MUCH more compute min_child_samples=2000, # Default = 20, so 100x larger. The magic sauce that controls overfitting. random_state=SEED, # The rest is standard repro and perf settings n_jobs=N_THREADS, deterministic=True, force_row_wise=True, verbose=-1, ) SAMPLE_WINDOW_DAYS = 756 # three years of daily returns MIN_DAILY_OBS = 250 # days of history a stock needs to be held DAYS_PER_MONTH = 21 # daily variance -> monthly units MIN_NAMES = 50 # below this a cross-section is not a portfolio KAPPA = 1.0 # ridge as a multiple of the mean eigenvalue TARGET_COL = "ret_exc_lead1m" FEATURE_LIST_COL = "features" # -------------------------------------------------------------------------------------- # Features and target # -------------------------------------------------------------------------------------- # -------------------------------------------------------------------------------------- # The daily returns matrix # -------------------------------------------------------------------------------------- def build_daily_matrix( daily_ret: pd.DataFrame, panel_ids: np.ndarray ) -> tuple[np.ndarray, np.ndarray, np.ndarray]: """Pivot the long daily table to a dense (n_dates, n_ids) array. Restricted to the panel's identifiers: the raw table is CRSP-wide, the panel is already size-screened. Dense with NaN for "did not trade"; every access is a contiguous trailing window. returns: dates (n_dates,) datetime64[D] ascending, ids (n_ids,) float64 ascending, M (n_dates, n_ids) float32 with NaN where absent """ rid = daily_ret["id"].to_numpy() keep = np.isin(rid, panel_ids) rid = rid[keep] rdate = daily_ret["date"].to_numpy()[keep].astype("datetime64[D]") rret = daily_ret["ret_exc"].to_numpy()[keep].astype(np.float32) dates, date_ix = np.unique(rdate, return_inverse=True) ids, id_ix = np.unique(rid, return_inverse=True) M = np.full((len(dates), len(ids)), np.nan, dtype=np.float32) M[date_ix, id_ix] = rret print(f"daily matrix {M.shape[0]:,} days x {M.shape[1]:,} ids, " f"{np.isfinite(M).mean():.1%} dense, kept {keep.sum():,} of {len(keep):,} rows", flush=True) return dates, ids, M def trailing_window( M: np.ndarray, end: int, cols: np.ndarray ) -> tuple[np.ndarray, np.ndarray]: """Demeaned trailing daily returns for one month's universe. `end` is one past the last usable daily row, so the window is [end-756, end) and includes the position date. Missing observations are set to zero, then the column is demeaned: a non-trading day is treated as no move. Fill-then-demean makes columns exactly mean-zero, which `S = X'X / T` needs. returns: X (T, n) float32 column-demeaned, enough (n,) bool """ lo = max(0, end - SAMPLE_WINDOW_DAYS) X = M[lo:end, cols] # (T, n) float32 copy obs = np.isfinite(X) enough = obs.sum(axis=0) >= MIN_DAILY_OBS # (n,) X = np.where(obs, X, np.float32(0.0)) X -= X.mean(axis=0, keepdims=True) return X, enough # -------------------------------------------------------------------------------------- # The portfolio operator # -------------------------------------------------------------------------------------- def shrunk_covariance(X: np.ndarray) -> tuple[np.ndarray, float]: """Ledoit-Wolf (2004) linear shrinkage toward a scaled identity. (T, n) -> (n, n) `Sigma = (1 - delta) S + delta * (tr(S)/n) * I` with `S = X'X / T`. Delta is closed-form, so there is no intensity to choose. X: (T, n) column-demeaned, finite. returns: Sigma (n, n), intensity delta in [0, 1] """ T, n = X.shape assert T >= 2 and n >= 2, X.shape assert np.isfinite(X).all() S = (X.T @ X) / T # (n, n) float32 # Intensity scalars in float64: sums over up to n^2 terms, two of them differenced. tr_S = float(np.trace(S, dtype=np.float64)) mu_lw = tr_S / n norm_S2 = float(np.square(S).sum(dtype=np.float64)) # ||S||_F^2 d2 = norm_S2 - 2.0 * mu_lw * tr_S + n * mu_lw**2 # ||S - mu I||_F^2 # b2 = (1/T^2) sum_t ||x_t x_t' - S||_F^2, expanded so only row norms are needed. row_sq = np.square(X).sum(axis=1, dtype=np.float64) # (T,) ||x_t||^2 b2 = float(np.square(row_sq).sum()) / T**2 - norm_S2 / T delta = min(max(b2, 0.0), d2) / d2 if d2 > 1e-300 else 1.0 S *= np.float32(1.0 - delta) S[np.diag_indices(n)] += np.float32(delta * mu_lw) return S, delta def tan_minus_gmv( mu: np.ndarray, Sigma: np.ndarray, kappa: float ) -> np.ndarray: """Dollar-neutral tangency minus GMV, from one solve. The GMV portfolio is fully invested minimum-variance under `A`. The tangency portfolio is Markowitz optimal under `A` (Tikhonov: `A` in place of `Sigma`); its weights also sum to one. The submitted portfolio is that tangency minus the GMV portfolio, scaled to unit gross, so its weights sum to zero. `a = A^-1 mu` and `b = A^-1 1` are intermediates, not portfolios: w_gmv = b / (1'b) w = (a - (1'a / 1'b) b) / ||.||_1 The solve is in `A = Sigma + lambda I`. Ridge `lambda = kappa trace(Sigma)/n` is the mean eigenvalue at kappa = 1. mu (n,), Sigma (n, n) -> w (n,) at unit gross summing to 0 """ n = mu.shape[0] assert Sigma.shape == (n, n), (Sigma.shape, n) assert np.isfinite(mu).all() lam = np.float32(kappa * float(np.trace(Sigma, dtype=np.float64)) / n) A = Sigma + lam * np.eye(n, dtype=np.float32) one = np.ones(n, dtype=np.float32) sol = np.linalg.solve(A, np.column_stack([mu, one])) # (n, 2) float32 a, b = sol[:, 0], sol[:, 1] sum_a, sum_b = float(one @ a), float(one @ b) assert sum_b > 0.0, f"1'Sigma^-1 1 must be positive, got {sum_b}" w = a - np.float32(sum_a / sum_b) * b # tangency minus GMV, sums to 0 w /= np.float32(np.abs(w).sum()) return w # -------------------------------------------------------------------------------------- # Entry point # -------------------------------------------------------------------------------------- def build_weights( chars: pd.DataFrame, features: pd.DataFrame, daily_ret: pd.DataFrame ) -> pd.DataFrame: """Submitted weights. Columns id, eom, w. `main` then keeps only `ctff_test` rows. """ np.random.seed(SEED) # Rule 18 t0 = time.perf_counter() feats = [str(v) for v in features[FEATURE_LIST_COL].tolist()] assert set(feats) <= set(chars.columns), "feature list names a column chars lacks" print(f"{len(chars):,} rows, {len(feats)} characteristics", flush=True) # Sort by (eom, id) so every month is a contiguous slice and the panel order is fixed. panel = chars[["id", "eom", TARGET_COL] + feats] panel = panel.sort_values(["eom", "id"], kind="stable").reset_index(drop=True) ids = panel["id"].to_numpy() # float64, integral values eom_out = panel["eom"].to_numpy() # returned unchanged, Rule 12 months, month_ix = np.unique(eom_out.astype("datetime64[D]"), return_inverse=True) n_months = len(months) starts = np.searchsorted(month_ix, np.arange(n_months), side="left") ends = np.searchsorted(month_ix, np.arange(n_months), side="right") print(f"{n_months} months, {months[0]} .. {months[-1]}", flush=True) X = panel[feats].to_numpy(dtype=np.float32) # (n_rows, n_feat), NaN allowed assert X.shape == (len(panel), len(feats)), X.shape print(f"loaded raw features in {time.perf_counter() - t0:.0f}s", flush=True) # Cross-sectional z-score, then arctan: market level out of the loss, tails bounded. g = pd.DataFrame({"m": month_ix, "r": panel[TARGET_COL]}).groupby("m")["r"] z = (panel[TARGET_COL] - g.transform("mean")) / g.transform("std") # Formation or validation rows may lack a forward return; those labels are not used. y = np.arctan(z.to_numpy(dtype=np.float32)) # (n_rows,) float32, NaN allowed assert y.shape == (len(panel),), y.shape dates, daily_ids, M = build_daily_matrix(daily_ret, np.unique(ids)) cols_all = np.clip(np.searchsorted(daily_ids, ids), 0, len(daily_ids) - 1) has_daily = daily_ids[cols_all] == ids print(f"{has_daily.mean():.1%} of panel rows have a daily column", flush=True) # Emit every month with enough history. Rule 12 scores only `ctff_test` rows; Rule 14's # validation panel is not guaranteed to populate that flag. traded = list(range(MIN_TRAIN_MONTHS, n_months)) assert traded, f"{n_months} months is fewer than MIN_TRAIN_MONTHS = {MIN_TRAIN_MONTHS}" anchor = traded[0] print(f"trading {len(traded)} months, {months[anchor]} .. {months[traded[-1]]}", flush=True) model = None fit_secs = 0.0 frames: list[pd.DataFrame] = [] for k, t in enumerate(traded, start=1): lo, hi = int(starts[t]), int(ends[t]) fit_rows = int(starts[t]) # months strictly before t refit = (t - anchor) % REFIT_EVERY == 0 if refit: t_fit = time.perf_counter() model = lgb.LGBMRegressor(**LGBM_PARAMS) model.fit(X[:fit_rows], y[:fit_rows]) fit_secs = time.perf_counter() - t_fit # yhat is already a within-month standardised expected return; the per-month # scalar cancels in the unit-gross normalisation. Booster directly: sklearn # `predict` warns on missing `Column_i` names; the two paths are identical. mu = model.booster_.predict(X[lo:hi]).astype(np.float32) # (n_names,) cols = cols_all[lo:hi][has_daily[lo:hi]] # side="right": dates[end-1] <= months[t]. Do not clip end upward: that would # reach past the position date (Rule 1). end = int(np.searchsorted(dates, months[t], side="right")) assert end >= MIN_DAILY_OBS, ( f"month {months[t]}: {end} daily rows precede it, need {MIN_DAILY_OBS}" ) Xd, enough = trailing_window(M, end, cols) usable = np.zeros(hi - lo, dtype=bool) usable[np.flatnonzero(has_daily[lo:hi])[enough]] = True assert usable.sum() >= MIN_NAMES, ( f"month {months[t]}: only {usable.sum()} usable names" ) Sigma, delta = shrunk_covariance( Xd[:, enough] * np.float32(np.sqrt(DAYS_PER_MONTH)) ) w = np.zeros(hi - lo, dtype=np.float32) w[usable] = tan_minus_gmv(mu[usable], Sigma, KAPPA) month = pd.DataFrame({"id": ids[lo:hi], "eom": eom_out[lo:hi]}) frames.append(month.assign(w=w)) print(f"{k:>3d}/{len(traded)} {months[t]} " f"train {months[0]}..{months[t - 1]} obs {fit_rows:>9,} " f"{'REFIT ' + format(fit_secs, '.0f') + 's' if refit else 'reuse '} " f"names {hi - lo:>5,} held {int(usable.sum()):>5,} " f"Sigma {Xd.shape[0]:>3d}d " f"{dates[max(0, end - SAMPLE_WINDOW_DAYS)]}..{dates[end - 1]} " f"delta {delta:.3f} " f"net {w.sum():+.4f} gross {np.abs(w).sum():>6.3f}", flush=True) out = pd.concat(frames, ignore_index=True) out["id"] = out["id"].astype(np.int64) # Rule 12 wants an integer assert list(out.columns) == ["id", "eom", "w"], out.columns assert out.notna().all().all(), "output contains missing values" print(f"done: {len(out):,} rows in {(time.perf_counter() - t0) / 60:.1f} min", flush=True) return out def is_test_row(flag: pd.Series) -> pd.Series: """Rows the harness scores. Boolean, 0/1, and '0'/'1' all occur in the wild.""" if pd.api.types.is_bool_dtype(flag): return flag.fillna(False) if pd.api.types.is_numeric_dtype(flag): return flag.fillna(0).ne(0) return flag.astype(str).str.strip().str.lower().isin(["1", "1.0", "true", "t"]) def restrict_to_test(chars: pd.DataFrame, weights: pd.DataFrame) -> pd.DataFrame: """Keep rows the harness scores (`ctff_test` true). If the flag is missing or has no true rows (Rule 14's validation panel), return the weights unchanged so the DataFrame stays non-empty. """ if "ctff_test" not in chars.columns: return weights mask = is_test_row(chars["ctff_test"]) if not mask.any(): return weights keys = chars.loc[mask, ["id", "eom"]].drop_duplicates() keys["id"] = keys["id"].astype(np.int64) out = weights.merge(keys, on=["id", "eom"], how="inner") assert len(out), "ctff_test is set but no weights match those rows" return out def main(chars: pd.DataFrame, features: pd.DataFrame, daily_ret: pd.DataFrame) -> pd.DataFrame: """Produce monthly portfolio weights. Returns columns id, eom, w (Rule 12). The harness imports this. Only test-period rows are returned. Local `__main__` writes the full panel of weights so in-sample Sharpe is still available. """ weights = build_weights(chars, features, daily_ret) return restrict_to_test(chars, weights) # -------------------------------------------------------------------------------------- # Local only. The harness imports `main` (Rule 11) and never reaches this block. # -------------------------------------------------------------------------------------- if __name__ == "__main__": from pathlib import Path RAW = Path(r"C:\data\CTF\raw") OUT = Path(__file__).resolve().parent / "results" t_load = time.perf_counter() chars = pd.read_parquet(RAW / "ctff_chars.parquet") features = pd.read_parquet(RAW / "ctff_features.parquet") daily_ret = pd.read_parquet(RAW / "ctff_daily_ret.parquet") print(f"loaded chars {chars.shape}, features {features.shape}, " f"daily {daily_ret.shape} in {time.perf_counter() - t_load:.0f}s", flush=True) weights = build_weights(chars, features, daily_ret) submitted = restrict_to_test(chars, weights) csv_mb = len(submitted.to_csv(index=False).encode("utf-8")) / 1_048_576 print(f"\nfull panel {len(weights):,} rows; harness would receive {len(submitted):,}", flush=True) print(f"Rule 12: {len(submitted):,} rows, id {submitted['id'].dtype}, " f"serialised {csv_mb:.1f} MB as CSV (limit 50)", flush=True) OUT.mkdir(parents=True, exist_ok=True) path = OUT / "ctf_w.parquet" weights.to_parquet(path, index=False) print(f"wrote {path} ({path.stat().st_size / 1_048_576:.1f} MB)", flush=True) print("score with: python submission/ctf_score.py", flush=True)