#!/usr/bin/env python3 """ Ride-the-rockets **50/50 blend** sleeve for Best Ideas. Combines: * near_52w_high top25 (extension riders) * ten_rockets top10 (concentrated relative winners) Equal capital each day → ``ride_rockets_5050_standard_daily.csv``. Example:: cd /Users/robzingale/trading_bot && PYTHONUNBUFFERED=1 .venv/bin/python \\ RenTech/strategy_stack/run_ride_rockets_5050_standard.py \\ --start 2016-01-04 --end 2026-04-02 --reuse-equity-cache """ from __future__ import annotations import argparse import json import sys from dataclasses import asdict, fields from pathlib import Path 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.data_loader import DataLoader from RenTech.strategy_stack.main import _compute_daily_backtest_features from RenTech.strategy_stack.run_sp500_dip_standard import _filter_equity_by_history from RenTech.strategy_stack.sp500_momentum_backtest import ( load_equity_panel, run_backtest, summarize_returns, yearly_stats, ) from RenTech.strategy_stack.sp500_momentum_index import ( Sp500MomentumConfig, ride_rockets_defaults, ) LOGS = _REPO / "RenTech" / "data" / "logs" DEFAULT_OUT = LOGS / "ride_rockets_5050_standard" def _cfg_from_patch(patch: dict) -> Sp500MomentumConfig: allowed = {f.name for f in fields(Sp500MomentumConfig)} base = asdict(Sp500MomentumConfig()) cleaned = {k: v for k, v in patch.items() if k in allowed} return Sp500MomentumConfig(**{**base, **cleaned}) def near25_patch() -> dict: return { **ride_rockets_defaults(top_n=25), "near_high_min_frac": 0.95, "selection_filters": ["ride_rockets", "near_high", "top_25"], } def ten10_patch() -> dict: return { **ride_rockets_defaults(top_n=10), "max_single_weight": 0.20, "score_weight_power": 1.5, "selection_filters": ["ride_rockets", "ten_rockets"], } 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("--yahoo-period", default="max") ap.add_argument("--capital", type=float, default=100_000.0) ap.add_argument("--reuse-equity-cache", action="store_true") ap.add_argument("--out-prefix", type=Path, default=DEFAULT_OUT) args = ap.parse_args() capital = float(args.capital) min_first = pd.Timestamp(args.start).normalize() - pd.Timedelta(days=400) equity = load_equity_panel( yahoo_period=args.yahoo_period, start=args.start, end=args.end, max_tickers=0, refresh_cache=not args.reuse_equity_cache, pit_universe=True, universe="sp500", ) equity = _filter_equity_by_history(equity, min_first) print(f"Universe: {len(equity)} names\n", flush=True) sleeves: dict[str, pd.Series] = {} sleeve_meta: dict[str, dict] = {} for name, patch in (("near25", near25_patch()), ("ten10", ten10_patch())): print(f"=== {name} ===", flush=True) r, _rebal, meta = run_backtest( equity_dict=equity, cfg=_cfg_from_patch(patch), start=args.start, end=args.end, yahoo_period=args.yahoo_period, capital=capital, refresh_shares=False, pit_universe=True, ) sleeves[name] = r sleeve_meta[name] = { k: meta[k] for k in ( "total_return_pct", "cagr_pct", "sharpe_daily", "max_drawdown_pct", "beta_vs_spy", ) } print( f" → {meta['total_return_pct']:+.1f}% Sharpe {meta['sharpe_daily']:.2f} " f"DD {meta['max_drawdown_pct']:.1f}%", flush=True, ) aligned = pd.concat( [sleeves["near25"].rename("near25"), sleeves["ten10"].rename("ten10")], axis=1, ).dropna() r_combo = (0.5 * aligned["near25"] + 0.5 * aligned["ten10"]).rename("daily_ret") spy_df = _compute_daily_backtest_features(DataLoader().fetch_daily("SPY", period=args.yahoo_period)) meta = summarize_returns(r_combo, capital=capital, spy_df=spy_df) rho = float(aligned["near25"].corr(aligned["ten10"])) prefix = args.out_prefix.expanduser().resolve() prefix.parent.mkdir(parents=True, exist_ok=True) daily_path = Path(f"{prefix}_daily.csv") yearly_path = Path(f"{prefix}_yearly.csv") meta_path = Path(f"{prefix}_meta.json") metrics_path = Path(f"{prefix}_metrics.txt") eq = capital * (1.0 + r_combo).cumprod() pnl = r_combo * capital pd.DataFrame( { "date": r_combo.index.strftime("%Y-%m-%d"), "daily_ret": r_combo.values, "daily_pnl_usd": pnl.values, "equity_usd": eq.values, "ret_near25": aligned["near25"].reindex(r_combo.index).values, "ret_ten10": aligned["ten10"].reindex(r_combo.index).values, # Long equity Reg-T style proxy (~50% of equity notional for a blended book). "margin_usd": (0.5 * capital), } ).to_csv(daily_path, index=False) yearly_stats(r_combo).to_csv(yearly_path, index=False) cmd = ( f"cd {_REPO} && PYTHONUNBUFFERED=1 .venv/bin/python " f"RenTech/strategy_stack/run_ride_rockets_5050_standard.py " f"--start {args.start} --end {args.end} --capital {capital:g}" ) payload = { **meta, "token": "ride_rockets", "strategy": "RideRockets5050", "sleeve_weights": {"near25": 0.5, "ten10": 0.5}, "sleeve_meta": sleeve_meta, "corr_sleeves_daily": round(rho, 4), "capital_usd": capital, "command": cmd, "daily_csv": str(daily_path), "yearly_csv": str(yearly_path), } meta_path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") metrics_path.write_text( "\n".join( [ "Ride-rockets 50/50 (near_52w_high top25 + ten_rockets top10)", f"Window: {args.start} → {args.end} capital=${capital:,.0f}", f"near25: {sleeve_meta['near25']}", f"ten10: {sleeve_meta['ten10']}", f"corr(daily): {rho:.3f}", f"COMBINE: return {meta['total_return_pct']:+.1f}% CAGR {meta['cagr_pct']:+.1f}% " f"Sharpe {meta['sharpe_daily']:.2f} maxDD {meta['max_drawdown_pct']:.1f}%", f"Command: {cmd}", f"Wrote {daily_path}", ] ) + "\n", encoding="utf-8", ) print( f"\n50/50: return {meta['total_return_pct']:+.1f}% CAGR {meta['cagr_pct']:+.1f}% " f"Sharpe {meta['sharpe_daily']:.2f} maxDD {meta['max_drawdown_pct']:.1f}% " f"corr {rho:.3f}", flush=True, ) print(f"Wrote {daily_path}", flush=True) if __name__ == "__main__": main()