#!/usr/bin/env python3 """ Diagnostics for the hourly swing-breakout backtest. Run from the **repository root** (parent of ``RenTech/``):: python3 RenTech/strategy_stack/diagnose_hourly_swing_breakout.py python3 RenTech/strategy_stack/diagnose_hourly_swing_breakout.py --no-grid --csv-out /tmp/swing_diag.csv Outputs: * Per-ticker: trade count, win rate, mean trade PnL, time-series cum return / rough Sharpe * Blended portfolio (``--blend equal|inv_vol|winners_equal``; matches ``main.py`` flags) * Small ``hold_bars`` × ``min_atr_mult`` grid on the portfolio (optional; uses same blend) Interpretation notes are printed at the end for quick human + agent review. """ from __future__ import annotations import argparse import os import sys import numpy as np import pandas as pd _REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) if _REPO_ROOT not in sys.path: sys.path.insert(0, _REPO_ROOT) from RenTech.strategy_stack.main import ( # noqa: E402 HOURLY_SWING_MACRO_ETF_TICKERS, _load_hourly_swing_universe, ) from RenTech.strategy_stack.swing_breakout_hourly import ( # noqa: E402 SwingBreakoutHourlyConfig, generate_portfolio_returns, simulate_ticker_hourly, ) # Rough annualization for US equity/ETF hourly (~6.5 RTH hours × 252). BARS_PER_YEAR = 252.0 * 6.5 def _series_stats(r: pd.Series) -> dict[str, float]: r = r.fillna(0.0).astype(np.float64) n = int(len(r)) if n < 2: return dict(n_bars=n, mean_bar=float("nan"), std=float("nan"), sharpe=float("nan"), cum=float("nan")) mu = float(r.mean()) sd = float(r.std(ddof=1)) sharpe = mu / sd * np.sqrt(BARS_PER_YEAR) if sd > 1e-12 else float("nan") cum = float((1.0 + r).prod() - 1.0) return dict(n_bars=n, mean_bar=mu, std=sd, sharpe_approx=sharpe, cum_ret=cum) def main() -> None: p = argparse.ArgumentParser(description=__doc__) p.add_argument("--intraday-period", default="730d", help="yfinance period for hourly bars") p.add_argument("--interval", default="1h", choices=("1h", "60m", "15m", "30m")) p.add_argument( "--universe", default="macro-etfs", choices=("macro-etfs", "sp100", "sp500"), help="Universe for loader (macro-etfs = HOURLY_SWING_MACRO_ETF_TICKERS list).", ) p.add_argument("--tickers", default="", help="Comma list overrides universe when non-empty") p.add_argument("--max-tickers", type=int, default=0, help="0 = all symbols in universe list") p.add_argument("--hold-bars", type=int, default=10) p.add_argument("--min-atr-mult", type=float, default=1.0) p.add_argument("--fractal-left", type=int, default=2) p.add_argument("--fractal-right", type=int, default=2) p.add_argument("--daily-atr", type=int, default=14) p.add_argument( "--close-confirm", action="store_true", help="Require bar close past the swing level (reduces wick-only breaks).", ) p.add_argument("--no-grid", action="store_true", help="Skip hold × mult sensitivity table") p.add_argument("--csv-out", default="", help="Write per-ticker summary CSV to this path") p.add_argument( "--blend", choices=("equal", "inv_vol", "winners_equal"), default="equal", help="Portfolio blend: equal mean | inverse asset vol | IS winners only (see --winner-epsilon).", ) p.add_argument( "--vol-window", type=int, default=20, help="inv_vol: trailing bars for std of underlying hourly returns (causal, shift 1).", ) p.add_argument( "--winner-epsilon", type=float, default=0.0, help="winners_equal: keep tickers with full-sample strat cum ret > this (in-sample selection).", ) args = p.parse_args() os.chdir(_REPO_ROOT) panels = _load_hourly_swing_universe( args.intraday_period, args.interval, universe=str(args.universe), max_tickers=int(args.max_tickers), tickers_csv=str(args.tickers), ) if len(panels) < 1: raise SystemExit("No tickers loaded; check network, period, or --tickers") base_cfg = SwingBreakoutHourlyConfig( fractal_left=int(args.fractal_left), fractal_right=int(args.fractal_right), daily_atr_period=int(args.daily_atr), min_atr_mult=float(args.min_atr_mult), hold_bars=int(args.hold_bars), close_confirms_breakout=bool(args.close_confirm), ) print("=" * 72) print("HOURLY SWING BREAKOUT — PER-TICKER (baseline config)") print("=" * 72) u_label = "custom --tickers" if str(args.tickers).strip() else str(args.universe) print( f"Universe: {u_label} | loaded={len(panels)} | " f"hold={base_cfg.hold_bars} | min_atr_mult={base_cfg.min_atr_mult} | " f"fractal L/R={base_cfg.fractal_left}/{base_cfg.fractal_right} | " f"close_confirm={base_cfg.close_confirms_breakout} | " f"blend={args.blend}" ) if args.universe == "macro-etfs" and not str(args.tickers).strip(): print(f"Macro list ({len(HOURLY_SWING_MACRO_ETF_TICKERS)}): {', '.join(HOURLY_SWING_MACRO_ETF_TICKERS)}") rows: list[dict[str, object]] = [] for tkr in sorted(panels.keys()): df = panels[tkr] r, n_tr, pnls = simulate_ticker_hourly(df, base_cfg, record_trade_pnls=True) st = _series_stats(r) pn = np.asarray(pnls, dtype=np.float64) win = float(np.mean(pn > 0.0)) if pn.size else float("nan") m_tr = float(np.mean(pn)) if pn.size else float("nan") med_tr = float(np.median(pn)) if pn.size else float("nan") rows.append( { "ticker": tkr, "bars": int(len(df)), "trades": int(n_tr), "n_pnls": int(len(pnls)), "win_rate": win, "mean_trade_pnl": m_tr, "median_trade_pnl": med_tr, "cum_ret": st["cum_ret"], "sharpe_approx": st["sharpe_approx"], } ) print( f" {tkr:5s} trades={n_tr:4d} win%={win * 100:5.1f} " f"mean_trade={m_tr * 100:6.3f}% med_trade={med_tr * 100:6.3f}% " f"cum={st['cum_ret'] * 100:7.2f}% sharpe*={st['sharpe_approx']:5.2f}" ) port, counts = generate_portfolio_returns( panels, base_cfg, blend=str(args.blend), vol_window=int(args.vol_window), winner_epsilon=float(args.winner_epsilon), verbose=True, ) pst = _series_stats(port) print("-" * 72) blab = { "equal": "equal-weight mean", "inv_vol": f"inverse-vol (window={int(args.vol_window)})", "winners_equal": f"winners_equal (eps={float(args.winner_epsilon):g})", }[str(args.blend)] print( f"PORTFOLIO ({blab}) " f"cum={pst['cum_ret'] * 100:.2f}% sharpe*={pst['sharpe_approx']:.2f} " f"bars={pst['n_bars']}" ) print(f" trades by ticker: {dict(sorted(counts.items()))}") if args.csv_out: pd.DataFrame(rows).to_csv(args.csv_out, index=False) print(f"Wrote {args.csv_out}") if not args.no_grid: print("\n" + "=" * 72) print(f"GRID: portfolio cum ret % and sharpe* (blend={args.blend})") print("=" * 72) holds = [1, 3, 5, 10, 15] mults = [0.5, 0.75, 1.0, 1.5, 2.0] hdr = "mult ->" + "".join(f"{m:>8.2f}" for m in mults) print(f"{'hold':>6} {hdr}") for h in holds: cells = [] for m in mults: cfg = SwingBreakoutHourlyConfig( fractal_left=base_cfg.fractal_left, fractal_right=base_cfg.fractal_right, daily_atr_period=base_cfg.daily_atr_period, min_atr_mult=m, hold_bars=h, close_confirms_breakout=base_cfg.close_confirms_breakout, ) pr, _ = generate_portfolio_returns( panels, cfg, blend=str(args.blend), vol_window=int(args.vol_window), winner_epsilon=float(args.winner_epsilon), verbose=False, ) gs = _series_stats(pr) cells.append(f"{gs['cum_ret'] * 100:5.1f}/{gs['sharpe_approx']:4.2f}") print(f"{h:>6} " + " ".join(f"{c:>12}" for c in cells)) print("\n" + "=" * 72) print("INTERPRETATION (for tuning strategy / code)") print("=" * 72) print( """ * win_rate / mean_trade_pnl: If mean_trade is negative but Sharpe looks OK, few large winners may dominate — check median_trade_pnl and cum_ret per ticker. * Persistent losers: Consider excluding those tickers, long-only, or a trend filter on that sleeve before changing global fractal parameters. * trades vs n_pnls: n_pnls can be trades + 1 if a position is still open at sample end. * Grid: Best cell on IS sample often overfits; prefer stable regions across adjacent (hold, mult) cells and confirm on a held-out time window. * sharpe* uses ~252×6.5 bars/year; for mixed sessions it is approximate only. * Costs/slippage are NOT deducted — stop-style entries are usually pessimistic live. * winners_equal uses full-sample per-ticker cum to pick names — optimistic IS bias; use only for exploration or after a train/hold split. """ ) if __name__ == "__main__": main()