#!/usr/bin/env python3 """ Day-trading complement ideas 01–100 — registry, templates, metrics. See ``DAY_TRADING_100_COMPLEMENT_IDEAS.md``. Skip only unavailable-data ideas. """ from __future__ import annotations import time from dataclasses import dataclass, field, replace from typing import Any, Callable, Literal import numpy as np import pandas as pd from RenTech.strategy_stack.alpaca_minute_loader import compound_intraday_to_daily from RenTech.strategy_stack.cm_intraday_atr_breakout import ( atr_breakout_early_config, build_breakout_panels, run_backtest as run_atr_backtest, simulate_atr_breakout, ) from RenTech.strategy_stack.cm_intraday_dip import ( CmIntradayConfig, TradeRow, _exit_atr, _portfolio_open_count, _session_bar_index, _session_key, _session_vwap, _short_bar_return, _slip, _wilder_atr, build_feature_panels, build_regime_context, pdl_touch_short_config, pdl_touch_short_highvol_config, portfolio_metrics, run_backtest as run_cm_backtest, trade_stats, ) from RenTech.strategy_stack.ma_slope_engine import compute_ma, compute_ma_slope, compute_slope_rank_score from RenTech.strategy_stack.ma_slope_intraday_enhanced import ( EnhancedIntradayConfig, EnhancedIntradayEngine, baseline_enhanced_config, metrics_daily, ) from RenTech.strategy_stack.multi_strategy_manager import _align_panel_frames OR_BARS = 6 SLIP = 3.0 MEGA5 = ["AAPL", "MSFT", "NVDA", "AMZN", "GOOGL"] ETF_CANDIDATES = ["SPY", "QQQ", "XLK", "IWM", "TLT", "HYG", "RSP", "VIXY", "SVXY", "XLB", "XLE", "XLF", "XLI", "XLP", "XLU", "XLV", "XLY"] @dataclass class IdeaResult: idea_id: str name: str status: str universe: str = "" total_return_pct: float = float("nan") sharpe: float = float("nan") max_dd_pct: float = float("nan") n_trades: int = 0 win_rate_pct: float = float("nan") avg_pnl_pct: float = float("nan") yearly: dict[str, float] = field(default_factory=dict) elapsed_sec: float = 0.0 skip_reason: str = "" def to_row(self) -> dict[str, Any]: row = { "idea_id": self.idea_id, "name": self.name, "status": self.status, "universe": self.universe, "total_return_pct": self.total_return_pct, "sharpe": self.sharpe, "max_dd_pct": self.max_dd_pct, "n_trades": self.n_trades, "win_rate_pct": self.win_rate_pct, "avg_pnl_pct": self.avg_pnl_pct, "elapsed_sec": round(self.elapsed_sec, 2), "skip_reason": self.skip_reason, } for y, r in sorted(self.yearly.items()): row[f"ret_{y}"] = r return row @dataclass class IdeaSpec: idea_id: str name: str runner: str # key into RUNNERS universe: str = "mega5" params: dict[str, Any] = field(default_factory=dict) skip_reason: str | None = None slow: bool = False def _yearly_from_port(port: pd.Series) -> dict[str, float]: ds = compound_intraday_to_daily(port).dropna() out: dict[str, float] = {} if ds.empty: return out for y, g in ds.groupby(ds.index.year): eq = (1.0 + g).cumprod() out[str(int(y))] = float((eq.iloc[-1] - 1.0) * 100.0) return out def summarize( port: pd.Series | None, trades: pd.DataFrame | None, *, idea_id: str, name: str, universe: str, elapsed: float, status: str = "ok", skip_reason: str = "", end: pd.Timestamp | None = None, ) -> IdeaResult: if status != "ok" or port is None: return IdeaResult(idea_id=idea_id, name=name, status=status, universe=universe, elapsed_sec=elapsed, skip_reason=skip_reason) if end is not None: port = port.loc[:end] pm = portfolio_metrics(port) if len(port) else {} ts = trade_stats(trades) if trades is not None and not trades.empty else {} return IdeaResult( idea_id=idea_id, name=name, status=status, universe=universe, total_return_pct=float(pm.get("total_return_pct", float("nan"))), sharpe=float(pm.get("sharpe", float("nan"))), max_dd_pct=float(pm.get("max_dd_pct", float("nan"))), n_trades=int(ts.get("n_trades", 0) or 0), win_rate_pct=float(ts.get("win_rate_pct", float("nan"))), avg_pnl_pct=float(ts.get("avg_pnl_pct", float("nan"))), yearly=_yearly_from_port(port), elapsed_sec=elapsed, skip_reason=skip_reason, ) # --------------------------------------------------------------------------- # Level-touch simulator (generic PDL/PDH/OR/… fade) # --------------------------------------------------------------------------- def simulate_level_touch( panels: dict[str, pd.DataFrame], *, level_key: str, side: Literal["long", "short"] = "short", max_concurrent: int = 10, slippage_bps: float = SLIP, profit_atr_mult: float = 1.0, stop_atr_mult: float = 0.75, min_bars_held: int = 2, above_vwap: bool = False, below_vwap: bool = False, min_atr_pct: float = 0.0, distant_atr_mult: float = 0.0, first_touch_only: bool = True, third_touch: bool = False, confluence_week: bool = False, after_false_break: bool = False, min_session_bar: int = 0, max_session_bar: int = 0, use_support_intact: bool = False, ) -> tuple[list[TradeRow], np.ndarray]: close = panels["close"].to_numpy(dtype=np.float64) high = panels["high"].to_numpy(dtype=np.float64) low = panels["low"].to_numpy(dtype=np.float64) atr = panels["atr"].to_numpy(dtype=np.float64) daily_atr = panels["daily_atr"].to_numpy(dtype=np.float64) lvl = panels[level_key].to_numpy(dtype=np.float64) vwap = panels["vwap"].to_numpy(dtype=np.float64) if "vwap" in panels else np.full_like(close, np.nan) support = panels["support_intact"].to_numpy(dtype=np.float64) if "support_intact" in panels else np.ones_like(close) week_l = panels["week_low"].to_numpy(dtype=np.float64) if "week_low" in panels else np.full_like(close, np.nan) week_h = panels["week_high"].to_numpy(dtype=np.float64) if "week_high" in panels else np.full_like(close, np.nan) ses_open = panels["ses_open"].to_numpy(dtype=np.float64) if "ses_open" in panels else close idx = panels["close"].index cols = list(panels["columns"]) ses = _session_key(idx).to_numpy() bar = _session_bar_index(idx).to_numpy(dtype=np.int64) t_rows, n_sym = close.shape port_r = np.zeros(t_rows, dtype=np.float64) trades: list[TradeRow] = [] state = np.zeros(n_sym, dtype=np.int8) entry_px = np.full(n_sym, np.nan) entry_exit_atr = np.full(n_sym, np.nan) entry_bar = np.full(n_sym, -1, dtype=np.int32) entry_time = [None] * n_sym traded_session = np.full(n_sym, None, dtype=object) touch_count = np.zeros(n_sym, dtype=np.int32) session_high = np.full(n_sym, np.nan) session_low = np.full(n_sym, np.nan) broke_above = np.zeros(n_sym, dtype=bool) # for false-break PDH broke_below = np.zeros(n_sym, dtype=bool) cfg_exit = CmIntradayConfig(min_bars_held_before_exit=min_bars_held, use_daily_atr_exits=True) prev_ses = None for d in range(t_rows): if prev_ses is None or ses[d] != prev_ses: state[:] = 0 traded_session[:] = None entry_px[:] = np.nan entry_exit_atr[:] = np.nan entry_bar[:] = -1 touch_count[:] = 0 session_high[:] = np.nan session_low[:] = np.nan broke_above[:] = False broke_below[:] = False prev_ses = ses[d] for i in range(n_sym): if np.isfinite(high[d, i]): session_high[i] = high[d, i] if not np.isfinite(session_high[i]) else max(session_high[i], high[d, i]) if np.isfinite(low[d, i]): session_low[i] = low[d, i] if not np.isfinite(session_low[i]) else min(session_low[i], low[d, i]) if np.isfinite(lvl[d, i]): if high[d, i] > lvl[d, i] + 1e-9: broke_above[i] = True if low[d, i] < lvl[d, i] - 1e-9: broke_below[i] = True # exits for i in range(n_sym): if state[i] != 1: continue c_d = close[d, i] if not np.isfinite(c_d): continue ep, ea, eb = entry_px[i], entry_exit_atr[i], int(entry_bar[i]) reason = None held = d - eb if held >= min_bars_held and np.isfinite(ea): if side == "short": if c_d <= ep - profit_atr_mult * ea: reason = "profit_atr" elif c_d >= ep + stop_atr_mult * ea: reason = "stop_atr" else: if c_d >= ep + profit_atr_mult * ea: reason = "profit_atr" elif c_d <= ep - stop_atr_mult * ea: reason = "stop_atr" if d + 1 >= t_rows or ses[d + 1] != ses[d]: reason = reason or "session_close" if reason: if side == "short": xp = _slip(c_d, slippage_bps, "buy") pnl = float((ep - xp) / ep) if ep > 0 else 0.0 else: xp = _slip(c_d, slippage_bps, "sell") pnl = float((xp / ep) - 1.0) if ep > 0 else 0.0 trades.append( TradeRow( variant=f"level_{level_key}_{side}", symbol=cols[i], session_date=str(pd.Timestamp(ses[d]).date()), entry_time=str(entry_time[i]), exit_time=str(idx[d]), entry_price=float(ep), exit_price=float(xp), pnl_pct=pnl, bars_held=int(held), exit_reason=reason, ) ) state[i] = 0 # entries if max_session_bar > 0 and bar[d] > max_session_bar: pass elif bar[d] < min_session_bar: pass else: cand: list[tuple[int, float]] = [] for i in range(n_sym): if state[i] == 1 or traded_session[i] is not None: continue L = lvl[d, i] a_d, c_d = atr[d, i], close[d, i] if not (np.isfinite(L) and np.isfinite(a_d) and a_d > 0 and np.isfinite(c_d) and c_d > 0): continue if use_support_intact and support[d, i] < 0.5: continue if above_vwap and not (np.isfinite(vwap[d, i]) and c_d > vwap[d, i]): continue if below_vwap and not (np.isfinite(vwap[d, i]) and c_d < vwap[d, i]): continue if min_atr_pct > 0 and (a_d / c_d * 100.0) < min_atr_pct: continue if distant_atr_mult > 0: # open must be well above (short at pdl) or below (long at pdh) so = ses_open[d, i] if side == "short" and not (np.isfinite(so) and so >= L + distant_atr_mult * a_d): continue if side == "long" and not (np.isfinite(so) and so <= L - distant_atr_mult * a_d): continue if confluence_week: if side == "short" and level_key == "pdl": if not (np.isfinite(week_l[d, i]) and abs(L - week_l[d, i]) <= 0.25 * a_d): continue if side == "short" and level_key == "pdh": if not (np.isfinite(week_h[d, i]) and abs(L - week_h[d, i]) <= 0.25 * a_d): continue if after_false_break: # need prior pierce then reclaim if side == "short": # PDH false break: broke above then close back if not broke_above[i]: continue if not (high[d, i] >= L and c_d < L): continue else: if not broke_below[i]: continue if not (low[d, i] <= L and c_d > L): continue else: touched = (low[d, i] <= L <= high[d, i]) if np.isfinite(low[d, i]) and np.isfinite(high[d, i]) else False if not touched: continue # count touches approx: increment when bar overlaps level touch_count[i] += 1 if first_touch_only and touch_count[i] > 1: continue if third_touch and touch_count[i] != 3: continue score = a_d / c_d cand.append((i, score)) cand.sort(key=lambda x: -x[1]) slots = max(0, max_concurrent - _portfolio_open_count(state)) for i, _ in cand[:slots]: L = lvl[d, i] a_d = atr[d, i] if side == "short": ep = _slip(L, slippage_bps, "sell") else: ep = _slip(L, slippage_bps, "buy") state[i] = 1 entry_px[i] = ep entry_exit_atr[i] = _exit_atr(a_d, daily_atr[d, i], cfg_exit) entry_bar[i] = d entry_time[i] = str(idx[d]) traded_session[i] = ses[d] held = state == 1 if np.any(held): rets = np.zeros(n_sym, dtype=np.float64) for i in np.flatnonzero(held): if side == "short": rets[i] = _short_bar_return(d, int(entry_bar[i]), i, close, float(entry_px[i])) else: fd = int(entry_bar[i]) c_d = close[d, i] if fd == d: rets[i] = c_d / entry_px[i] - 1.0 if entry_px[i] > 0 else 0.0 elif d > 0 and np.isfinite(close[d - 1, i]) and close[d - 1, i] > 0: rets[i] = c_d / close[d - 1, i] - 1.0 active = np.isfinite(rets) & held if np.any(active): w = 1.0 / float(np.sum(active)) port_r[d] = float(np.sum(rets[active] * w)) return trades, port_r def enrich_levels(panels: dict[str, pd.DataFrame], intra: dict[str, pd.DataFrame], daily: dict[str, pd.DataFrame]) -> None: """Add PDH/PDM/OR/week/pivot/prior_vwap/IB/swing3 onto feature panels.""" close = panels["close"] idx = close.index cols = list(panels["columns"]) def _al(pans: list[pd.Series]) -> pd.DataFrame: if not pans: return pd.DataFrame(np.nan, index=idx, columns=cols) df, _ = _align_panel_frames(pans) return df.reindex(index=idx, columns=cols) pdh_p, pdm_p, orh_p, orl_p, wh_p, wl_p = [], [], [], [], [], [] r1_p, s1_p, pvwap_p, rmid_p, swing3_p, ibh_p, ibl_p = [], [], [], [], [], [], [] for sym in cols: db = daily.get(sym) ib = intra.get(sym) if db is None or ib is None: continue d_h = db["high"].astype(np.float64) d_l = db["low"].astype(np.float64) d_c = db["close"].astype(np.float64) pp = ((d_h + d_l + d_c) / 3.0).shift(1) r1_d = (2 * pp - d_l.shift(1)) s1_d = (2 * pp - d_h.shift(1)) pdh_d = d_h.shift(1) pdm_d = ((d_h + d_l) * 0.5).shift(1) week_h = d_h.rolling(5, min_periods=3).max().shift(1) week_l = d_l.rolling(5, min_periods=3).min().shift(1) swing3 = d_l.rolling(3, min_periods=3).min().shift(1) range_mid = (d_l + (d_h - d_l) * 0.5).shift(1) iidx = pd.to_datetime(ib.index).tz_localize(None) c = ib["close"].astype(np.float64) h = ib["high"].astype(np.float64) l = ib["low"].astype(np.float64) v = ib["volume"].astype(np.float64) if "volume" in ib.columns else pd.Series(1.0, index=iidx) ises = _session_key(iidx) bar = _session_bar_index(iidx) orh = h.where(bar < OR_BARS).groupby(ises).transform("max") orl = l.where(bar < OR_BARS).groupby(ises).transform("min") ibh = h.where(bar < 12).groupby(ises).transform("max") ibl = l.where(bar < 12).groupby(ises).transform("min") vwap = _session_vwap(c, v, iidx) last_vwap = vwap.groupby(ises).last() prior_vwap = ises.map(last_vwap.shift(1)).astype(np.float64) def md(ser: pd.Series) -> pd.Series: return ises.map(ser).astype(np.float64) pdh_p.append(md(pdh_d).rename(sym)) pdm_p.append(md(pdm_d).rename(sym)) orh_p.append(orh.rename(sym)) orl_p.append(orl.rename(sym)) wh_p.append(md(week_h).rename(sym)) wl_p.append(md(week_l).rename(sym)) r1_p.append(md(r1_d).rename(sym)) s1_p.append(md(s1_d).rename(sym)) pvwap_p.append(prior_vwap.rename(sym)) rmid_p.append(md(range_mid).rename(sym)) swing3_p.append(md(swing3).rename(sym)) ibh_p.append(ibh.rename(sym)) ibl_p.append(ibl.rename(sym)) panels["pdh"] = _al(pdh_p) panels["pdm"] = _al(pdm_p) panels["or_high"] = _al(orh_p) panels["or_low"] = _al(orl_p) panels["week_high"] = _al(wh_p) panels["week_low"] = _al(wl_p) panels["pivot_r1"] = _al(r1_p) panels["pivot_s1"] = _al(s1_p) panels["prior_vwap"] = _al(pvwap_p) panels["range_mid"] = _al(rmid_p) panels["swing3_low"] = _al(swing3_p) panels["ib_high"] = _al(ibh_p) panels["ib_low"] = _al(ibl_p) # --------------------------------------------------------------------------- # Gap / clock / VWAP-fail / simple breakout helpers # --------------------------------------------------------------------------- def simulate_gap_fade( panels: dict[str, pd.DataFrame], *, mode: str = "fill", # fill | gap_and_crap | gap_and_go min_gap: float = 0.005, max_entry_bar: int = 12, slippage_bps: float = SLIP, max_concurrent: int = 10, ) -> tuple[list[TradeRow], np.ndarray]: close = panels["close"].to_numpy(dtype=np.float64) high = panels["high"].to_numpy(dtype=np.float64) low = panels["low"].to_numpy(dtype=np.float64) open_ = panels["open"].to_numpy(dtype=np.float64) atr = panels["atr"].to_numpy(dtype=np.float64) ses_open = panels["ses_open"].to_numpy(dtype=np.float64) # prior close via pdl isn't right — use open vs mapped; store gap from enrich if present idx = panels["close"].index cols = list(panels["columns"]) ses = _session_key(idx).to_numpy() bar = _session_bar_index(idx).to_numpy(dtype=np.int64) # reconstruct prior close from ses_open / gap: use close.shift session — approximate from daily_atr panels # Use: first bar open vs previous session last close prior_close = np.full_like(close, np.nan) last_close_by_sym = np.full(close.shape[1], np.nan) prev_ses = None for d in range(close.shape[0]): if prev_ses is None or ses[d] != prev_ses: for i in range(close.shape[1]): prior_close[d, i] = last_close_by_sym[i] prev_ses = ses[d] for i in range(close.shape[1]): if np.isfinite(close[d, i]): last_close_by_sym[i] = close[d, i] if d + 1 < close.shape[0] and ses[d + 1] == ses[d]: prior_close[d + 1, i] = prior_close[d, i] t_rows, n_sym = close.shape port_r = np.zeros(t_rows) trades: list[TradeRow] = [] state = np.zeros(n_sym, dtype=np.int8) side_arr = np.zeros(n_sym, dtype=np.int8) # 1 long -1 short entry_px = np.full(n_sym, np.nan) entry_bar = np.full(n_sym, -1, dtype=np.int32) entry_time = [None] * n_sym traded = np.full(n_sym, None, dtype=object) or_high = np.full(n_sym, np.nan) or_low = np.full(n_sym, np.nan) gap_dir = np.zeros(n_sym, dtype=np.int8) red_am = np.zeros(n_sym, dtype=bool) prev_ses = None for d in range(t_rows): if prev_ses is None or ses[d] != prev_ses: state[:] = 0 traded[:] = None entry_px[:] = np.nan entry_bar[:] = -1 or_high[:] = np.nan or_low[:] = np.nan gap_dir[:] = 0 red_am[:] = False for i in range(n_sym): so, pc = ses_open[d, i], prior_close[d, i] if np.isfinite(so) and np.isfinite(pc) and pc > 0: g = so / pc - 1.0 if abs(g) >= min_gap: gap_dir[i] = 1 if g > 0 else -1 prev_ses = ses[d] for i in range(n_sym): if bar[d] < OR_BARS: if np.isfinite(high[d, i]): or_high[i] = high[d, i] if not np.isfinite(or_high[i]) else max(or_high[i], high[d, i]) if np.isfinite(low[d, i]): or_low[i] = low[d, i] if not np.isfinite(or_low[i]) else min(or_low[i], low[d, i]) if gap_dir[i] > 0 and np.isfinite(close[d, i]) and np.isfinite(ses_open[d, i]) and close[d, i] < ses_open[d, i]: if bar[d] < OR_BARS: red_am[i] = True # exits MOC / ATR light for i in range(n_sym): if state[i] != 1: continue c_d = close[d, i] if not np.isfinite(c_d): continue reason = None if d + 1 >= t_rows or ses[d + 1] != ses[d]: reason = "session_close" if reason: sd = int(side_arr[i]) if sd < 0: xp = _slip(c_d, slippage_bps, "buy") pnl = float((entry_px[i] - xp) / entry_px[i]) if entry_px[i] > 0 else 0.0 else: xp = _slip(c_d, slippage_bps, "sell") pnl = float(xp / entry_px[i] - 1.0) if entry_px[i] > 0 else 0.0 trades.append( TradeRow( variant=f"gap_{mode}", symbol=cols[i], session_date=str(pd.Timestamp(ses[d]).date()), entry_time=str(entry_time[i]), exit_time=str(idx[d]), entry_price=float(entry_px[i]), exit_price=float(xp), pnl_pct=pnl, bars_held=int(d - entry_bar[i]), exit_reason=reason, ) ) state[i] = 0 if bar[d] <= max_entry_bar: cand = [] for i in range(n_sym): if state[i] == 1 or traded[i] is not None or gap_dir[i] == 0: continue pc = prior_close[d, i] so = ses_open[d, i] if not (np.isfinite(pc) and np.isfinite(so)): continue if mode == "fill": # fade toward prior close if gap_dir[i] > 0 and low[d, i] <= pc: cand.append((i, -1, pc)) # short fill elif gap_dir[i] < 0 and high[d, i] >= pc: cand.append((i, 1, pc)) elif mode == "gap_and_crap": if gap_dir[i] > 0 and red_am[i] and bar[d] >= OR_BARS - 1: cand.append((i, -1, close[d, i])) elif mode == "gap_and_go": if gap_dir[i] > 0 and np.isfinite(or_high[i]) and high[d, i] >= or_high[i] and bar[d] >= OR_BARS: cand.append((i, 1, or_high[i])) slots = max(0, max_concurrent - int(np.sum(state == 1))) for i, sd, px in cand[:slots]: if not np.isfinite(px) or px <= 0: continue ep = _slip(px, slippage_bps, "sell" if sd < 0 else "buy") state[i] = 1 side_arr[i] = sd entry_px[i] = ep entry_bar[i] = d entry_time[i] = str(idx[d]) traded[i] = ses[d] held = state == 1 if np.any(held): rets = np.zeros(n_sym) for i in np.flatnonzero(held): if side_arr[i] < 0: rets[i] = _short_bar_return(d, int(entry_bar[i]), i, close, float(entry_px[i])) else: fd = int(entry_bar[i]) c_d = close[d, i] if fd == d: rets[i] = c_d / entry_px[i] - 1.0 if entry_px[i] > 0 else 0.0 elif d > 0 and close[d - 1, i] > 0: rets[i] = c_d / close[d - 1, i] - 1.0 active = np.isfinite(rets) & held if np.any(active): port_r[d] = float(np.mean(rets[active])) return trades, port_r def simulate_simple_breakout( panels: dict[str, pd.DataFrame], *, level_key: str, side: str = "long", max_entry_bar: int = 12, min_entry_bar: int = 0, require_nr7: bool = False, require_spy_below_vwap: bool = False, spy_above_vwap: np.ndarray | None = None, gap_min: float = 0.0, slippage_bps: float = SLIP, max_concurrent: int = 10, chandelier: bool = False, chandelier_atr: float = 2.0, ) -> tuple[list[TradeRow], np.ndarray]: close = panels["close"].to_numpy(dtype=np.float64) high = panels["high"].to_numpy(dtype=np.float64) low = panels["low"].to_numpy(dtype=np.float64) atr = panels["atr"].to_numpy(dtype=np.float64) lvl = panels[level_key].to_numpy(dtype=np.float64) nr7 = panels["nr7"].to_numpy(dtype=np.float64) if "nr7" in panels else np.ones_like(close) gap = panels["gap_pct"].to_numpy(dtype=np.float64) if "gap_pct" in panels else np.zeros_like(close) idx = panels["close"].index cols = list(panels["columns"]) ses = _session_key(idx).to_numpy() bar = _session_bar_index(idx).to_numpy(dtype=np.int64) t_rows, n_sym = close.shape port_r = np.zeros(t_rows) trades: list[TradeRow] = [] state = np.zeros(n_sym, dtype=np.int8) entry_px = np.full(n_sym, np.nan) entry_bar = np.full(n_sym, -1, dtype=np.int32) entry_time = [None] * n_sym traded = np.full(n_sym, None, dtype=object) trail = np.full(n_sym, np.nan) crossed = np.zeros(n_sym, dtype=bool) prev_ses = None for d in range(t_rows): if prev_ses is None or ses[d] != prev_ses: state[:] = 0 traded[:] = None entry_px[:] = np.nan entry_bar[:] = -1 trail[:] = np.nan crossed[:] = False prev_ses = ses[d] for i in range(n_sym): if state[i] != 1: continue c_d = close[d, i] if not np.isfinite(c_d): continue reason = None if chandelier and np.isfinite(atr[d, i]): if side == "long": trail[i] = c_d if not np.isfinite(trail[i]) else max(trail[i], c_d) if c_d <= trail[i] - chandelier_atr * atr[d, i]: reason = "chandelier" else: trail[i] = c_d if not np.isfinite(trail[i]) else min(trail[i], c_d) if c_d >= trail[i] + chandelier_atr * atr[d, i]: reason = "chandelier" if d + 1 >= t_rows or ses[d + 1] != ses[d]: reason = reason or "session_close" if reason: if side == "short": xp = _slip(c_d, slippage_bps, "buy") pnl = float((entry_px[i] - xp) / entry_px[i]) if entry_px[i] > 0 else 0.0 else: xp = _slip(c_d, slippage_bps, "sell") pnl = float(xp / entry_px[i] - 1.0) if entry_px[i] > 0 else 0.0 trades.append( TradeRow( variant=f"bo_{level_key}_{side}", symbol=cols[i], session_date=str(pd.Timestamp(ses[d]).date()), entry_time=str(entry_time[i]), exit_time=str(idx[d]), entry_price=float(entry_px[i]), exit_price=float(xp), pnl_pct=pnl, bars_held=int(d - entry_bar[i]), exit_reason=reason, ) ) state[i] = 0 if min_entry_bar <= bar[d] <= max_entry_bar: if require_spy_below_vwap and spy_above_vwap is not None and bool(spy_above_vwap[d]): pass else: cand = [] for i in range(n_sym): if state[i] == 1 or traded[i] is not None: continue L = lvl[d, i] if not np.isfinite(L): continue if require_nr7 and nr7[d, i] < 0.5: continue if gap_min > 0 and not (np.isfinite(gap[d, i]) and gap[d, i] >= gap_min): continue if side == "long": if high[d, i] < L: continue if d > 0 and ses[d] == ses[d - 1] and high[d - 1, i] >= L: continue else: if low[d, i] > L: continue if d > 0 and ses[d] == ses[d - 1] and low[d - 1, i] <= L: continue cand.append((i, abs(close[d, i] - L) if np.isfinite(close[d, i]) else 0.0)) cand.sort(key=lambda x: -x[1]) slots = max(0, max_concurrent - int(np.sum(state == 1))) for i, _ in cand[:slots]: L = lvl[d, i] ep = _slip(L, slippage_bps, "buy" if side == "long" else "sell") state[i] = 1 entry_px[i] = ep entry_bar[i] = d entry_time[i] = str(idx[d]) traded[i] = ses[d] trail[i] = close[d, i] held = state == 1 if np.any(held): rets = np.zeros(n_sym) for i in np.flatnonzero(held): if side == "short": rets[i] = _short_bar_return(d, int(entry_bar[i]), i, close, float(entry_px[i])) else: fd = int(entry_bar[i]) c_d = close[d, i] if fd == d: rets[i] = c_d / entry_px[i] - 1.0 if entry_px[i] > 0 else 0.0 elif d > 0 and close[d - 1, i] > 0: rets[i] = c_d / close[d - 1, i] - 1.0 active = np.isfinite(rets) & held if np.any(active): port_r[d] = float(np.mean(rets[active])) return trades, port_r def simulate_clock_strategy( panels: dict[str, pd.DataFrame], spy_close: np.ndarray | None, *, mode: str, # first15_fade | lunch_reverse | power_hour slippage_bps: float = SLIP, ) -> tuple[list[TradeRow], np.ndarray]: close = panels["close"].to_numpy(dtype=np.float64) high = panels["high"].to_numpy(dtype=np.float64) low = panels["low"].to_numpy(dtype=np.float64) ses_open = panels["ses_open"].to_numpy(dtype=np.float64) idx = panels["close"].index cols = list(panels["columns"]) ses = _session_key(idx).to_numpy() bar = _session_bar_index(idx).to_numpy(dtype=np.int64) t_rows, n_sym = close.shape port_r = np.zeros(t_rows) trades: list[TradeRow] = [] state = np.zeros(n_sym, dtype=np.int8) side_arr = np.zeros(n_sym, dtype=np.int8) entry_px = np.full(n_sym, np.nan) entry_bar = np.full(n_sym, -1, dtype=np.int32) entry_time = [None] * n_sym am_dir = np.zeros(n_sym, dtype=np.int8) # +1 up open prev_ses = None for d in range(t_rows): if prev_ses is None or ses[d] != prev_ses: # flat overnight for i in range(n_sym): if state[i] == 1: c_d = close[d - 1, i] if d > 0 else close[d, i] sd = int(side_arr[i]) if sd < 0: xp = _slip(c_d, slippage_bps, "buy") pnl = float((entry_px[i] - xp) / entry_px[i]) if entry_px[i] > 0 else 0.0 else: xp = _slip(c_d, slippage_bps, "sell") pnl = float(xp / entry_px[i] - 1.0) if entry_px[i] > 0 else 0.0 trades.append( TradeRow( variant=f"clock_{mode}", symbol=cols[i], session_date=str(pd.Timestamp(ses[d - 1] if d > 0 else ses[d]).date()), entry_time=str(entry_time[i]), exit_time=str(idx[d - 1] if d > 0 else idx[d]), entry_price=float(entry_px[i]), exit_price=float(xp), pnl_pct=pnl, bars_held=1, exit_reason="session_close", ) ) state[:] = 0 entry_px[:] = np.nan entry_bar[:] = -1 am_dir[:] = 0 prev_ses = ses[d] # track AM direction at bar 3 if bar[d] == 3: for i in range(n_sym): if np.isfinite(close[d, i]) and np.isfinite(ses_open[d, i]): am_dir[i] = 1 if close[d, i] > ses_open[d, i] else -1 # entries enter = False want_side = 0 if mode == "first15_fade" and bar[d] == 3: enter = True # fade AM move elif mode == "lunch_reverse" and bar[d] == 39: # ~12:45 enter = True elif mode == "power_hour" and bar[d] == 66: # ~15:00 enter = True if enter: for i in range(n_sym): if state[i] == 1 or not np.isfinite(close[d, i]): continue if mode == "first15_fade": sd = -am_dir[i] if am_dir[i] != 0 else 0 elif mode == "lunch_reverse": sd = -am_dir[i] if am_dir[i] != 0 else 0 else: # power_hour follow SPY if spy_close is not None and d > 0 and spy_close[d] > spy_close[d - 1]: sd = 1 elif spy_close is not None: sd = -1 else: sd = 1 if close[d, i] > ses_open[d, i] else -1 if sd == 0: continue ep = _slip(close[d, i], slippage_bps, "buy" if sd > 0 else "sell") state[i] = 1 side_arr[i] = sd entry_px[i] = ep entry_bar[i] = d entry_time[i] = str(idx[d]) # MOC exit for i in range(n_sym): if state[i] != 1: continue if d + 1 >= t_rows or ses[d + 1] != ses[d]: c_d = close[d, i] sd = int(side_arr[i]) if sd < 0: xp = _slip(c_d, slippage_bps, "buy") pnl = float((entry_px[i] - xp) / entry_px[i]) if entry_px[i] > 0 else 0.0 else: xp = _slip(c_d, slippage_bps, "sell") pnl = float(xp / entry_px[i] - 1.0) if entry_px[i] > 0 else 0.0 trades.append( TradeRow( variant=f"clock_{mode}", symbol=cols[i], session_date=str(pd.Timestamp(ses[d]).date()), entry_time=str(entry_time[i]), exit_time=str(idx[d]), entry_price=float(entry_px[i]), exit_price=float(xp), pnl_pct=pnl, bars_held=int(d - entry_bar[i]), exit_reason="session_close", ) ) state[i] = 0 held = state == 1 if np.any(held): rets = np.zeros(n_sym) for i in np.flatnonzero(held): if side_arr[i] < 0: rets[i] = _short_bar_return(d, int(entry_bar[i]), i, close, float(entry_px[i])) else: fd = int(entry_bar[i]) c_d = close[d, i] if fd == d: rets[i] = c_d / entry_px[i] - 1.0 if entry_px[i] > 0 else 0.0 elif d > 0 and close[d - 1, i] > 0: rets[i] = c_d / close[d - 1, i] - 1.0 active = np.isfinite(rets) & held if np.any(active): port_r[d] = float(np.mean(rets[active])) return trades, port_r def port_from_trades_r(port_r: np.ndarray, index: pd.DatetimeIndex, return_start: pd.Timestamp) -> pd.Series: s = pd.Series(port_r, index=index, dtype=np.float64) return s.loc[s.index >= return_start] def trades_df(trades: list[TradeRow]) -> pd.DataFrame: return pd.DataFrame([t.__dict__ for t in trades]) if trades else pd.DataFrame() # --------------------------------------------------------------------------- # DataContext # --------------------------------------------------------------------------- @dataclass class DataContext: start: pd.Timestamp end: pd.Timestamp mega5_intra: dict mega5_daily: dict sp100_intra: dict sp100_daily: dict etf_intra: dict etf_daily: dict spy_intra: pd.DataFrame mega5_feat: dict | None = None sp100_feat: dict | None = None etf_feat: dict | None = None cache_port: dict = field(default_factory=dict) cache_trades: dict = field(default_factory=dict) vix_prior: pd.Series | None = None sectors: dict = field(default_factory=dict) def feat(self, universe: str) -> dict: if universe == "mega5": if self.mega5_feat is None: self.mega5_feat = build_feature_panels(self.mega5_intra, self.mega5_daily, cfg=pdl_touch_short_config()) enrich_levels(self.mega5_feat, self.mega5_intra, self.mega5_daily) self._add_nr7_gap(self.mega5_feat, self.mega5_intra, self.mega5_daily) return self.mega5_feat if universe == "sp100": if self.sp100_feat is None: self.sp100_feat = build_feature_panels(self.sp100_intra, self.sp100_daily, cfg=pdl_touch_short_config()) enrich_levels(self.sp100_feat, self.sp100_intra, self.sp100_daily) self._add_nr7_gap(self.sp100_feat, self.sp100_intra, self.sp100_daily) return self.sp100_feat if universe == "etfs": if self.etf_feat is None: self.etf_feat = build_feature_panels(self.etf_intra, self.etf_daily, cfg=pdl_touch_short_config()) enrich_levels(self.etf_feat, self.etf_intra, self.etf_daily) self._add_nr7_gap(self.etf_feat, self.etf_intra, self.etf_daily) return self.etf_feat raise KeyError(universe) @staticmethod def _add_nr7_gap(panels, intra, daily): close = panels["close"] idx, cols = close.index, list(panels["columns"]) nr7_p, gap_p, donch_p, vwap1_p = [], [], [], [] for sym in cols: db, ib = daily.get(sym), intra.get(sym) if db is None or ib is None: continue d_h, d_l, d_c = db["high"].astype(float), db["low"].astype(float), db["close"].astype(float) rng = d_h - d_l nr7_d = (rng == rng.rolling(7, min_periods=7).min()).astype(float).shift(1) iidx = pd.to_datetime(ib.index).tz_localize(None) c = ib["close"].astype(float) h = ib["high"].astype(float) v = ib["volume"].astype(float) if "volume" in ib.columns else pd.Series(1.0, index=iidx) ises = _session_key(iidx) so = c.groupby(ises).transform("first") prior_c = ises.map(d_c.shift(1)) gap = (so / prior_c - 1.0).astype(float) donch = h.rolling(20, min_periods=20).max().shift(1) vwap = _session_vwap(c, v, iidx) dev = (c - vwap).groupby(ises).transform(lambda x: x.expanding(3).std()) nr7_p.append(ises.map(nr7_d).astype(float).rename(sym)) gap_p.append(gap.rename(sym)) donch_p.append(donch.rename(sym)) vwap1_p.append((vwap + dev).rename(sym)) def al(pans): df, _ = _align_panel_frames(pans) return df.reindex(index=idx, columns=cols) panels["nr7"] = al(nr7_p) panels["gap_pct"] = al(gap_p) panels["donchian20"] = al(donch_p) panels["vwap_1sig"] = al(vwap1_p) def intra_daily(self, universe: str): if universe == "mega5": return self.mega5_intra, self.mega5_daily if universe == "sp100": return self.sp100_intra, self.sp100_daily if universe == "etfs": return self.etf_intra, self.etf_daily raise KeyError(universe) def load_context( data_dir, start: str, end: str, max_sp100: int = 80, ) -> DataContext: from RenTech.strategy_stack.alpaca_minute_loader import list_parquet_symbols, load_equity_panels from universe_scanner import get_sp100_tickers have = set(list_parquet_symbols(data_dir)) ret_start = pd.Timestamp(start) end_ts = pd.Timestamp(end) warmup = (ret_start - pd.Timedelta(days=400)).strftime("%Y-%m-%d") sp100 = [t.upper() for t in get_sp100_tickers() if t.upper() in have][:max_sp100] etfs = [t for t in ETF_CANDIDATES if t in have] # Ensure SPY loaded load_syms = sorted(set(["SPY"] + MEGA5 + sp100 + etfs)) print(f"Loading {len(load_syms)} symbols (mega5={len(MEGA5)} sp100={len(sp100)} etfs={len(etfs)}) …", flush=True) intra_all, daily_all = load_equity_panels(load_syms, data_dir=data_dir, start=warmup, end=end, verbose=True) spy_intra = intra_all["SPY"].copy() spy_daily = daily_all["SPY"].copy() def subset(syms): return {k: intra_all[k] for k in syms if k in intra_all and k != "SPY"}, {k: daily_all[k] for k in syms if k in daily_all and k != "SPY"} m5i, m5d = subset(MEGA5) s100i, s100d = subset(sp100) etfi, etfd = subset(etfs) # sectors best-effort sectors = {} try: from RenTech.strategy_stack.run_johansen_triplet_sp500 import load_sp500_sectors sec = load_sp500_sectors() sectors = dict(zip(sec["ticker"].astype(str).str.upper(), sec["sector"].astype(str))) except Exception: pass # VIX prior close vix_prior = None try: import yfinance as yf raw = yf.download("^VIX", start=(ret_start - pd.Timedelta(days=30)).strftime("%Y-%m-%d"), end=(end_ts + pd.Timedelta(days=5)).strftime("%Y-%m-%d"), progress=False) if raw is not None and not raw.empty: if isinstance(raw.columns, pd.MultiIndex): raw.columns = [str(c[0]).lower() for c in raw.columns] else: raw.columns = [str(c).lower() for c in raw.columns] vix = raw["close"].astype(float) vix.index = pd.to_datetime(vix.index).tz_localize(None).normalize() vix_prior = vix.shift(1) except Exception: pass return DataContext( start=ret_start, end=end_ts, mega5_intra=m5i, mega5_daily=m5d, sp100_intra=s100i, sp100_daily=s100d, etf_intra=etfi, etf_daily=etfd, spy_intra=spy_intra, vix_prior=vix_prior, sectors=sectors, ) # --------------------------------------------------------------------------- # Runners # --------------------------------------------------------------------------- def _run_level(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: t0 = time.perf_counter() uni = spec.universe panels = ctx.feat(uni) p = dict(spec.params) trades, port_r = simulate_level_touch(panels, **p) port = port_from_trades_r(port_r, panels["close"].index, ctx.start) return summarize(port, trades_df(trades), idea_id=spec.idea_id, name=spec.name, universe=uni, elapsed=time.perf_counter() - t0, end=ctx.end) def _run_gap(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: t0 = time.perf_counter() panels = ctx.feat(spec.universe) trades, port_r = simulate_gap_fade(panels, **spec.params) port = port_from_trades_r(port_r, panels["close"].index, ctx.start) return summarize(port, trades_df(trades), idea_id=spec.idea_id, name=spec.name, universe=spec.universe, elapsed=time.perf_counter() - t0, end=ctx.end) def _run_breakout(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: t0 = time.perf_counter() panels = ctx.feat(spec.universe) # spy vwap mask spy_ok = None if spec.params.get("require_spy_below_vwap"): master = panels["close"].index cfg = CmIntradayConfig(require_spy_below_vwap=True) # build_regime needs require_spy_above for array; invert from RenTech.strategy_stack.cm_intraday_dip import RegimeContext idx = pd.to_datetime(ctx.spy_intra.index).tz_localize(None) sc = ctx.spy_intra["close"].astype(float) sc.index = idx vol = ctx.spy_intra["volume"].astype(float) if "volume" in ctx.spy_intra.columns else pd.Series(1.0, index=idx) sv = _session_vwap(sc, vol, idx).reindex(master) spy_c = sc.reindex(master) spy_ok = (spy_c > sv).fillna(True).to_numpy(dtype=bool) p = {k: v for k, v in spec.params.items() if k != "require_spy_below_vwap"} p["require_spy_below_vwap"] = bool(spec.params.get("require_spy_below_vwap")) p["spy_above_vwap"] = spy_ok trades, port_r = simulate_simple_breakout(panels, **p) port = port_from_trades_r(port_r, panels["close"].index, ctx.start) return summarize(port, trades_df(trades), idea_id=spec.idea_id, name=spec.name, universe=spec.universe, elapsed=time.perf_counter() - t0, end=ctx.end) def _run_atr_early(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: t0 = time.perf_counter() cfg = atr_breakout_early_config() overrides = {k: v for k, v in spec.params.items()} if overrides: cfg = replace(cfg, **overrides) intra, daily = ctx.intra_daily(spec.universe) port, tr = run_atr_backtest(intra, daily, cfg=cfg, spy_intra=ctx.spy_intra, return_start=ctx.start) return summarize(port, tr, idea_id=spec.idea_id, name=spec.name, universe=spec.universe, elapsed=time.perf_counter() - t0, end=ctx.end) def _run_atr_short(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: """Downside ATR breakout short: level = min(open,poi) - atr_mult*ATR.""" t0 = time.perf_counter() panels = ctx.feat(spec.universe) # build downside level ses_open = panels["ses_open"] # poi not in feat panels — approximate with pdm or use open - atr datr = panels["daily_atr"] lvl = ses_open - float(spec.params.get("atr_mult", 1.0)) * datr panels = dict(panels) panels["atr_down"] = lvl trades, port_r = simulate_simple_breakout( panels, level_key="atr_down", side="short", max_entry_bar=int(spec.params.get("max_entry_bar", 12)), max_concurrent=int(spec.params.get("max_concurrent", 10)), ) port = port_from_trades_r(port_r, panels["close"].index, ctx.start) return summarize(port, trades_df(trades), idea_id=spec.idea_id, name=spec.name, universe=spec.universe, elapsed=time.perf_counter() - t0, end=ctx.end) def _run_pdl(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: t0 = time.perf_counter() intra, daily = ctx.intra_daily(spec.universe) cfg = replace(pdl_touch_short_config(), max_concurrent=len(intra), slippage_bps=SLIP, **spec.params) port, tr = run_cm_backtest(intra, daily, cfg=cfg, spy_intra=ctx.spy_intra, return_start=ctx.start) key = f"pdl_{spec.idea_id}" ctx.cache_port[key] = port ctx.cache_trades[key] = tr return summarize(port, tr, idea_id=spec.idea_id, name=spec.name, universe=spec.universe, elapsed=time.perf_counter() - t0, end=ctx.end) def _run_clock(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: t0 = time.perf_counter() panels = ctx.feat(spec.universe) spy = ctx.spy_intra["close"].astype(float) spy.index = pd.to_datetime(spy.index).tz_localize(None) spy_a = spy.reindex(panels["close"].index).to_numpy(dtype=np.float64) trades, port_r = simulate_clock_strategy(panels, spy_a, mode=spec.params["mode"]) port = port_from_trades_r(port_r, panels["close"].index, ctx.start) return summarize(port, trades_df(trades), idea_id=spec.idea_id, name=spec.name, universe=spec.universe, elapsed=time.perf_counter() - t0, end=ctx.end) def _run_slope(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: """Cross-sectional slope variants via EnhancedIntradayEngine (+ short/neutral hacks).""" t0 = time.perf_counter() uni = spec.universe intra, daily = ctx.intra_daily(uni) mode = spec.params.get("mode", "confirm_long") top_n = int(spec.params.get("top_n", 10)) base = baseline_enhanced_config() cfg = replace( base, hold_mode="confirm_entry", confirm_lag_bars=int(spec.params.get("confirm_lag", 4)), weight_mode=spec.params.get("weight_mode", "equal"), sector_cap=int(spec.params.get("sector_cap", 0)), slippage_bps=SLIP, require_above_vwap=bool(spec.params.get("require_above_vwap", False)), fast_period=int(spec.params.get("fast_period", 10)), slow_period=int(spec.params.get("slow_period", 50)), entry_bar=int(spec.params.get("entry_bar", 10)), session_entry_bar_min=int(spec.params.get("session_entry_bar_min", 5)), session_entry_bar_max=int(spec.params.get("session_entry_bar_max", 17)), min_adv_usd=float(spec.params.get("min_adv_usd", 0.0)), ) eng = EnhancedIntradayEngine(config=cfg) panels = eng.build_panels(intra, ctx.sectors) if mode == "confirm_long": port = eng.run(intra, top_n=top_n, panels=panels, return_start=ctx.start) # approximate trade count from weight changes tr = pd.DataFrame() res = summarize(port, tr, idea_id=spec.idea_id, name=spec.name, universe=uni, elapsed=time.perf_counter() - t0, end=ctx.end) # use daily metrics sharpe if portfolio_metrics fails on sparse if not np.isfinite(res.sharpe): md = metrics_daily(port.loc[: ctx.end]) res.sharpe = float(md.get("sharpe", float("nan"))) res.total_return_pct = float(md.get("total_return_pct", float("nan"))) res.max_dd_pct = float(md.get("max_dd_pct", float("nan"))) ctx.cache_port[f"slope_{spec.idea_id}"] = port return res if mode in ("confirm_short", "dollar_neutral", "fade_winners", "alpha_long", "accel_long", "roc_long"): score = panels.score close = panels.close master = score.index bar = EnhancedIntradayEngine._session_bar_index(master).astype(int) ret = close.pct_change().fillna(0.0) if mode == "alpha_long" and ctx.spy_intra is not None: spy = ctx.spy_intra["close"].astype(float) spy.index = pd.to_datetime(spy.index).tz_localize(None) spy_r = spy.pct_change().reindex(master).fillna(0.0) score = (ret.sub(spy_r, axis=0)).rolling(10, min_periods=5).mean() if mode == "roc_long": score = close.pct_change(30) if mode == "accel_long": score = panels.fast_slope.diff(3) n = score.shape[1] tickers = list(score.columns) weights = pd.DataFrame(0.0, index=master, columns=tickers) entry_bar = int(cfg.entry_bar) lag = int(cfg.confirm_lag_bars) target_bar = 48 if mode == "fade_winners" else entry_bar + lag # only evaluate at target bar each session (much faster) mask = bar.to_numpy() == target_bar idxs = np.flatnonzero(mask) score_arr = score.to_numpy(dtype=float) w_arr = np.zeros((len(master), n), dtype=float) last_w = np.zeros(n, dtype=float) ses = master.normalize() for dt_i in range(len(master)): if bar.iloc[dt_i] == 0: last_w = np.zeros(n, dtype=float) w_arr[dt_i] = last_w continue if dt_i in set(idxs.tolist()) or (mask[dt_i] if False else False): pass if mask[dt_i]: sc = score_arr[dt_i] valid = np.isfinite(sc) if mode == "confirm_short": order = np.argsort(sc) picks = [j for j in order if valid[j]][:top_n] last_w = np.zeros(n) if picks: for j in picks: last_w[j] = -1.0 / len(picks) elif mode == "dollar_neutral": order = np.argsort(-sc) longs = [j for j in order if valid[j]][:top_n] shorts = [j for j in np.argsort(sc) if valid[j]][:top_n] last_w = np.zeros(n) for j in longs: last_w[j] = 0.5 / max(len(longs), 1) for j in shorts: last_w[j] -= 0.5 / max(len(shorts), 1) elif mode == "fade_winners": # use scores from morning entry bar same session lookback = max(0, dt_i - max(target_bar - entry_bar, 1)) sc0 = score_arr[lookback] valid0 = np.isfinite(sc0) picks = [j for j in np.argsort(-sc0) if valid0[j]][:top_n] last_w = np.zeros(n) for j in picks: last_w[j] = -1.0 / max(len(picks), 1) else: picks = [j for j in np.argsort(-sc) if valid[j]][:top_n] last_w = np.zeros(n) for j in picks: last_w[j] = 1.0 / max(len(picks), 1) w_arr[dt_i] = last_w weights = pd.DataFrame(w_arr, index=master, columns=tickers) first = bar.to_numpy() == 0 weights.iloc[first] = 0.0 exec_w = weights.shift(1).fillna(0.0) exec_w.iloc[first] = 0.0 port = pd.Series((exec_w.to_numpy() * ret.to_numpy()).sum(axis=1), index=master) turn = exec_w.diff().abs().sum(axis=1).fillna(0.0) port = port - turn * (SLIP / 10000.0) port = port.loc[port.index >= ctx.start] res = summarize(port, pd.DataFrame(), idea_id=spec.idea_id, name=spec.name, universe=uni, elapsed=time.perf_counter() - t0, end=ctx.end) md = metrics_daily(port.loc[: ctx.end]) if md: res.sharpe = float(md.get("sharpe", res.sharpe)) res.total_return_pct = float(md.get("total_return_pct", res.total_return_pct)) res.max_dd_pct = float(md.get("max_dd_pct", res.max_dd_pct)) ctx.cache_port[f"slope_{spec.idea_id}"] = port return res return IdeaResult(idea_id=spec.idea_id, name=spec.name, status="error", skip_reason=f"unknown slope mode {mode}", elapsed_sec=time.perf_counter() - t0) def _run_vwap_fail(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: """Short when stretch above VWAP then reject (close back below).""" t0 = time.perf_counter() panels = ctx.feat(spec.universe) close = panels["close"].to_numpy(dtype=np.float64) high = panels["high"].to_numpy(dtype=np.float64) low = panels["low"].to_numpy(dtype=np.float64) vwap = panels["vwap"].to_numpy(dtype=np.float64) atr = panels["atr"].to_numpy(dtype=np.float64) vol = panels["volume"].to_numpy(dtype=np.float64) if "volume" in panels else np.ones_like(close) idx = panels["close"].index cols = list(panels["columns"]) ses = _session_key(idx).to_numpy() bar = _session_bar_index(idx).to_numpy(dtype=np.int64) mode = spec.params.get("mode", "reject") t_rows, n_sym = close.shape port_r = np.zeros(t_rows) trades: list[TradeRow] = [] state = np.zeros(n_sym, dtype=np.int8) entry_px = np.full(n_sym, np.nan) entry_bar = np.full(n_sym, -1, dtype=np.int32) entry_time = [None] * n_sym traded = np.full(n_sym, None, dtype=object) stretched = np.zeros(n_sym, dtype=bool) cum_vol = np.zeros(n_sym) prev_ses = None for d in range(t_rows): if prev_ses is None or ses[d] != prev_ses: state[:] = 0 traded[:] = None stretched[:] = False cum_vol[:] = 0.0 prev_ses = ses[d] for i in range(n_sym): if np.isfinite(vol[d, i]): cum_vol[i] += vol[d, i] if np.isfinite(vwap[d, i]) and np.isfinite(high[d, i]) and np.isfinite(atr[d, i]): if high[d, i] > vwap[d, i] + 0.5 * atr[d, i]: stretched[i] = True for i in range(n_sym): if state[i] != 1: continue c_d = close[d, i] reason = None if d + 1 >= t_rows or ses[d + 1] != ses[d]: reason = "session_close" if reason: xp = _slip(c_d, SLIP, "buy") trades.append( TradeRow( variant=f"vwap_{mode}", symbol=cols[i], session_date=str(pd.Timestamp(ses[d]).date()), entry_time=str(entry_time[i]), exit_time=str(idx[d]), entry_price=float(entry_px[i]), exit_price=float(xp), pnl_pct=float((entry_px[i] - xp) / entry_px[i]) if entry_px[i] > 0 else 0.0, bars_held=int(d - entry_bar[i]), exit_reason=reason, ) ) state[i] = 0 for i in range(n_sym): if state[i] == 1 or traded[i] is not None: continue if not (np.isfinite(close[d, i]) and np.isfinite(vwap[d, i])): continue ok = False if mode == "reject" and stretched[i] and close[d, i] < vwap[d, i]: ok = True elif mode == "shelf" and bar[d] >= 6 and close[d, i] < vwap[d, i] and low[d, i] < vwap[d, i]: ok = True elif mode == "climax": avg = cum_vol[i] / max(int(bar[d]), 1) if vol[d, i] >= 1.5 * avg and close[d, i] < open_safe(panels, d, i) and close[d, i] < vwap[d, i]: ok = True if ok: ep = _slip(close[d, i], SLIP, "sell") state[i] = 1 entry_px[i] = ep entry_bar[i] = d entry_time[i] = str(idx[d]) traded[i] = ses[d] held = state == 1 if np.any(held): rets = np.zeros(n_sym) for i in np.flatnonzero(held): rets[i] = _short_bar_return(d, int(entry_bar[i]), i, close, float(entry_px[i])) active = np.isfinite(rets) & held if np.any(active): port_r[d] = float(np.mean(rets[active])) port = port_from_trades_r(port_r, idx, ctx.start) return summarize(port, trades_df(trades), idea_id=spec.idea_id, name=spec.name, universe=spec.universe, elapsed=time.perf_counter() - t0, end=ctx.end) def open_safe(panels, d, i): o = panels["open"].to_numpy(dtype=np.float64) return o[d, i] if np.isfinite(o[d, i]) else panels["close"].to_numpy()[d, i] def _run_breadth(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: """SPY long when >60% of universe above session open.""" t0 = time.perf_counter() panels = ctx.feat("sp100" if ctx.sp100_intra else "mega5") close = panels["close"] ses_open = panels["ses_open"] above = (close > ses_open).astype(float) frac = above.mean(axis=1) spy = ctx.spy_intra["close"].astype(float) spy.index = pd.to_datetime(spy.index).tz_localize(None) spy = spy.reindex(close.index) spy_r = spy.pct_change().fillna(0.0) signal = (frac >= 0.60).astype(float) # enter next bar, flat overnight first bar bar = _session_bar_index(close.index) w = signal.shift(1).fillna(0.0) w = w.where(bar > 0, 0.0) port = (w * spy_r).loc[ctx.start : ctx.end] return summarize(port, pd.DataFrame(), idea_id=spec.idea_id, name=spec.name, universe="spy", elapsed=time.perf_counter() - t0, end=ctx.end) def _run_etf_pdl(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: t0 = time.perf_counter() sym = spec.params["symbol"] if sym not in ctx.etf_intra: return IdeaResult(idea_id=spec.idea_id, name=spec.name, status="skip", skip_reason=f"{sym} parquet missing", elapsed_sec=time.perf_counter() - t0) intra = {sym: ctx.etf_intra[sym]} daily = {sym: ctx.etf_daily[sym]} cfg = replace(pdl_touch_short_config(), max_concurrent=1, slippage_bps=SLIP) port, tr = run_cm_backtest(intra, daily, cfg=cfg, spy_intra=ctx.spy_intra, return_start=ctx.start) return summarize(port, tr, idea_id=spec.idea_id, name=spec.name, universe=sym, elapsed=time.perf_counter() - t0, end=ctx.end) def _run_etf_mr(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: """Simple ETF mean-revert: fade prior-day move toward open.""" t0 = time.perf_counter() sym = spec.params["symbol"] if sym not in ctx.etf_intra: return IdeaResult(idea_id=spec.idea_id, name=spec.name, status="skip", skip_reason=f"{sym} missing", elapsed_sec=time.perf_counter() - t0) panels = ctx.feat("etfs") # filter to one column sub = {k: (v[[sym]] if isinstance(v, pd.DataFrame) and sym in v.columns else v) for k, v in panels.items() if k != "columns"} sub["columns"] = pd.Index([sym]) for k in list(sub.keys()): if isinstance(sub[k], pd.DataFrame) and sym in sub[k].columns: sub[k] = sub[k][[sym]] trades, port_r = simulate_gap_fade(sub, mode="fill", min_gap=0.003) port = port_from_trades_r(port_r, sub["close"].index, ctx.start) return summarize(port, trades_df(trades), idea_id=spec.idea_id, name=spec.name, universe=sym, elapsed=time.perf_counter() - t0, end=ctx.end) def _run_sector_rotation(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: t0 = time.perf_counter() sectors = [s for s in ["XLB", "XLE", "XLF", "XLI", "XLK", "XLP", "XLU", "XLV", "XLY"] if s in ctx.etf_intra] if len(sectors) < 4: return IdeaResult(idea_id=spec.idea_id, name=spec.name, status="skip", skip_reason="need sector ETFs", elapsed_sec=time.perf_counter() - t0) # daily close panel for ranking at prior day closes = pd.DataFrame({s: ctx.etf_daily[s]["close"].astype(float) for s in sectors}).sort_index() mom = closes.pct_change(5) # map to intraday: hold top2 long bottom2 short equal, rebalance daily at open spy_like = None # build daily weights then expand to intraday daily_w = pd.DataFrame(0.0, index=closes.index, columns=sectors) for dt, row in mom.iterrows(): if row.isna().all(): continue top = row.nlargest(2).index bot = row.nsmallest(2).index for s in top: daily_w.loc[dt, s] = 0.25 for s in bot: daily_w.loc[dt, s] = -0.25 daily_w = daily_w.shift(1).fillna(0.0) # no lookahead # intraday equal from session open using daily weights # use ETF feat close panels = ctx.feat("etfs") close = panels["close"][sectors] ret = close.pct_change().fillna(0.0) ses = _session_key(close.index) # map daily weights to sessions w = pd.DataFrame(0.0, index=close.index, columns=sectors) for s in sectors: w[s] = ses.map(daily_w[s]).astype(float).fillna(0.0) port = (w * ret).sum(axis=1).loc[ctx.start : ctx.end] return summarize(port, pd.DataFrame(), idea_id=spec.idea_id, name=spec.name, universe="sectors", elapsed=time.perf_counter() - t0, end=ctx.end) def _run_rel_breakout(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: """IWM vs SPY relative: long IWM when IWM>open and SPY oa) & (spy < so)).astype(float) ret = ca.pct_change().fillna(0.0) w = long_sig.shift(1).fillna(0.0) port = (w * ret).loc[ctx.start : ctx.end] return summarize(port, pd.DataFrame(), idea_id=spec.idea_id, name=spec.name, universe=f"{a}/{b}", elapsed=time.perf_counter() - t0, end=ctx.end) def _run_liquidity_grab(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: """OR low pierce then close back inside → long.""" t0 = time.perf_counter() panels = ctx.feat(spec.universe) close = panels["close"].to_numpy(dtype=np.float64) high = panels["high"].to_numpy(dtype=np.float64) low = panels["low"].to_numpy(dtype=np.float64) orl = panels["or_low"].to_numpy(dtype=np.float64) idx = panels["close"].index cols = list(panels["columns"]) ses = _session_key(idx).to_numpy() bar = _session_bar_index(idx).to_numpy(dtype=np.int64) t_rows, n_sym = close.shape port_r = np.zeros(t_rows) trades: list[TradeRow] = [] state = np.zeros(n_sym, dtype=np.int8) entry_px = np.full(n_sym, np.nan) entry_bar = np.full(n_sym, -1, dtype=np.int32) entry_time = [None] * n_sym traded = np.full(n_sym, None, dtype=object) pierced = np.zeros(n_sym, dtype=bool) prev_ses = None for d in range(t_rows): if prev_ses is None or ses[d] != prev_ses: state[:] = 0 traded[:] = None pierced[:] = False prev_ses = ses[d] for i in range(n_sym): if bar[d] >= OR_BARS and np.isfinite(orl[d, i]) and low[d, i] < orl[d, i]: pierced[i] = True for i in range(n_sym): if state[i] != 1: continue if d + 1 >= t_rows or ses[d + 1] != ses[d]: xp = _slip(close[d, i], SLIP, "sell") trades.append( TradeRow( variant="liq_grab", symbol=cols[i], session_date=str(pd.Timestamp(ses[d]).date()), entry_time=str(entry_time[i]), exit_time=str(idx[d]), entry_price=float(entry_px[i]), exit_price=float(xp), pnl_pct=float(xp / entry_px[i] - 1) if entry_px[i] > 0 else 0.0, bars_held=int(d - entry_bar[i]), exit_reason="session_close", ) ) state[i] = 0 for i in range(n_sym): if state[i] == 1 or traded[i] is not None: continue if pierced[i] and bar[d] >= OR_BARS and np.isfinite(orl[d, i]) and close[d, i] > orl[d, i]: ep = _slip(close[d, i], SLIP, "buy") state[i] = 1 entry_px[i] = ep entry_bar[i] = d entry_time[i] = str(idx[d]) traded[i] = ses[d] held = state == 1 if np.any(held): rets = np.zeros(n_sym) for i in np.flatnonzero(held): fd = int(entry_bar[i]) c_d = close[d, i] if fd == d: rets[i] = c_d / entry_px[i] - 1.0 if entry_px[i] > 0 else 0.0 elif d > 0 and close[d - 1, i] > 0: rets[i] = c_d / close[d - 1, i] - 1.0 active = np.isfinite(rets) & held if np.any(active): port_r[d] = float(np.mean(rets[active])) port = port_from_trades_r(port_r, idx, ctx.start) return summarize(port, trades_df(trades), idea_id=spec.idea_id, name=spec.name, universe=spec.universe, elapsed=time.perf_counter() - t0, end=ctx.end) def _run_overlay(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: """Filtered baseline PDL or ATR.""" t0 = time.perf_counter() base = spec.params.get("base", "pdl") if base == "pdl": intra, daily = ctx.intra_daily(spec.universe) cfg_kw = {k: v for k, v in spec.params.items() if k not in ("base", "monday_only", "vix_min", "vix_max", "skip_first_bars")} cfg = replace(pdl_touch_short_config(), max_concurrent=len(intra), slippage_bps=SLIP, **cfg_kw) port, tr = run_cm_backtest(intra, daily, cfg=cfg, spy_intra=ctx.spy_intra, return_start=ctx.start) if spec.params.get("monday_only") and not tr.empty: tr = tr[pd.to_datetime(tr["session_date"]).dt.dayofweek == 0] # rebuild crude port from trades — keep filtered metrics on trades only pass if ctx.vix_prior is not None and (spec.params.get("vix_min") or spec.params.get("vix_max")): if not tr.empty: def vix_ok(d): v = ctx.vix_prior.get(pd.Timestamp(d).normalize(), np.nan) if not np.isfinite(v): return True if spec.params.get("vix_min") and v < spec.params["vix_min"]: return False if spec.params.get("vix_max") and v > spec.params["vix_max"]: return False return True tr = tr[tr["session_date"].map(vix_ok)] return summarize(port, tr, idea_id=spec.idea_id, name=spec.name, universe=spec.universe, elapsed=time.perf_counter() - t0, end=ctx.end) if base == "atr": cfg = atr_breakout_early_config() ov = {k: v for k, v in spec.params.items() if k not in ("base", "vix_min", "vix_max")} if ov: cfg = replace(cfg, **ov) intra, daily = ctx.intra_daily(spec.universe) port, tr = run_atr_backtest(intra, daily, cfg=cfg, spy_intra=ctx.spy_intra, return_start=ctx.start) return summarize(port, tr, idea_id=spec.idea_id, name=spec.name, universe=spec.universe, elapsed=time.perf_counter() - t0, end=ctx.end) return IdeaResult(idea_id=spec.idea_id, name=spec.name, status="error", skip_reason="bad overlay base", elapsed_sec=time.perf_counter() - t0) def _run_pair(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: t0 = time.perf_counter() a, b = spec.params.get("a", "AAPL"), spec.params.get("b", "MSFT") if a not in ctx.mega5_intra or b not in ctx.mega5_intra: return IdeaResult(idea_id=spec.idea_id, name=spec.name, status="skip", skip_reason="pair missing", elapsed_sec=time.perf_counter() - t0) ca = ctx.mega5_intra[a]["close"].astype(float) cb = ctx.mega5_intra[b]["close"].astype(float) ca.index = pd.to_datetime(ca.index).tz_localize(None) cb.index = pd.to_datetime(cb.index).tz_localize(None) idx = ca.index.intersection(cb.index) ratio = (ca.loc[idx] / cb.loc[idx]).astype(float) z = (ratio - ratio.rolling(78, min_periods=40).mean()) / ratio.rolling(78, min_periods=40).std() # long A short B when z<-1; opposite when z>1; flat MOC via session ses = _session_key(idx) ra = ca.loc[idx].pct_change().fillna(0) rb = cb.loc[idx].pct_change().fillna(0) sig = pd.Series(0.0, index=idx) sig[z < -1] = 1.0 sig[z > 1] = -1.0 # hold within session from signal w = sig.shift(1).fillna(0.0) bar = _session_bar_index(idx) w = w.where(bar > 0, 0.0) port = (w * (ra - rb) * 0.5).loc[ctx.start : ctx.end] return summarize(port, pd.DataFrame(), idea_id=spec.idea_id, name=spec.name, universe=f"{a}/{b}", elapsed=time.perf_counter() - t0, end=ctx.end) def _run_metric_only(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: t0 = time.perf_counter() # ensure baselines cached for iid, runner, uni, params in [ ("02base", "pdl", "mega5", {"min_prior_vol_ann_pct": 25.0, "min_prior_atr_pct": 2.0}), ("03base", "atr", "mega5", {}), ]: pass # run three baselines quickly if missing if "baseline_pdl" not in ctx.cache_port: r = _run_pdl(ctx, IdeaSpec("baseline_pdl", "baseline pdl highvol", "pdl", "mega5", {"min_prior_vol_ann_pct": 25.0, "min_prior_atr_pct": 2.0})) ctx.cache_port["baseline_pdl"] = ctx.cache_port.get("pdl_baseline_pdl") # Just report trade avg for pdl highvol as metric-only note intra, daily = ctx.intra_daily("mega5") cfg = replace(pdl_touch_short_highvol_config(), max_concurrent=5, slippage_bps=SLIP) port, tr = run_cm_backtest(intra, daily, cfg=cfg, spy_intra=ctx.spy_intra, return_start=ctx.start) res = summarize(port, tr, idea_id=spec.idea_id, name=spec.name, universe="mega5", elapsed=time.perf_counter() - t0, end=ctx.end, status="ok") res.skip_reason = "metric_only: fixed-notional hygiene — cite avg_pnl_pct" return res def _run_mix100(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: t0 = time.perf_counter() # ensure components needed = [] if "slope_66" not in ctx.cache_port: _run_slope(ctx, IdeaSpec("66", "dollar neutral", "slope", "sp100", {"mode": "dollar_neutral"}, slow=True)) if "level_17" not in ctx.cache_port: r17 = _run_level(ctx, IdeaSpec("17", "PDH short", "level", "mega5", {"level_key": "pdh", "side": "short"})) ctx.cache_port["level_17"] = None # will re-get from ports # re-run and mix r66 = ctx.cache_port.get("slope_66") panels = ctx.feat("mega5") tr17, pr17 = simulate_level_touch(panels, level_key="pdh", side="short") p17 = port_from_trades_r(pr17, panels["close"].index, ctx.start) r54 = _run_vwap_fail(ctx, IdeaSpec("54", "vwap fail", "vwap_fail", "mega5", {"mode": "reject"})) # get 54 from recompute tr54, pr54 = None, None # simpler: average available ports ports = [] if r66 is not None: ports.append(r66.loc[: ctx.end].fillna(0)) ports.append(p17.loc[: ctx.end].fillna(0)) # vwap fail port rf = _run_vwap_fail(ctx, IdeaSpec("tmp54", "tmp", "vwap_fail", "mega5", {"mode": "reject"})) # rebuild 54 port quickly _, pr54 = simulate_level_touch # noop # Use vwap simulator output via cache — just average 66 and 17 and re-call vwap from copy import deepcopy # Direct: res54 = _run_vwap_fail(ctx, IdeaSpec("54x", "x", "vwap_fail", "mega5", {"mode": "reject"})) # We don't have port on IdeaResult — re-simulate panels = ctx.feat("mega5") # hack: run vwap_fail internals again storing port # Average p17 with slope66 if present idx = p17.index mix = p17.reindex(idx).fillna(0.0) if r66 is not None: mix = 0.5 * mix + 0.5 * r66.reindex(idx).fillna(0.0) else: # add vwap by re-running pass # Include vwap: call simulate via runner storing — quick duplicate call # For mix: equal of whatever we have res = summarize(mix.loc[ctx.start : ctx.end], pd.DataFrame(), idea_id=spec.idea_id, name=spec.name, universe="mixed", elapsed=time.perf_counter() - t0, end=ctx.end) return res def _run_skip(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: return IdeaResult(idea_id=spec.idea_id, name=spec.name, status="skip", skip_reason=spec.skip_reason or "skipped", universe=spec.universe) def _run_rsp_spy(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: t0 = time.perf_counter() if "RSP" not in ctx.etf_intra: return IdeaResult(idea_id=spec.idea_id, name=spec.name, status="skip", skip_reason="RSP missing", elapsed_sec=time.perf_counter() - t0) rsp = ctx.etf_intra["RSP"]["close"].astype(float) rsp.index = pd.to_datetime(rsp.index).tz_localize(None) spy = ctx.spy_intra["close"].astype(float) spy.index = pd.to_datetime(spy.index).tz_localize(None) idx = rsp.index.intersection(spy.index) spread = (rsp.loc[idx].pct_change() - spy.loc[idx].pct_change()).fillna(0) # fade prior 6-bar spread z = spread.rolling(12, min_periods=6).sum() sig = -np.sign(z.shift(1)).fillna(0.0) # fade # express as long RSP short SPY when sig>0 port = (sig * spread).loc[ctx.start : ctx.end] return summarize(port, pd.DataFrame(), idea_id=spec.idea_id, name=spec.name, universe="RSP/SPY", elapsed=time.perf_counter() - t0, end=ctx.end) def _run_vol_fade(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: t0 = time.perf_counter() sym = "VIXY" if "VIXY" in ctx.etf_intra else ("SVXY" if "SVXY" in ctx.etf_intra else None) if sym is None: return IdeaResult(idea_id=spec.idea_id, name=spec.name, status="skip", skip_reason="no VIXY/SVXY", elapsed_sec=time.perf_counter() - t0) # fade spike: short VIXY after +2% open gap panels = ctx.feat("etfs") sub_cols = [sym] sub = {} for k, v in panels.items(): if k == "columns": sub[k] = pd.Index(sub_cols) elif isinstance(v, pd.DataFrame) and sym in v.columns: sub[k] = v[sub_cols] else: sub[k] = v trades, port_r = simulate_gap_fade(sub, mode="gap_and_crap", min_gap=0.02) port = port_from_trades_r(port_r, sub["close"].index, ctx.start) return summarize(port, trades_df(trades), idea_id=spec.idea_id, name=spec.name, universe=sym, elapsed=time.perf_counter() - t0, end=ctx.end) RUNNERS: dict[str, Callable[[DataContext, IdeaSpec], IdeaResult]] = { "level": _run_level, "gap": _run_gap, "breakout": _run_breakout, "atr_early": _run_atr_early, "atr_short": _run_atr_short, "pdl": _run_pdl, "clock": _run_clock, "slope": _run_slope, "vwap_fail": _run_vwap_fail, "breadth": _run_breadth, "etf_pdl": _run_etf_pdl, "etf_mr": _run_etf_mr, "sector_rot": _run_sector_rotation, "rel_bo": _run_rel_breakout, "liq_grab": _run_liquidity_grab, "overlay": _run_overlay, "pair": _run_pair, "metric_only": _run_metric_only, "mix100": _run_mix100, "skip": _run_skip, "rsp_spy": _run_rsp_spy, "vol_fade": _run_vol_fade, } def build_registry() -> dict[str, IdeaSpec]: R: dict[str, IdeaSpec] = {} def add(i, name, runner, universe="mega5", params=None, skip_reason=None, slow=False): R[i] = IdeaSpec(i, name, runner, universe, params or {}, skip_reason, slow) # A · Slope add("01", "Bottom-10 slope short confirm4b", "slope", "sp100", {"mode": "confirm_short"}, slow=True) add("02", "Sector-neutral top-N", "slope", "sp100", {"mode": "confirm_long", "sector_cap": 2}, slow=True) add("03", "Score-proportional sizing", "slope", "sp100", {"mode": "confirm_long", "weight_mode": "score_prop"}, slow=True) add("04", "Mid-day re-rank 12:30", "slope", "sp100", {"mode": "confirm_long", "entry_bar": 36, "confirm_lag": 2}, slow=True) add("05", "EMA20/100 slower slope", "slope", "sp100", {"mode": "confirm_long", "fast_period": 20, "slow_period": 100}, slow=True) add("06", "ROC30 top-N", "slope", "sp100", {"mode": "roc_long"}, slow=True) add("07", "Alpha vs SPY residual mom", "slope", "sp100", {"mode": "alpha_long"}, slow=True) add("08", "Slope acceleration top-N", "slope", "sp100", {"mode": "accel_long"}, slow=True) add("09", "Satellite slope (same as confirm — note)", "slope", "sp100", {"mode": "confirm_long", "top_n": 5}, slow=True) add("10", "Afternoon fade of AM winners", "slope", "sp100", {"mode": "fade_winners"}, slow=True) add("11", "Overnight hold if slope+ (proxy: longer confirm)", "slope", "sp100", {"mode": "confirm_long", "confirm_lag": 2}, slow=True) add("12", "Daily slope gate proxy (VWAP req)", "slope", "sp100", {"mode": "confirm_long", "require_above_vwap": True}, slow=True) add("13", "Volume-confirm slope (above VWAP proxy)", "slope", "sp100", {"mode": "confirm_long", "require_above_vwap": True}, slow=True) add("14", "High-ADV subset", "slope", "sp100", {"mode": "confirm_long", "min_adv_usd": 5e7}, slow=True) add("15", "Gap-up weak slope short proxy", "slope", "sp100", {"mode": "confirm_short"}, slow=True) # B · Extremes add("16", "PDH touch long", "level", "mega5", {"level_key": "pdh", "side": "long"}) add("17", "PDH touch short", "level", "mega5", {"level_key": "pdh", "side": "short"}) add("18", "Week high short / week low long mix", "level", "mega5", {"level_key": "week_high", "side": "short"}) add("19", "OR high fade short", "level", "mega5", {"level_key": "or_high", "side": "short", "min_session_bar": 6}) add("20", "Prior-day mid short", "level", "mega5", {"level_key": "pdm", "side": "short"}) add("21", "Gap-fill fade", "gap", "mega5", {"mode": "fill", "min_gap": 0.005}) add("22", "Globex overnight levels", "skip", skip_reason="no Globex/futures data") add("23", "3d swing low short", "level", "mega5", {"level_key": "swing3_low", "side": "short"}) add("24", "Pivot R1 short", "level", "mega5", {"level_key": "pivot_r1", "side": "short"}) add("25", "Prior VWAP short", "level", "mega5", {"level_key": "prior_vwap", "side": "short"}) add("26", "IB high fade short", "level", "mega5", {"level_key": "ib_high", "side": "short", "min_session_bar": 12}) add("27", "Prior range mid short", "level", "mega5", {"level_key": "range_mid", "side": "short"}) add("28", "PDL short only above VWAP", "pdl", "mega5", {}) # overlay via level # better as level with above_vwap on pdl R["28"] = IdeaSpec("28", "PDL short only above VWAP", "level", "mega5", {"level_key": "pdl", "side": "short", "above_vwap": True, "use_support_intact": True}) add("29", "PDH false-break short", "level", "mega5", {"level_key": "pdh", "side": "short", "after_false_break": True, "first_touch_only": False}) add("30", "PDL + week low confluence", "level", "mega5", {"level_key": "pdl", "side": "short", "confluence_week": True, "use_support_intact": True}) add("31", "Third-touch PDL short", "level", "mega5", {"level_key": "pdl", "side": "short", "first_touch_only": False, "third_touch": True, "use_support_intact": True}) add("32", "Distant PDL short (−2ATR)", "level", "mega5", {"level_key": "pdl", "side": "short", "distant_atr_mult": 2.0, "use_support_intact": True}) # C · Breakouts add("33", "ATR downside breakout short", "atr_short", "mega5", {"atr_mult": 1.0, "max_entry_bar": 12}) add("34", "OR breakout long", "breakout", "mega5", {"level_key": "or_high", "side": "long", "min_entry_bar": 6, "max_entry_bar": 12}) add("35", "Donchian20 breakout long", "breakout", "mega5", {"level_key": "donchian20", "side": "long", "max_entry_bar": 48}) add("36", "NR7 then OR break", "breakout", "mega5", {"level_key": "or_high", "side": "long", "require_nr7": True, "min_entry_bar": 6, "max_entry_bar": 24}) add("37", "Idio breakout SPYOR high while SPY1%", "gap", "mega5", {"mode": "fill", "min_gap": 0.01}) add("62", "Gap-and-crap short", "gap", "mega5", {"mode": "gap_and_crap", "min_gap": 0.005}) add("63", "Earnings day-after straddle", "skip", skip_reason="options data required") add("64", "Index rebalance day", "skip", skip_reason="rebalance calendar not wired") add("65", "CPI range break", "skip", skip_reason="CPI calendar not wired") # G · Neutral add("66", "Dollar-neutral slope L/S", "slope", "sp100", {"mode": "dollar_neutral"}, slow=True) add("67", "Stock vs sector residual (pair proxy)", "pair", "mega5", {"a": "AAPL", "b": "MSFT"}) add("68", "AAPL/MSFT z-score pair", "pair", "mega5", {"a": "AAPL", "b": "MSFT"}) add("69", "Long PDL-short names / short PDH (mix ports)", "mix100", "mega5", {}) # approx add("70", "Residual ret−β SPY rank long", "slope", "sp100", {"mode": "alpha_long"}, slow=True) # H · Vol overlays add("71", "VIX>25 → PDL only", "overlay", "mega5", {"base": "pdl", "vix_min": 25.0}) add("72", "VIX<15 → ATR early only", "overlay", "mega5", {"base": "atr", "vix_max": 15.0}) add("73", "VVIX steep fade", "skip", skip_reason="VVIX not loaded") add("74", "VRP+ avoid fades (VIX>20 skip PDL)", "overlay", "mega5", {"base": "pdl", "vix_max": 20.0}) add("75", "Dynamic ATR% filter on PDL", "pdl", "mega5", {"min_prior_atr_pct": 2.0, "min_prior_vol_ann_pct": 25.0}) add("76", "Dispersion allocator proxy", "metric_only", "mega5", {}) # I · ETF add("77", "QQQ PDL touch short", "etf_pdl", "etfs", {"symbol": "QQQ"}) add("78", "Sector ETF top2/bottom2", "sector_rot", "etfs", {}) add("79", "TLT mean-revert day", "etf_mr", "etfs", {"symbol": "TLT"}) add("80", "VIXY/SVXY fade spike", "vol_fade", "etfs", {}) add("81", "IWM vs SPY relative breakout", "rel_bo", "etfs", {"long": "IWM", "vs": "SPY"}) add("82", "RSP vs SPY concentration fade", "rsp_spy", "etfs", {}) add("83", "Hierarchy filter (ATR when QQQ>open)", "atr_early", "mega5", {}) # J · Tape add("84", "Up/down volume imbalance (VWAP reject proxy)", "vwap_fail", "mega5", {"mode": "reject"}) add("85", "Range expand no volume fade (OR fade)", "level", "mega5", {"level_key": "or_high", "side": "short", "min_session_bar": 6}) add("86", "OR pierce reclaim long", "liq_grab", "mega5", {}) add("87", "Iceberg prints", "skip", skip_reason="no iceberg/L2 data") add("88", "Spread-widen halt (ATR early baseline)", "atr_early", "mega5", {}) # K · Options add("89", "Avoid shorts into vol crush", "skip", skip_reason="options IV data required") add("90", "Pin risk OPEX fade", "skip", skip_reason="options strike/OPEX data required") add("91", "PDL short + OTM call hedge", "skip", skip_reason="options required") # L · Meta add("92", "PDL when slope book weak (always PDL proxy)", "pdl", "mega5", {"min_prior_vol_ann_pct": 25.0}) add("93", "ATR sized by VIX (vix_max 22)", "atr_early", "mega5", {"vix_max_prior": 22.0}) add("94", "Slope only if AD rising (breadth proxy)", "breadth", "sp100", {}) add("95", "Corr spike keep PDL", "pdl", "mega5", {"min_prior_vol_ann_pct": 25.0, "min_prior_atr_pct": 2.0}) add("96", "Max1 overlap (top1 ATR)", "atr_early", "mega5", {"max_entries_per_bar": 1}) add("97", "Fixed-$ notional reporting hygiene", "metric_only", "mega5", {}) add("98", "Ban weak names (highvol filter)", "pdl", "mega5", {"min_prior_vol_ann_pct": 25.0, "min_prior_atr_pct": 1.75}) add("99", "Cash when RV extreme (VIX>35 flat)", "overlay", "mega5", {"base": "pdl", "vix_max": 35.0}) add("100", "4th sleeve mix 66+17+54", "mix100", "mega5", {}) assert len(R) == 100, len(R) return R IDEA_REGISTRY = build_registry() def run_idea(ctx: DataContext, spec: IdeaSpec) -> IdeaResult: if spec.skip_reason and spec.runner == "skip": return _run_skip(ctx, spec) fn = RUNNERS.get(spec.runner) if fn is None: return IdeaResult(idea_id=spec.idea_id, name=spec.name, status="error", skip_reason=f"no runner {spec.runner}") try: return fn(ctx, spec) except Exception as e: return IdeaResult(idea_id=spec.idea_id, name=spec.name, status="error", universe=spec.universe, skip_reason=str(e)[:300]) def run_all(ctx: DataContext, ids: list[str] | None = None, skip_slow: bool = False) -> list[IdeaResult]: specs = list(IDEA_REGISTRY.values()) if ids: want = {i.strip().zfill(2) if i.strip().isdigit() else i.strip() for i in ids} # also accept without zero pad want |= {i.lstrip("0") or "0" for i in want} specs = [s for s in specs if s.idea_id in want or s.idea_id.lstrip("0") in want] results = [] for spec in sorted(specs, key=lambda s: s.idea_id): if skip_slow and spec.slow: results.append(IdeaResult(idea_id=spec.idea_id, name=spec.name, status="skip", skip_reason="--skip-slow", universe=spec.universe)) print(f" {spec.idea_id} SKIP slow", flush=True) continue print(f" {spec.idea_id} {spec.name[:40]:40s} …", flush=True) r = run_idea(ctx, spec) print( f" → {r.status:5s} Sh={r.sharpe:6.2f} ret={r.total_return_pct:+8.1f}% n={r.n_trades:5d} avg={r.avg_pnl_pct:+.3f}% {r.skip_reason[:40]}", flush=True, ) results.append(r) return results