| """Policy validation: does the construction rule actually improve risk-adjusted outcomes? |
| |
| For each seed book (concentrated tech / income / mid-cap), at each quarter-end over the |
| last ~5 years: estimate the shrinkage covariance from the trailing 2 years, pick 3 adds |
| from a fixed universe by (a) the greedy variance-reduction rule, (b) random draws (mean of |
| 30), (c) top trailing-12-month momentum, size each add at 8% pro-rata, hold 12 months, and |
| record realized volatility, max drawdown, return and Sharpe. Walk-forward: only information |
| available at the rebalance date is used. Output: data/policy_backtest.json + a markdown table. |
| |
| Run: .venv/bin/python scripts/backtest_fit_policy.py |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import sys |
| from pathlib import Path |
|
|
| import numpy as np |
| import pandas as pd |
|
|
| ROOT = Path(__file__).resolve().parent.parent |
| sys.path.insert(0, str(ROOT)) |
| from src import construct as CX |
| from src.market_data import fetch_history, load_instruments |
|
|
| SEEDS = { |
| "tech_concentrated": {"NVDA": 0.35, "AMD": 0.2, "PLTR": 0.15, "TSLA": 0.15, "MSFT": 0.15}, |
| "income_rates": {"O": 0.25, "DUK": 0.2, "VICI": 0.2, "ENB": 0.2, "SPG": 0.15}, |
| "quality_midcap": {"CROX": 0.25, "DECK": 0.25, "ETSY": 0.2, "ROKU": 0.15, "CELH": 0.15}, |
| } |
| UNIVERSE = ["ED", "NOC", "BSX", "CRH", "MO", "VLO", "EXPD", "JLL", "ES", "AMP", "CMCSA", "CL", |
| "UBER", "TSM", "AMZN", "VEEV", "LIN", "JNJ", "PG", "KO", "XOM", "CAT", "UNH", "WMT", |
| "JPM", "BRK-B", "LLY", "ABBV", "PEP", "HD", "GOOGL", "META", "COST", "AVGO", "V"] |
| UNIVERSE += [i["ticker"] for i in load_instruments()] |
| W_ADD, HOLD_DAYS, N_RANDOM = 0.08, 252, 30 |
|
|
|
|
| def _stats(ret: pd.Series) -> dict: |
| if len(ret) < 100: |
| return {} |
| curve = (1 + ret).cumprod() |
| dd = float((curve / curve.cummax() - 1).min()) |
| vol = float(ret.std() * np.sqrt(252)) |
| tot = float(curve.iloc[-1] - 1) |
| return {"return": tot, "vol": vol, "max_dd": dd, "sharpe": float(ret.mean() / ret.std() * np.sqrt(252)) if ret.std() > 0 else 0.0} |
|
|
|
|
| def _hold(px: pd.DataFrame, w: dict, start, end) -> pd.Series | None: |
| sub = px.loc[(px.index > start) & (px.index <= end), list(w)] |
| if len(sub) < 100: |
| return None |
| rets = sub.pct_change(fill_method=None).fillna(0.0) |
| wv = np.array([w[t] for t in sub.columns]); wv = wv / wv.sum() |
| return pd.Series(rets.to_numpy() @ wv, index=sub.index) |
|
|
|
|
| def main() -> None: |
| tickers = sorted(set(sum((list(s) for s in SEEDS.values()), [])) | set(UNIVERSE)) |
| px = fetch_history(tickers, period="8y") |
| px.index = pd.to_datetime(px.index) |
| px = px.dropna(axis=1, thresh=int(len(px) * 0.6)) |
| dates = pd.date_range("2021-03-31", "2025-06-30", freq="QE") |
| rng = np.random.default_rng(0) |
| results: dict = {} |
| for seed_name, seed in SEEDS.items(): |
| seed = {t: w for t, w in seed.items() if t in px.columns} |
| rows = {"greedy": [], "random": [], "momentum": [], "seed_only": []} |
| for d in dates: |
| hist = px.loc[px.index <= d] |
| end = d + pd.Timedelta(days=int(HOLD_DAYS * 365 / 252)) |
| if end > px.index[-1]: |
| break |
| univ = [t for t in UNIVERSE if t in hist.columns and t not in seed |
| and int(hist[t].iloc[-500:].notna().sum()) >= 400] |
| if len(univ) < 6: |
| continue |
| rets = CX.returns_matrix(hist, None, list(seed) + univ) |
| cov, tick, _ = CX.shrunk_cov(rets) |
| cands = [{"ticker": t, "fit": {"score": 0.0}} for t in univ if t in tick] |
| picks = [p["ticker"] for p in CX.greedy_select(seed, cands, cov, tick, max_adds=3, w_add=W_ADD)] |
| mom = hist[univ].iloc[-1] / hist[univ].iloc[-252] - 1 |
| mom_picks = list(mom.dropna().sort_values(ascending=False).index[:3]) |
|
|
| def book(adds): |
| w = {t: v * (1 - W_ADD * len(adds)) for t, v in seed.items()} |
| for a in adds: |
| w[a] = W_ADD |
| return w |
| for name, adds in (("greedy", picks), ("momentum", mom_picks), ("seed_only", [])): |
| r = _hold(px, book(adds), d, end) |
| if r is not None: |
| rows[name].append(_stats(r)) |
| rand = [] |
| for _ in range(N_RANDOM): |
| adds = list(rng.choice(univ, size=3, replace=False)) |
| r = _hold(px, book(adds), d, end) |
| if r is not None: |
| rand.append(_stats(r)) |
| if rand: |
| rows["random"].append({k: float(np.mean([x[k] for x in rand])) for k in rand[0]}) |
| results[seed_name] = {name: {k: round(float(np.mean([x[k] for x in v])), 4) for k in v[0]} | {"n_periods": len(v)} |
| for name, v in rows.items() if v} |
| out = ROOT / "data" / "policy_backtest.json" |
| out.write_text(json.dumps(results, indent=1)) |
| print("| seed | policy | periods | avg 12m return | avg vol | avg max DD | avg Sharpe |") |
| print("|---|---|---|---|---|---|---|") |
| for s, pols in results.items(): |
| for p, m in pols.items(): |
| print(f"| {s} | {p} | {m['n_periods']} | {m['return'] * 100:+.1f}% | {m['vol'] * 100:.1f}% | {m['max_dd'] * 100:.1f}% | {m['sharpe']:.2f} |") |
| print("saved", out) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
