""" Strategy (Extended): Walk-Forward Factor Selection with Meta-Selection WHAT THIS IS ============ 04-StrategyTemplate.py with its EXTENSION POINTs filled in. The four-stage architecture, the timing rules and EVERY hyperparameter are unchanged. The only difference is that the candidate MENU is larger: three new ideas were added as options, and the walk-forward meta layer decides -- out of sample, using only past data -- whether they earn their place. WHAT WAS ADDED (and why, stated before any result was looked at) ================================================================ 1. EXTENSION POINT 2 -- selection on |Sharpe| with sign-flipping. The baseline can only hold a factor long, as constructed. A characteristic with a persistently NEGATIVE Sharpe is just as tradable -- short it. This roughly doubles the usable factor universe at zero extra data cost. Added: "AbsSharpe" and "ClusterAbsSharpe". 2. EXTENSION POINT 3 -- risk-based factor weighting. EW ignores risk entirely; MeanVar needs a mean estimate, which is the noisiest input in finance. Inverse-volatility and equal-risk-contribution need only a covariance, so they are the standard middle ground. Added: "InvVol" and "RiskParity". 3. EXTENSION POINT 4 -- a wider diversification rung, n_factors = 20. The baseline offers 5 and 10. If diversification helps, the meta layer should be allowed to say so. Menu: 3 windows x 3 n_factors x 4 selection rules x 5 weighting rules = 180 configurations (the baseline's 36 are a strict subset). WHAT WAS **NOT** CHANGED ======================== No hyperparameter was re-tuned. REBAL_MONTHS, META_TOP_K, META_MIN_OBS, MIN_OBS, MIN_STOCKS, MIN_HISTORY, SHRINKAGE and N_CLUSTERS_MULT hold their baseline values. The meta-criterion is still the expanding out-of-sample Sharpe. Nothing was adjusted after seeing test-period results -- that would re-introduce the configuration selection bias this design exists to remove. TEMPORAL INTEGRITY IS UNCHANGED =============================== The sign of a flipped factor is itself a decision, so it is chosen at each rebalance from the SAME lagged history as everything else (LS rows <= rp - 1 month) and then carried, fixed, through the holding block and into the stock-level weights. No stage outside Stage 1 reads ret_exc_lead1m. RUN IT ====== python 05-StrategyExtended.py --data C:/data/jkp CTF Admin Modifications (2026-09-24): -------------------------------------- 1. Narrowed the two exception handlers in select_cluster_sharpe and select_cluster_abs_sharpe from "except Exception" to "except (ValueError, RuntimeError)". Reason: the pipeline's security checks reject exception handlers that catch every exception without re-raising. The fallback to plain Sharpe ranking is unchanged for the errors clustering actually raises on a degenerate distance matrix. 2. Added flush=True to the print statements on the path main() runs. Reason: ensures progress output appears in the job log immediately rather than when the output buffer fills. 3. Added [CTF-DEBUG] progress statements to the Stage 1, Stage 2 and Stage 4 loops. Reason: each stage previously reported only its start and end, which makes a long run on the full dataset hard to monitor. """ import argparse import time import numpy as np import pandas as pd from scipy.cluster.hierarchy import linkage, fcluster from scipy.spatial.distance import squareform # --------------------------------------------------------------------------- # Hyperparameters -- IDENTICAL to 04-StrategyTemplate.py. Not re-tuned. # --------------------------------------------------------------------------- REBAL_MONTHS = 6 # rebalance / meta-selection frequency (months) META_TOP_K = 3 # how many configs the meta layer blends META_MIN_OBS = 120 # months of config history before meta-selection starts MIN_OBS = 60 # months of data required for a factor to be usable MIN_STOCKS = 10 # min stocks per side of a factor portfolio MIN_HISTORY = 24 # months of history before the first rebalance SHRINKAGE = 0.3 # covariance shrinkage in mean-variance weighting N_CLUSTERS_MULT = 3 # clusters = n_factors * this, in ClusterSharpe ERC_ITERS = 60 # fixed-point iterations for RiskParity (deterministic) # =========================================================================== # Stage 1: factor portfolio returns (UNCHANGED from the baseline) # =========================================================================== def compute_factor_ls_returns(chars, feature_names, min_stocks, min_obs): """ Linear-rank long-short return for every feature, every month. CONVENTION: the LS value stored at month t uses ret_exc_lead1m, i.e. it is the return earned over month t+1, and is KNOWN only at the end of month t+1. All later stages respect this. """ print(" Stage 1: factor LS returns...", flush=True) t0 = time.time() valid_features = [f for f in feature_names if f in chars.columns] df = chars[["eom", "ret_exc_lead1m"] + valid_features] df = df[df["ret_exc_lead1m"].notna()] months = sorted(df["eom"].unique()) ls_matrix = np.full((len(months), len(valid_features)), np.nan) for m_idx, eom in enumerate(months): if m_idx % 120 == 0: print(f"[CTF-DEBUG] Stage 1: month {m_idx}/{len(months)} " f"({time.time()-t0:.0f}s)", flush=True) month_data = df[df["eom"] == eom] if len(month_data) < min_stocks * 2: continue ret = month_data["ret_exc_lead1m"].values ranks = month_data[valid_features].rank(pct=True).values w = np.nan_to_num(ranks, nan=0.5) - 0.5 not_nan = ~np.isnan(month_data[valid_features].values) w = w * not_nan num = (ret[:, None] * w).sum(axis=0) den = np.abs(w).sum(axis=0) ok = (not_nan.sum(axis=0) >= min_stocks * 2) & (den > 1e-10) row = np.full(len(valid_features), np.nan) row[ok] = num[ok] / den[ok] ls_matrix[m_idx] = row ls_df = pd.DataFrame(ls_matrix, index=pd.to_datetime(months), columns=valid_features).sort_index() ls_df = ls_df[ls_df.columns[ls_df.notna().sum() >= min_obs]] print(f" {ls_df.shape[0]} months x {ls_df.shape[1]} factors " f"({time.time()-t0:.0f}s)", flush=True) return ls_df # =========================================================================== # Selection rules: history -> (ordered factor names, signs) # # Signature change vs the baseline: every rule now returns a (factors, signs) # pair. signs[k] is +1 to hold factor k as constructed, -1 to short it. The # baseline rules always return +1, so they behave exactly as before. # =========================================================================== def _sharpe_series(tc): stds = tc.std().replace(0, np.nan) return ((tc.mean() / stds) * np.sqrt(12)).fillna(0) def _signs_from(scores, factors): """+1/-1 per factor, never 0 (a dead-flat factor is held long).""" s = np.sign(scores[factors].values).astype(float) s[s == 0] = 1.0 return s def select_sharpe(tc, valid, n_factors): """Top-N factors by annualized Sharpe over the lookback slice (long only).""" picks = list(_sharpe_series(tc[valid]).nlargest( min(n_factors, len(valid))).index) return picks, np.ones(len(picks)) def select_abs_sharpe(tc, valid, n_factors): """ EXTENSION POINT 2. Top-N by |Sharpe|, shorting any factor whose Sharpe is negative. A reliably bad characteristic is a reliably good short; the baseline simply discards that information. """ scores = _sharpe_series(tc[valid]) picks = list(scores.abs().nlargest(min(n_factors, len(valid))).index) return picks, _signs_from(scores, picks) def _cluster_pick(tc, valid, n_factors, scores, rank_by): """ Shared machinery for the cluster rules: cluster factors on (1 - |corr|), then take the best factor from each of the top N clusters. Correlation uses |corr| so a factor and its mirror image land in the same cluster -- which is exactly what sign-flipping needs. """ tcv = tc[valid] corr = tcv.corr().fillna(0).clip(-1, 1) dist = (1.0 - corr.abs()).values.copy() np.fill_diagonal(dist, 0) dist = np.clip((dist + dist.T) / 2, 0, None) condensed = np.nan_to_num(squareform(dist, checks=False), nan=1.0, posinf=1.0, neginf=0.0) Z = linkage(condensed, method="average") n_clusters = min(max(n_factors * N_CLUSTERS_MULT, 20), len(valid)) labels = fcluster(Z, t=n_clusters, criterion="maxclust") cdf = pd.DataFrame({"factor": valid, "cluster": labels, "score": rank_by.fillna(-999).values}) best = cdf.loc[cdf.groupby("cluster")["score"].idxmax()] best = best[best["score"] > -998].sort_values("score", ascending=False) return best.head(n_factors)["factor"].tolist() def select_cluster_sharpe(tc, valid, n_factors): """Diversified top-N by Sharpe (long only) -- the baseline rule.""" if len(tc) < 12 or len(valid) < n_factors: return select_sharpe(tc, valid, n_factors) scores = _sharpe_series(tc[valid]) try: picks = _cluster_pick(tc, valid, n_factors, scores, scores) # CTF-FIX: catch only what clustering raises on a degenerate distance matrix except (ValueError, RuntimeError): return select_sharpe(tc, valid, n_factors) return picks, np.ones(len(picks)) def select_cluster_abs_sharpe(tc, valid, n_factors): """EXTENSION POINT 2. Diversified top-N by |Sharpe|, with sign-flipping.""" if len(tc) < 12 or len(valid) < n_factors: return select_abs_sharpe(tc, valid, n_factors) scores = _sharpe_series(tc[valid]) try: picks = _cluster_pick(tc, valid, n_factors, scores, scores.abs()) # CTF-FIX: catch only what clustering raises on a degenerate distance matrix except (ValueError, RuntimeError): return select_abs_sharpe(tc, valid, n_factors) return picks, _signs_from(scores, picks) SELECT_FN = { "Sharpe": select_sharpe, "ClusterSharpe": select_cluster_sharpe, "AbsSharpe": select_abs_sharpe, # EXTENSION POINT 2 "ClusterAbsSharpe": select_cluster_abs_sharpe, # EXTENSION POINT 2 } # =========================================================================== # Weighting rules: signed history -> weight vector (sums to 1) # # Signature change vs the baseline: each rule receives `tcs`, the lookback # slice of the chosen factors ALREADY multiplied by their signs. So every # rule sees series whose expected return it is trying to harvest, and the # baseline rules are unchanged in behaviour. # =========================================================================== def weight_ew(tcs): n = tcs.shape[1] return np.full(n, 1.0 / n) def weight_rank_sharpe(tcs): """Best factor by Sharpe gets the largest weight (rank weights).""" order = _sharpe_series(tcs).rank(method="first") w = order.values.astype(float) return w / w.sum() def weight_mean_var(tcs): """w ~ Sigma^{-1} mu with diagonal shrinkage; long-only, sums to 1.""" tcf = tcs.fillna(0) n = tcs.shape[1] if len(tcf) < max(12, n + 2): return np.full(n, 1.0 / n) mu, cov = tcf.mean().values, tcf.cov().values cov = (1 - SHRINKAGE) * cov + SHRINKAGE * (np.trace(cov) / n) * np.eye(n) try: w = np.linalg.inv(cov) @ mu except np.linalg.LinAlgError: return np.full(n, 1.0 / n) w = np.maximum(w, 0) return w / w.sum() if w.sum() > 1e-10 else np.full(n, 1.0 / n) def weight_inv_vol(tcs): """ EXTENSION POINT 3. w_k ~ 1 / sigma_k. Uses only a volatility estimate -- no expected return, which is the noisiest thing one can estimate. """ n = tcs.shape[1] sd = tcs.fillna(0).std().values if not np.isfinite(sd).all() or (sd <= 1e-12).any(): return np.full(n, 1.0 / n) w = 1.0 / sd return w / w.sum() def weight_risk_parity(tcs): """ EXTENSION POINT 3. Equal risk contribution: every factor supplies the same share of portfolio variance. Deterministic fixed-point iteration from equal weights; falls back to EW if the covariance is degenerate. """ n = tcs.shape[1] tcf = tcs.fillna(0) if len(tcf) < max(12, n + 2): return np.full(n, 1.0 / n) cov = tcf.cov().values cov = (1 - SHRINKAGE) * cov + SHRINKAGE * (np.trace(cov) / n) * np.eye(n) if not np.isfinite(cov).all(): return np.full(n, 1.0 / n) w = np.full(n, 1.0 / n) for _ in range(ERC_ITERS): mrc = cov @ w # marginal risk contribution if not np.isfinite(mrc).all() or (np.abs(mrc) < 1e-18).any(): return np.full(n, 1.0 / n) w = w * (1.0 / mrc) # push weight toward low-risk names s = w.sum() if not np.isfinite(s) or s <= 1e-18: return np.full(n, 1.0 / n) w = w / s w = np.maximum(w, 0) return w / w.sum() if w.sum() > 1e-10 else np.full(n, 1.0 / n) WEIGHT_FN = { "EW": weight_ew, "RankSharpe": weight_rank_sharpe, "MeanVar": weight_mean_var, "InvVol": weight_inv_vol, # EXTENSION POINT 3 "RiskParity": weight_risk_parity, # EXTENSION POINT 3 } # =========================================================================== # The candidate menu. Still a full symmetric cross, just a wider one. # =========================================================================== CONFIG_GRID = [ {"window": w, "n_factors": n, "selection": s, "weighting": g} for w in (10, 20, None) # lookback years; None = expanding for n in (5, 10, 20) # EXTENSION POINT 4: added the 20 rung for s in SELECT_FN # selection rules registered above for g in WEIGHT_FN # weighting rules registered above ] def config_label(cfg): w = "all" if cfg["window"] is None else f"{cfg['window']}y" return f"{w}/{cfg['n_factors']}/{cfg['selection']}/{cfg['weighting']}" # =========================================================================== # Stage 2: walk-forward simulation of every config # =========================================================================== def simulate_configs(ls_returns, configs, rebal_months, min_obs, min_history): """ At each rebalance rp, every config selects factors/signs/weights from LS rows <= rp - 1 month, then earns the realized LS returns until the next rebalance. Returns the per-config schedules and out-of-sample returns. """ print(" Stage 2: walk-forward config simulation...", flush=True) t0 = time.time() all_months = ls_returns.index.sort_values() rebal_points = list(all_months[::rebal_months]) oos = pd.DataFrame(np.nan, index=all_months, columns=range(len(configs))) schedule = {} sel_cache = {} for i, rp in enumerate(rebal_points): if i % 10 == 0: print(f"[CTF-DEBUG] Stage 2: rebalance {i}/{len(rebal_points)} " f"({time.time()-t0:.0f}s)", flush=True) sel_end = rp - pd.DateOffset(months=1) # last KNOWN LS row ls_avail = ls_returns.loc[:sel_end] if len(ls_avail.dropna(how="all")) < min_history: continue next_rp = rebal_points[i + 1] if i + 1 < len(rebal_points) else None block = (all_months[(all_months >= rp) & (all_months < next_rp)] if next_rp is not None else all_months[all_months >= rp]) threshold = min(min_obs, max(6, len(ls_avail) // 2)) valid = list(ls_avail.columns[ls_avail.notna().sum() >= threshold]) if not valid: continue per_cfg = {} for ci, cfg in enumerate(configs): if cfg["window"] is None: tc = ls_avail else: start = (sel_end - pd.DateOffset(years=cfg["window"]) + pd.DateOffset(months=1)) tc = ls_avail.loc[start:] tc = tc.fillna(0) if len(tc) < 12: continue key = (rp, cfg["window"], cfg["selection"], cfg["n_factors"]) if key not in sel_cache: sel_cache[key] = SELECT_FN[cfg["selection"]]( tc, valid, cfg["n_factors"]) factors, signs = sel_cache[key] if not factors: continue # Every weighting rule sees the SIGNED history, so a shorted # factor is treated as the positive-carry series it now is. tcs = tc[factors] * signs weights = WEIGHT_FN[cfg["weighting"]](tcs) per_cfg[ci] = (factors, signs, weights) oos.loc[block, ci] = ( (ls_returns.loc[block, factors].fillna(0).values * signs) @ weights) if per_cfg: schedule[rp] = per_cfg print(f" {len(configs)} configs x {len(schedule)} rebalances " f"({time.time()-t0:.0f}s)", flush=True) return rebal_points, schedule, oos # =========================================================================== # Stage 3: meta-selection (UNCHANGED from the baseline) # =========================================================================== def meta_select(rebal_points, schedule, oos, configs, top_k, meta_min_obs): """ At each rebalance, blend the top_k configs by the annualized Sharpe of their OWN out-of-sample returns up to t-1. Until any config has meta_min_obs months of history, hold an equal blend of all configs. """ print(" Stage 3: meta-selection...", flush=True) chosen, log = {}, [] for rp in rebal_points: if rp not in schedule: continue active = list(schedule[rp].keys()) hist = oos.loc[:rp - pd.DateOffset(months=1)] counts = hist.notna().sum() eligible = [ci for ci in active if counts[ci] >= meta_min_obs] if eligible: score = {} for ci in eligible: s = hist[ci].dropna() sd = s.std() score[ci] = (s.mean() / sd) * np.sqrt(12) if sd > 0 else -np.inf top = sorted(score, key=score.get, reverse=True)[:top_k] mode = "meta" else: top, mode = active, "ew_prior" chosen[rp] = top log.append({"rebal": rp, "mode": mode, "chosen": ";".join(config_label(configs[ci]) for ci in top)}) n_meta = sum(1 for r in log if r["mode"] == "meta") print(f" {len(chosen)} rebalances ({n_meta} meta-selected)", flush=True) return chosen, log # =========================================================================== # Stage 4: stock-level weights # =========================================================================== def compute_stock_weights(chars, schedule, chosen): """ Blend the chosen configs equally and map factor weights to stocks: w_i = (1/K) * sum_cfg sum_k s_k * fw_k * (rank_i - 0.5) / sum_j|rank_j - 0.5| where s_k is the factor's sign. Each config's stock portfolio exactly replicates its factor-level return, so Stage 3's statistics describe the portfolio actually held. """ print(" Stage 4: stock-level weights...", flush=True) t0 = time.time() rebal_dates = sorted(chosen.keys()) if not rebal_dates: return pd.DataFrame(columns=["id", "eom", "w"]) all_factors = sorted({f for rp in rebal_dates for ci in chosen[rp] for f in schedule[rp][ci][0]}) available = [f for f in all_factors if f in chars.columns] df_work = chars[["id", "eom"] + available] months = sorted(df_work["eom"].unique()) results = [] for m_idx, eom in enumerate(months): if m_idx % 120 == 0: print(f"[CTF-DEBUG] Stage 4: month {m_idx}/{len(months)} " f"({time.time()-t0:.0f}s)", flush=True) applicable = [d for d in rebal_dates if d <= eom] if not applicable: continue rp = applicable[-1] cfg_ids = chosen[rp] month_data = df_work[df_work["eom"] == eom] if len(month_data) < 50: continue needed = sorted({f for ci in cfg_ids for f in schedule[rp][ci][0] if f in month_data.columns}) if not needed: continue ranks_all = month_data[needed].rank(pct=True) stock_w = np.zeros(len(month_data)) blend = 1.0 / len(cfg_ids) for ci in cfg_ids: factors, signs, weights = schedule[rp][ci] for factor, sgn, fw in zip(factors, signs, weights): if factor not in ranks_all.columns: continue r = ranks_all[factor].values nan_mask = np.isnan(r) rw = np.where(nan_mask, 0.0, r - 0.5) denom = np.abs(rw).sum() if denom < 1e-10 or (~nan_mask).sum() < MIN_STOCKS * 2: continue stock_w += blend * sgn * fw * rw / denom nz = np.abs(stock_w) > 1e-15 if nz.any(): results.append(pd.DataFrame({"id": month_data["id"].values[nz], "eom": eom, "w": stock_w[nz]})) output = pd.concat(results, ignore_index=True) print(f" {len(output):,} rows, {output['eom'].nunique()} months " f"({time.time()-t0:.0f}s)", flush=True) return output # =========================================================================== # Entry point (course interface): main(chars, features, daily_ret) # =========================================================================== def main(chars, features, daily_ret): feature_names = features["features"].tolist() chars = chars.copy() chars["eom"] = pd.to_datetime(chars["eom"]) n_months = chars["eom"].nunique() print(f"Data: {len(chars):,} stock-months, {len(feature_names)} " f"features, {n_months} months", flush=True) if n_months < 200: # small validation datasets: loosen thresholds min_obs, min_stocks = max(6, n_months // 5), 5 min_history, meta_min_obs = 12, max(24, n_months // 3) else: min_obs, min_stocks = MIN_OBS, MIN_STOCKS min_history, meta_min_obs = MIN_HISTORY, META_MIN_OBS ls = compute_factor_ls_returns(chars, feature_names, min_stocks, min_obs) rebal_points, schedule, oos = simulate_configs( ls, CONFIG_GRID, REBAL_MONTHS, min_obs, min_history) chosen, meta_log = meta_select( rebal_points, schedule, oos, CONFIG_GRID, META_TOP_K, meta_min_obs) main.meta_log = meta_log # diagnostics for graders/notebooks main.config_oos = oos main.config_labels = [config_label(c) for c in CONFIG_GRID] output = compute_stock_weights(chars, schedule, chosen) assert set(output.columns) == {"id", "eom", "w"} assert output["w"].notna().all() and len(output) > 0 return output # =========================================================================== # Stand-alone runner with evaluation # =========================================================================== if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--data", required=True, help="directory containing ctff_*.parquet") ap.add_argument("--out", default=None, help="path for the weights CSV " "(default: /weights.csv)") args = ap.parse_args() chars = pd.read_parquet(f"{args.data}/ctff_chars.parquet") features = pd.read_parquet(f"{args.data}/ctff_features.parquet") weights = main(chars, features, pd.DataFrame(columns=["id", "date", "ret_exc"])) # Gross portfolio returns: weights at eom t earn ret_exc_lead1m of t chars["eom"] = pd.to_datetime(chars["eom"]) weights["eom"] = pd.to_datetime(weights["eom"]) merged = weights.merge( chars[["id", "eom", "ret_exc_lead1m"]].dropna(), on=["id", "eom"], how="inner") port = (merged["w"] * merged["ret_exc_lead1m"]).groupby( merged["eom"]).sum().sort_index() def sharpe(r): return r.mean() / r.std() * np.sqrt(12) if len(r) >= 12 else np.nan print("\n SHARPE RATIOS BY PERIOD (gross)") for name, (a, b) in { "1960-1989": ("1960", "1989"), "1990-2003": ("1990", "2003"), "2004-2013": ("2004", "2013"), "2014-2023": ("2014", "2023"), "1990-2023": ("1990", "2023"), }.items(): sub = port.loc[a:b] print(f" {name}: {sharpe(sub):7.3f} ({len(sub)} months)") # One-way turnover (per month, as fraction of gross book) piv = weights.pivot_table(index="eom", columns="id", values="w", fill_value=0.0) to = piv.diff().abs().sum(axis=1) / 2 print(f"\n Avg monthly one-way turnover: {to.iloc[1:].mean():.3f}") out_path = args.out or f"{args.data}/weights.csv" submission = weights[["id", "eom", "w"]].copy() submission["eom"] = submission["eom"].dt.strftime("%Y-%m-%d") submission.to_csv(out_path, index=False, float_format="%.11g") print(f" Wrote {len(submission):,} weight rows -> {out_path}")