#!/usr/bin/env python3 """ Rank **50** QuantifiedStrategies-style systematic ETF sleeves. Scores each sleeve on 2016+ (Best Ideas window) and 1993+ (full SPY history). Composite **book_fit** = Sharpe(2016) × (1 − |ρ_vs_SPY|) — favors diversifiers. Example:: cd /Users/robzingale/trading_bot && PYTHONUNBUFFERED=1 \\ .venv/bin/python RenTech/strategy_stack/run_qs_systematic_rank50.py \\ --start 2016-01-04 --end 2026-04-02 """ from __future__ import annotations import argparse import json import sys from pathlib import Path import numpy as np import pandas as pd _REPO = Path(__file__).resolve().parents[2] if str(_REPO) not in sys.path: sys.path.insert(0, str(_REPO)) from RenTech.strategy_stack.qs_systematic_library import SYSTEMATIC_STRATEGIES from RenTech.strategy_stack.run_qs_top_ideas_backtest import ( _align, _fetch_ohlc, _metrics, _spy_close_to_close, ) LOGS = _REPO / "RenTech" / "data" / "logs" DEFAULT_OUT = LOGS / "qs_systematic_rank50" def _run_one(fn, panels: dict[str, pd.DataFrame], idx: pd.DatetimeIndex) -> pd.Series: import inspect sig = inspect.signature(fn) kwargs: dict = {} if "tlt" in sig.parameters: kwargs["tlt"] = panels["tlt"] if "gld" in sig.parameters: kwargs["gld"] = panels["gld"] if "vix" in sig.parameters: kwargs["vix"] = panels["vix"] if "xlp" in sig.parameters: kwargs["xlp"] = panels["xlp"] if "xlu" in sig.parameters: kwargs["xlu"] = panels["xlu"] if "qqq" in sig.parameters: kwargs["qqq"] = panels["qqq"] r = fn(panels["spy"], **kwargs) return r.reindex(idx).fillna(0.0) def main() -> None: ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) ap.add_argument("--start", default="2016-01-04") ap.add_argument("--end", default="2026-04-02") ap.add_argument("--long-start", default="1993-01-29") ap.add_argument("--capital", type=float, default=100_000.0) ap.add_argument("--yahoo-period", default="max") ap.add_argument("--out-prefix", type=Path, default=DEFAULT_OUT) args = ap.parse_args() start = pd.Timestamp(args.start) end = pd.Timestamp(args.end) if str(args.end).strip() else None long_start = pd.Timestamp(args.long_start) cap = float(args.capital) raw = { "spy": _fetch_ohlc("SPY", args.yahoo_period), "tlt": _fetch_ohlc("TLT", args.yahoo_period), "gld": _fetch_ohlc("GLD", args.yahoo_period), "vix": _fetch_ohlc("^VIX", args.yahoo_period), "xlp": _fetch_ohlc("XLP", args.yahoo_period), "xlu": _fetch_ohlc("XLU", args.yahoo_period), "qqq": _fetch_ohlc("QQQ", args.yahoo_period), } spy_win = _align(raw["spy"], start, end) spy_long = _align(raw["spy"], long_start, end) panels_win = {k: _align(v, start, end) for k, v in raw.items()} panels_long = {k: _align(v, long_start, end) for k, v in raw.items()} spy_bh = _spy_close_to_close(spy_win) rows: list[dict] = [] daily_win: dict[str, pd.Series] = {} for sid, fn in SYSTEMATIC_STRATEGIES.items(): if fn is None: continue r_win = _run_one(fn, panels_win, spy_win.index) r_long = _run_one(fn, panels_long, spy_long.index) m_win = _metrics(r_win, cap) m_long = _metrics(r_long, cap) rho = float(pd.DataFrame({"s": r_win, "spy": spy_bh}).corr().iloc[0, 1]) book_fit = m_win["sharpe"] * (1.0 - abs(rho)) rows.append( { "rank_id": sid, "sharpe_2016": m_win["sharpe"], "sharpe_1993": m_long["sharpe"], "return_pct_2016": m_win["total_return_pct"], "cagr_2016": m_win["cagr_pct"], "max_dd_2016": m_win["max_dd_pct"], "pct_invested": m_win["pct_days_invested"], "rho_spy": round(rho, 3), "book_fit": round(book_fit, 3), "return_pct_1993": m_long["total_return_pct"], "max_dd_1993": m_long["max_dd_pct"], } ) daily_win[sid] = r_win df = pd.DataFrame(rows) df_by_sharpe = df.sort_values("sharpe_2016", ascending=False).reset_index(drop=True) df_by_fit = df.sort_values("book_fit", ascending=False).reset_index(drop=True) df_by_sharpe.insert(0, "rank_sharpe", range(1, len(df_by_sharpe) + 1)) df_by_fit.insert(0, "rank_book_fit", range(1, len(df_by_fit) + 1)) merged = df_by_sharpe.merge( df_by_fit[["rank_id", "rank_book_fit"]], on="rank_id", how="left", ) prefix = args.out_prefix.expanduser().resolve() prefix.parent.mkdir(parents=True, exist_ok=True) rank_path = Path(f"{prefix}_ranking.csv") meta_path = Path(f"{prefix}_meta.json") metrics_path = Path(f"{prefix}_metrics.txt") merged.to_csv(rank_path, index=False) print(f"\n=== Top 15 by Sharpe ({start.date()} → {end.date() if end else 'latest'}) ===") cols = ["rank_sharpe", "rank_id", "sharpe_2016", "return_pct_2016", "max_dd_2016", "rho_spy", "pct_invested"] print(merged[cols].head(15).to_string(index=False)) print(f"\n=== Top 15 by book_fit (Sharpe × low SPY correlation) ===") fit_sorted = merged.sort_values("book_fit", ascending=False) print(fit_sorted[["rank_book_fit", "rank_id", "book_fit", "sharpe_2016", "rho_spy", "max_dd_2016"]].head(15).to_string(index=False)) print(f"\n=== Bottom 5 by Sharpe ===") print(merged[cols].tail(5).to_string(index=False)) meta = { "command": " ".join(sys.argv), "start": str(start.date()), "end": str(end.date()) if end else "", "long_start": str(long_start.date()), "n_strategies": len(merged), "top10_sharpe": merged.head(10)["rank_id"].tolist(), "top10_book_fit": fit_sorted.head(10)["rank_id"].tolist(), "rows": merged.to_dict(orient="records"), } meta_path.write_text(json.dumps(meta, indent=2)) lines = [ f"QS systematic rank-50 {start.date()} → {end.date() if end else 'latest'}", f"Long history from {long_start.date()}", "", "Top 15 by Sharpe (2016+):", merged[cols].head(15).to_string(index=False), "", "Top 15 by book_fit:", fit_sorted[["rank_book_fit", "rank_id", "book_fit", "sharpe_2016", "rho_spy", "max_dd_2016"]] .head(15) .to_string(index=False), ] metrics_path.write_text("\n".join(lines) + "\n") print(f"\nWrote {rank_path}") print(f"Wrote {meta_path}") if __name__ == "__main__": main()