""" Intraday Cracking-Markets-style dip variants on 5-minute RTH bars. Variants -------- A. VWAP reclaim — stretch > k×ATR below VWAP, enter on close back above VWAP. B. Prior-day-low — limit at prior session low when name is above daily SMA(200). C. Relative washout — ≥2× typical 5m move vs session open and underperform SPY; cross-sectional top-N by ATR%. D. PDL bounce short — break below prior-day low, short failed retest (dead-cat / support→resistance). """ from __future__ import annotations from dataclasses import dataclass from enum import Enum from typing import Dict, Literal import numpy as np import pandas as pd from RenTech.strategy_stack.multi_strategy_manager import _align_panel_frames VariantName = Literal["vwap_reclaim", "pdl_touch", "relative_washout", "pdl_bounce_short", "pdl_touch_short"] ConfirmMode = Literal["none", "hammer", "vol_climax", "higher_low", "any"] class Variant(str, Enum): VWAP_RECLAIM = "vwap_reclaim" PDL_TOUCH = "pdl_touch" RELATIVE_WASHOUT = "relative_washout" PDL_BOUNCE_SHORT = "pdl_bounce_short" PDL_TOUCH_SHORT = "pdl_touch_short" @dataclass(frozen=True) class CmIntradayConfig: variant: VariantName = "vwap_reclaim" bar_minutes: int = 5 atr_period: int = 14 sma_trend: int = 200 # A: VWAP reclaim vwap_stretch_atr: float = 1.5 min_session_bars: int = 3 # skip first ~15 min for VWAP stability # B: prior-day-low long pdl_limit_buffer_atr: float = 0.0 # D: PDL breakdown + dead-cat short pdl_retest_buffer_atr: float = 0.15 # high within this × ATR of PDL counts as retest min_break_atr: float = 0.1 # session low must pierce PDL by this × ATR require_support_intact: bool = True # prior close held above PDL entering session require_spy_below_vwap: bool = False # optional short-favorable regime # C: relative washout move_mult: float = 2.0 rel_spy_min: float = 0.005 top_n: int = 10 limit_atr_mult: float = 0.5 # Shared exits profit_atr_mult: float = 0.5 stop_atr_mult: float = 0.5 exit_on_prior_high: bool = True min_bars_held_before_exit: int = 2 slippage_bps: float = 3.0 # Portfolio max_concurrent: int = 10 # --- Tweaks --- require_spy_above_vwap: bool = False vix_max_prior: float = 0.0 # 0 = off; else prior-session VIX must be below this confirm_mode: ConfirmMode = "none" use_daily_atr_exits: bool = False vol_climax_mult: float = 1.5 # Symbol eligibility (prior-day, no lookahead) min_prior_atr_pct: float = 0.0 # prior daily ATR(14) / close × 100; 0 = off min_prior_vol_ann_pct: float = 0.0 # prior 20d ann realized vol × 100; 0 = off vol_lookback: int = 20 @dataclass class RegimeContext: """Bar-aligned regime gates (same index as feature panels).""" spy_above_vwap: np.ndarray vix_ok: np.ndarray def baseline_config(variant: VariantName = "vwap_reclaim") -> CmIntradayConfig: return CmIntradayConfig(variant=variant) def tweaked_config(variant: VariantName = "vwap_reclaim") -> CmIntradayConfig: """Combined tweak preset from research notes.""" return CmIntradayConfig( variant=variant, vwap_stretch_atr=2.5, rel_spy_min=0.01, move_mult=2.5, profit_atr_mult=1.0, stop_atr_mult=0.75, use_daily_atr_exits=True, require_spy_above_vwap=True, vix_max_prior=20.0, confirm_mode="any", ) def pdl_short_config() -> CmIntradayConfig: """Short dead-cat bounce at prior-day low after breakdown (support → resistance).""" return CmIntradayConfig( variant="pdl_bounce_short", min_session_bars=6, profit_atr_mult=0.75, stop_atr_mult=0.5, use_daily_atr_exits=True, pdl_retest_buffer_atr=0.15, min_break_atr=0.1, require_support_intact=True, require_spy_below_vwap=False, min_bars_held_before_exit=1, exit_on_prior_high=False, ) def pdl_touch_short_config() -> CmIntradayConfig: """Mirror of PDL limit long: short at the touch (fade support hold).""" return CmIntradayConfig( variant="pdl_touch_short", profit_atr_mult=1.0, stop_atr_mult=0.75, use_daily_atr_exits=True, min_bars_held_before_exit=2, exit_on_prior_high=False, require_support_intact=True, ) def pdl_touch_short_highvol_config() -> CmIntradayConfig: """PDL touch short on higher-vol names only (drops low-ATR defensives).""" return CmIntradayConfig( variant="pdl_touch_short", profit_atr_mult=1.0, stop_atr_mult=0.75, use_daily_atr_exits=True, min_bars_held_before_exit=2, exit_on_prior_high=False, require_support_intact=True, min_prior_atr_pct=1.75, min_prior_vol_ann_pct=25.0, ) def pdl_touch_short_spy_below_vwap_config() -> CmIntradayConfig: """PDL touch short only when SPY is below session VWAP (idea 55).""" return CmIntradayConfig( variant="pdl_touch_short", profit_atr_mult=1.0, stop_atr_mult=0.75, use_daily_atr_exits=True, min_bars_held_before_exit=2, exit_on_prior_high=False, require_support_intact=True, require_spy_below_vwap=True, ) @dataclass class TradeRow: variant: str symbol: str session_date: str entry_time: str exit_time: str entry_price: float exit_price: float pnl_pct: float bars_held: int exit_reason: str def _session_key(index: pd.DatetimeIndex) -> pd.Series: idx = pd.to_datetime(index).tz_localize(None) return pd.Series(idx.normalize(), index=idx) def _session_bar_index(index: pd.DatetimeIndex) -> pd.Series: idx = pd.to_datetime(index).tz_localize(None) ses = _session_key(idx) out = np.zeros(len(idx), dtype=np.int64) prev = None n = 0 for i, s in enumerate(ses.values): if prev is None or s != prev: n = 0 prev = s else: n += 1 out[i] = n return pd.Series(out, index=idx) def _session_vwap(close: pd.Series, volume: pd.Series, index: pd.DatetimeIndex) -> pd.Series: ses = _session_key(index) v = volume.reindex(index).fillna(0.0).astype(np.float64) c = close.astype(np.float64) df = pd.DataFrame({"c": c.values, "v": v.values, "ses": ses.values}, index=index) df["pv"] = df["c"] * df["v"] df["cpv"] = df.groupby("ses")["pv"].cumsum() df["cv"] = df.groupby("ses")["v"].cumsum().replace(0, np.nan) return (df["cpv"] / df["cv"]).astype(np.float64) def _wilder_atr(high: pd.Series, low: pd.Series, close: pd.Series, period: int) -> pd.Series: h = high.astype(np.float64) l = low.astype(np.float64) c = close.astype(np.float64) prev_c = c.shift(1) tr = pd.concat([(h - l).abs(), (h - prev_c).abs(), (l - prev_c).abs()], axis=1).max(axis=1) return tr.ewm(alpha=1.0 / float(period), adjust=False, min_periods=period).mean() def _portfolio_open_count(state: np.ndarray, pending: np.ndarray | None = None) -> int: n = int(np.sum(state == 1)) if pending is not None: n += int(np.sum(pending)) return n def _can_exit(d: int, entry_bar: int, cfg: CmIntradayConfig) -> bool: return int(d - entry_bar) >= int(cfg.min_bars_held_before_exit) def _slip(px: float, bps: float, side: str) -> float: m = float(bps) / 10_000.0 if side == "buy": return px * (1.0 + m) if side == "sell": return px * (1.0 - m) raise ValueError(side) def _load_vix_prior_ok(master_index: pd.DatetimeIndex, max_vix: float) -> np.ndarray: import yfinance as yf ses = _session_key(master_index) start = pd.Timestamp(ses.min()) - pd.Timedelta(days=14) end = pd.Timestamp(ses.max()) + pd.Timedelta(days=2) raw = yf.download("^VIX", start=start.strftime("%Y-%m-%d"), end=end.strftime("%Y-%m-%d"), progress=False) if raw is None or raw.empty: return np.ones(len(master_index), dtype=bool) 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] col = "close" if "close" in raw.columns else raw.columns[0] vix = raw[col].astype(float) vix.index = pd.to_datetime(vix.index).tz_localize(None).normalize() prior = vix.shift(1) ok = (prior < float(max_vix)).astype(float) mapped = ses.map(ok).fillna(1.0).astype(bool).to_numpy() return mapped def build_regime_context( master_index: pd.DatetimeIndex, spy_intra: pd.DataFrame, cfg: CmIntradayConfig, ) -> RegimeContext: n = len(master_index) spy_ok = np.ones(n, dtype=bool) vix_ok = np.ones(n, dtype=bool) if (cfg.require_spy_above_vwap or cfg.require_spy_below_vwap) and spy_intra is not None and not spy_intra.empty: idx = pd.to_datetime(spy_intra.index).tz_localize(None) sc = spy_intra["close"].astype(np.float64) sc.index = idx vol = spy_intra["volume"].astype(np.float64) if "volume" in spy_intra.columns else pd.Series(1.0, index=idx) sv = _session_vwap(sc, vol, idx).reindex(master_index) spy_c = sc.reindex(master_index) spy_ok = (spy_c > sv).fillna(False).to_numpy(dtype=bool) if float(cfg.vix_max_prior) > 0: vix_ok = _load_vix_prior_ok(master_index, float(cfg.vix_max_prior)) return RegimeContext(spy_above_vwap=spy_ok, vix_ok=vix_ok) def _regime_allows(d: int, regime: RegimeContext | None, cfg: CmIntradayConfig) -> bool: if regime is None: return True if cfg.require_spy_above_vwap and not bool(regime.spy_above_vwap[d]): return False if cfg.require_spy_below_vwap and bool(regime.spy_above_vwap[d]): return False if float(cfg.vix_max_prior) > 0 and not bool(regime.vix_ok[d]): return False return True def _symbol_vol_allows(d: int, i: int, panels: dict[str, pd.DataFrame], cfg: CmIntradayConfig) -> bool: if float(cfg.min_prior_atr_pct) > 0: atr_pct = panels.get("prior_atr_pct") if atr_pct is None: return False v = float(atr_pct.to_numpy(dtype=np.float64)[d, i]) if not (np.isfinite(v) and v >= float(cfg.min_prior_atr_pct)): return False if float(cfg.min_prior_vol_ann_pct) > 0: vol_ann = panels.get("prior_vol_ann") if vol_ann is None: return False v = float(vol_ann.to_numpy(dtype=np.float64)[d, i]) if not (np.isfinite(v) and v >= float(cfg.min_prior_vol_ann_pct)): return False return True def _short_bar_return(d: int, fd: int, i: int, close: np.ndarray, entry_px: float) -> float: c_d = close[d, i] if not np.isfinite(c_d): return 0.0 if fd == d: return (entry_px - c_d) / entry_px if entry_px > 0 else 0.0 if d > 0 and np.isfinite(close[d - 1, i]) and close[d - 1, i] > 0: return (close[d - 1, i] - c_d) / close[d - 1, i] return 0.0 def _pdl_rejection_bar( d: int, i: int, *, open_: np.ndarray, high: np.ndarray, low: np.ndarray, close: np.ndarray, pdl: np.ndarray, atr: np.ndarray, cfg: CmIntradayConfig, ) -> bool: """Failed reclaim: retest PDL from below, close back under (support → resistance).""" p, a = pdl[d, i], atr[d, i] o, h, l, c = open_[d, i], high[d, i], low[d, i], close[d, i] if not all(np.isfinite(x) for x in (p, a, o, h, l, c)) or a <= 0: return False retest_hi = p - float(cfg.pdl_retest_buffer_atr) * a if h < retest_hi: return False if c >= p: return False # Shooting-star style rejection optional boost (upper wick) body = abs(c - o) upper_wick = h - max(o, c) return upper_wick >= body or c < o def _entry_confirmed( d: int, i: int, cfg: CmIntradayConfig, *, open_: np.ndarray, high: np.ndarray, low: np.ndarray, close: np.ndarray, volume: np.ndarray, ) -> bool: mode = cfg.confirm_mode if mode == "none": return True o, h, l, c = open_[d, i], high[d, i], low[d, i], close[d, i] if not all(np.isfinite(x) for x in (o, h, l, c)): return False body = abs(c - o) lower_wick = min(o, c) - l hammer = lower_wick > max(2.0 * body, 1e-9) and c >= o vol_ok = False if d >= 6 and np.isfinite(volume[d, i]): v_prev = volume[d - 5 : d, i] if v_prev.size > 0 and np.any(np.isfinite(v_prev)): v_avg = float(np.nanmean(v_prev)) if np.isfinite(v_avg) and v_avg > 0 and np.isfinite(close[d - 1, i]): vol_ok = volume[d, i] >= float(cfg.vol_climax_mult) * v_avg and c > close[d - 1, i] higher_low = False if d >= 1 and np.isfinite(close[d - 1, i]) and np.isfinite(low[d - 1, i]): higher_low = c > close[d - 1, i] and l > low[d - 1, i] if mode == "hammer": return bool(hammer) if mode == "vol_climax": return bool(vol_ok) if mode == "higher_low": return bool(higher_low) if mode == "any": return bool(hammer or vol_ok or higher_low) return True def _exit_atr(entry_atr: float, daily_atr: float, cfg: CmIntradayConfig) -> float: if cfg.use_daily_atr_exits and np.isfinite(daily_atr) and daily_atr > 0: return float(daily_atr) return float(entry_atr) def build_feature_panels( intra: Dict[str, pd.DataFrame], daily: Dict[str, pd.DataFrame], *, cfg: CmIntradayConfig | None = None, ) -> dict[str, pd.DataFrame]: """Aligned intraday panels + daily context (no lookahead on daily fields).""" close_p, open_p, high_p, low_p, vol_p, atr_p, vwap_p, datr_p = [], [], [], [], [], [], [], [] uptrend_p, pdl_p, ses_open_p, support_p = [], [], [], [] prior_atr_pct_p, prior_vol_ann_p = [], [] need_vol = cfg is not None and float(cfg.min_prior_vol_ann_pct) > 0 need_atr = cfg is not None and float(cfg.min_prior_atr_pct) > 0 vol_lb = int(cfg.vol_lookback) if cfg is not None else 20 for sym in sorted(intra.keys()): ib = intra[sym] db = daily.get(sym) if db is None or db.empty: continue idx = pd.to_datetime(ib.index).tz_localize(None) c = ib["close"].astype(np.float64) o = ib["open"].astype(np.float64) if "open" in ib.columns else c.copy() 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=idx) atr = _wilder_atr(h, l, c, 14) vwap = _session_vwap(c, v, idx) d_idx = pd.to_datetime(db.index).tz_localize(None) d_close = db["close"].astype(np.float64) d_high = db["high"].astype(np.float64) d_low = db["low"].astype(np.float64) sma = d_close.rolling(200, min_periods=200).mean() uptrend_d = (d_close.shift(1) > sma.shift(1)).astype(np.float64) pdl_d = d_low.shift(1) d_atr_d = _wilder_atr(d_high, d_low, d_close, 14).shift(1) support_d = (d_close.shift(1) > d_low.shift(1)).astype(np.float64) prior_atr_pct_d = (d_atr_d / d_close.shift(1) * 100.0) prior_vol_ann_d = ( d_close.pct_change().rolling(vol_lb, min_periods=vol_lb).std(ddof=1) * np.sqrt(252.0) * 100.0 ).shift(1) ses = _session_key(idx) uptrend = ses.map(uptrend_d).astype(np.float64) pdl = ses.map(pdl_d).astype(np.float64) d_atr = ses.map(d_atr_d).astype(np.float64) support_ok = ses.map(support_d).astype(np.float64) prior_atr_pct = ses.map(prior_atr_pct_d).astype(np.float64) prior_vol_ann = ses.map(prior_vol_ann_d).astype(np.float64) ses_open = c.groupby(ses).transform("first") close_p.append(c.rename(sym)) open_p.append(o.rename(sym)) high_p.append(h.rename(sym)) low_p.append(l.rename(sym)) vol_p.append(v.rename(sym)) atr_p.append(atr.rename(sym)) vwap_p.append(vwap.rename(sym)) datr_p.append(d_atr.rename(sym)) uptrend_p.append(uptrend.rename(sym)) pdl_p.append(pdl.rename(sym)) ses_open_p.append(ses_open.rename(sym)) support_p.append(support_ok.rename(sym)) if need_atr: prior_atr_pct_p.append(prior_atr_pct.rename(sym)) if need_vol: prior_vol_ann_p.append(prior_vol_ann.rename(sym)) if not close_p: raise ValueError("no symbols with intraday+daily data") close_df, cols = _align_panel_frames(close_p) open_df, _ = _align_panel_frames(open_p) high_df, _ = _align_panel_frames(high_p) low_df, _ = _align_panel_frames(low_p) vol_df, _ = _align_panel_frames(vol_p) atr_df, _ = _align_panel_frames(atr_p) vwap_df, _ = _align_panel_frames(vwap_p) datr_df, _ = _align_panel_frames(datr_p) uptrend_df, _ = _align_panel_frames(uptrend_p) pdl_df, _ = _align_panel_frames(pdl_p) ses_open_df, _ = _align_panel_frames(ses_open_p) support_df, _ = _align_panel_frames(support_p) out = { "close": close_df.reindex(columns=cols), "open": open_df.reindex(columns=cols), "high": high_df.reindex(columns=cols), "low": low_df.reindex(columns=cols), "volume": vol_df.reindex(columns=cols), "atr": atr_df.reindex(columns=cols), "daily_atr": datr_df.reindex(columns=cols), "vwap": vwap_df.reindex(columns=cols), "uptrend": uptrend_df.reindex(columns=cols), "pdl": pdl_df.reindex(columns=cols), "support_intact": support_df.reindex(columns=cols), "ses_open": ses_open_df.reindex(columns=cols), "columns": pd.Index(cols), } if need_atr: prior_atr_df, _ = _align_panel_frames(prior_atr_pct_p) out["prior_atr_pct"] = prior_atr_df.reindex(columns=cols) if need_vol: prior_vol_df, _ = _align_panel_frames(prior_vol_ann_p) out["prior_vol_ann"] = prior_vol_df.reindex(columns=cols) return out def _simulate_variant_a( panels: dict[str, pd.DataFrame], cfg: CmIntradayConfig, regime: RegimeContext | None = None, ) -> tuple[list[TradeRow], np.ndarray]: close = panels["close"].to_numpy(dtype=np.float64) open_ = panels["open"].to_numpy(dtype=np.float64) low = panels["low"].to_numpy(dtype=np.float64) high = panels["high"].to_numpy(dtype=np.float64) volume = panels["volume"].to_numpy(dtype=np.float64) atr = panels["atr"].to_numpy(dtype=np.float64) daily_atr = panels["daily_atr"].to_numpy(dtype=np.float64) vwap = panels["vwap"].to_numpy(dtype=np.float64) uptrend = panels["uptrend"].to_numpy(dtype=np.float64) idx = panels["close"].index cols = list(panels["columns"]) bar_in_ses = _session_bar_index(idx).to_numpy(dtype=np.int64) ses = _session_key(idx).to_numpy() t_rows, n_bars, n_sym = close.shape[0], close.shape[0], close.shape[1] port_r = np.zeros(t_rows, dtype=np.float64) trades: list[TradeRow] = [] state = np.zeros(n_sym, dtype=np.int8) # 0 flat, 1 in trade entry_px = np.full(n_sym, np.nan) entry_atr = 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 stretched = np.zeros(n_sym, dtype=bool) traded_session = np.full(n_sym, None, dtype=object) slip = float(cfg.slippage_bps) stretch_k = float(cfg.vwap_stretch_atr) prof_k = float(cfg.profit_atr_mult) stop_k = float(cfg.stop_atr_mult) max_conc = int(cfg.max_concurrent) prev_ses = None for d in range(t_rows): if prev_ses is None or ses[d] != prev_ses: state[:] = 0 stretched[:] = False traded_session[:] = None entry_px[:] = np.nan entry_atr[:] = np.nan entry_exit_atr[:] = np.nan entry_bar[:] = -1 prev_ses = ses[d] if bar_in_ses[d] < int(cfg.min_session_bars): continue # 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 = entry_px[i], entry_exit_atr[i] eb = int(entry_bar[i]) reason = None if _can_exit(d, eb, cfg): if np.isfinite(ea) and c_d >= ep + prof_k * ea: reason = "profit_atr" elif np.isfinite(ea) and c_d <= ep - stop_k * ea: reason = "stop_atr" elif cfg.exit_on_prior_high and d > eb: if np.isfinite(high[d - 1, i]) and c_d > high[d - 1, i]: reason = "prior_high" if d + 1 >= t_rows or ses[d + 1] != ses[d]: reason = reason or "session_close" if reason: xp = _slip(c_d, slip, "sell") trades.append( TradeRow( variant="vwap_reclaim", 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=float(xp / ep - 1.0), bars_held=int(d - entry_bar[i]), exit_reason=reason, ) ) state[i] = 0 # entries (rank by ATR% when slots limited) cand: list[tuple[int, float]] = [] for i in range(n_sym): if state[i] == 1 or traded_session[i] is not None: continue if uptrend[d, i] < 0.5: continue if not _regime_allows(d, regime, cfg): continue c_d, vw, a_d = close[d, i], vwap[d, i], atr[d, i] if not (np.isfinite(c_d) and np.isfinite(vw) and np.isfinite(a_d) and a_d > 0): continue dist = vw - c_d if dist > stretch_k * a_d: stretched[i] = True if stretched[i] and c_d > vw: if not _entry_confirmed( d, i, cfg, open_=open_, high=high, low=low, close=close, volume=volume ): continue cand.append((i, a_d / c_d)) cand.sort(key=lambda x: -x[1]) slots = max(0, max_conc - _portfolio_open_count(state)) for i, _ in cand[:slots]: c_d, a_d = close[d, i], atr[d, i] ep = _slip(c_d, slip, "buy") state[i] = 1 entry_px[i] = ep entry_atr[i] = a_d entry_exit_atr[i] = _exit_atr(a_d, daily_atr[d, i], cfg) entry_bar[i] = d entry_time[i] = str(idx[d]) traded_session[i] = ses[d] # bar return for open positions held = state == 1 if np.any(held): rets = np.zeros(n_sym, dtype=np.float64) 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 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 _simulate_variant_b( panels: dict[str, pd.DataFrame], cfg: CmIntradayConfig, regime: RegimeContext | None = None, ) -> tuple[list[TradeRow], np.ndarray]: close = panels["close"].to_numpy(dtype=np.float64) open_ = panels["open"].to_numpy(dtype=np.float64) low = panels["low"].to_numpy(dtype=np.float64) high = panels["high"].to_numpy(dtype=np.float64) volume = panels["volume"].to_numpy(dtype=np.float64) atr = panels["atr"].to_numpy(dtype=np.float64) daily_atr = panels["daily_atr"].to_numpy(dtype=np.float64) uptrend = panels["uptrend"].to_numpy(dtype=np.float64) pdl = panels["pdl"].to_numpy(dtype=np.float64) idx = panels["close"].index cols = list(panels["columns"]) bar_in_ses = _session_bar_index(idx).to_numpy(dtype=np.int64) ses = _session_key(idx).to_numpy() t_rows, n_sym = close.shape[0], close.shape[1] 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_atr = 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 session_high_at_entry = np.full(n_sym, np.nan) traded_session = np.full(n_sym, None, dtype=object) slip = float(cfg.slippage_bps) prof_k = float(cfg.profit_atr_mult) stop_k = float(cfg.stop_atr_mult) max_conc = int(cfg.max_concurrent) session_high = np.full(n_sym, np.nan) 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_atr[:] = np.nan entry_exit_atr[:] = np.nan entry_bar[:] = -1 session_high[:] = np.nan session_high_at_entry[:] = np.nan 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]) # 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 = entry_px[i], entry_exit_atr[i] eb = int(entry_bar[i]) reason = None if _can_exit(d, eb, cfg): if np.isfinite(ea) and c_d >= ep + prof_k * ea: reason = "profit_atr" elif np.isfinite(session_high_at_entry[i]) and c_d >= session_high_at_entry[i]: reason = "session_high" elif np.isfinite(pdl[d, i]) and np.isfinite(ea) and c_d <= ep - stop_k * ea: reason = "stop_atr" if d + 1 >= t_rows or ses[d + 1] != ses[d]: reason = reason or "session_close" if reason: xp = _slip(c_d, slip, "sell") trades.append( TradeRow( variant="pdl_touch", 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=float(xp / ep - 1.0), bars_held=int(d - entry_bar[i]), exit_reason=reason, ) ) state[i] = 0 # limit fill at PDL (rank by ATR% when slots limited) fill_cand: list[tuple[int, float]] = [] for i in range(n_sym): if state[i] == 1 or traded_session[i] is not None: continue if uptrend[d, i] < 0.5: continue if not _regime_allows(d, regime, cfg): continue lim = pdl[d, i] a_d = atr[d, i] c_d = close[d, i] if not (np.isfinite(lim) and np.isfinite(a_d) and np.isfinite(low[d, i]) and np.isfinite(c_d)): continue if low[d, i] <= lim: if not _entry_confirmed( d, i, cfg, open_=open_, high=high, low=low, close=close, volume=volume ): continue fill_cand.append((i, a_d / c_d)) fill_cand.sort(key=lambda x: -x[1]) slots = max(0, max_conc - _portfolio_open_count(state)) for i, _ in fill_cand[:slots]: lim = pdl[d, i] a_d = atr[d, i] ep = _slip(lim, slip, "buy") state[i] = 1 entry_px[i] = ep entry_atr[i] = a_d entry_exit_atr[i] = _exit_atr(a_d, daily_atr[d, i], cfg) entry_bar[i] = d entry_time[i] = str(idx[d]) session_high_at_entry[i] = session_high[i] 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): 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 _simulate_variant_pdl_touch_short( panels: dict[str, pd.DataFrame], cfg: CmIntradayConfig, regime: RegimeContext | None = None, ) -> tuple[list[TradeRow], np.ndarray]: """Short limit at PDL when price touches — mirror of ``pdl_touch`` long.""" close = panels["close"].to_numpy(dtype=np.float64) low = panels["low"].to_numpy(dtype=np.float64) high = panels["high"].to_numpy(dtype=np.float64) atr = panels["atr"].to_numpy(dtype=np.float64) daily_atr = panels["daily_atr"].to_numpy(dtype=np.float64) pdl = panels["pdl"].to_numpy(dtype=np.float64) support = panels["support_intact"].to_numpy(dtype=np.float64) idx = panels["close"].index cols = list(panels["columns"]) ses = _session_key(idx).to_numpy() t_rows, n_sym = close.shape[0], close.shape[1] 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 session_high_at_entry = np.full(n_sym, np.nan) traded_session = np.full(n_sym, None, dtype=object) session_high = np.full(n_sym, np.nan) slip = float(cfg.slippage_bps) prof_k = float(cfg.profit_atr_mult) stop_k = float(cfg.stop_atr_mult) max_conc = int(cfg.max_concurrent) 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 session_high[:] = np.nan session_high_at_entry[:] = np.nan 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]) 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 = entry_px[i], entry_exit_atr[i] eb = int(entry_bar[i]) reason = None if _can_exit(d, eb, cfg): if np.isfinite(ea) and c_d <= ep - prof_k * ea: reason = "profit_atr" elif np.isfinite(session_high_at_entry[i]) and c_d >= session_high_at_entry[i]: reason = "session_high" elif np.isfinite(ea) and c_d >= ep + stop_k * ea: reason = "stop_atr" if d + 1 >= t_rows or ses[d + 1] != ses[d]: reason = reason or "session_close" if reason: xp = _slip(c_d, slip, "buy") trades.append( TradeRow( variant="pdl_touch_short", 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=float((ep - xp) / ep) if ep > 0 else 0.0, bars_held=int(d - entry_bar[i]), exit_reason=reason, ) ) state[i] = 0 fill_cand: list[tuple[int, float]] = [] for i in range(n_sym): if state[i] == 1 or traded_session[i] is not None: continue if cfg.require_support_intact and support[d, i] < 0.5: continue if not _symbol_vol_allows(d, i, panels, cfg): continue if not _regime_allows(d, regime, cfg): continue lim = pdl[d, i] a_d = atr[d, i] c_d = close[d, i] if not (np.isfinite(lim) and np.isfinite(a_d) and np.isfinite(low[d, i]) and np.isfinite(c_d)): continue if low[d, i] <= lim: fill_cand.append((i, a_d / c_d)) fill_cand.sort(key=lambda x: -x[1]) slots = max(0, max_conc - _portfolio_open_count(state)) for i, _ in fill_cand[:slots]: lim = pdl[d, i] a_d = atr[d, i] ep = _slip(lim, slip, "sell") state[i] = 1 entry_px[i] = ep entry_exit_atr[i] = _exit_atr(a_d, daily_atr[d, i], cfg) entry_bar[i] = d entry_time[i] = str(idx[d]) session_high_at_entry[i] = session_high[i] 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): 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): w = 1.0 / float(np.sum(active)) port_r[d] = float(np.sum(rets[active] * w)) return trades, port_r def _simulate_variant_c( panels: dict[str, pd.DataFrame], spy_close: pd.Series, spy_open: pd.Series, cfg: CmIntradayConfig, regime: RegimeContext | None = None, ) -> tuple[list[TradeRow], np.ndarray]: close = panels["close"].to_numpy(dtype=np.float64) open_ = panels["open"].to_numpy(dtype=np.float64) low = panels["low"].to_numpy(dtype=np.float64) high = panels["high"].to_numpy(dtype=np.float64) volume = panels["volume"].to_numpy(dtype=np.float64) atr = panels["atr"].to_numpy(dtype=np.float64) daily_atr = panels["daily_atr"].to_numpy(dtype=np.float64) ses_open = panels["ses_open"].to_numpy(dtype=np.float64) idx = panels["close"].index cols = list(panels["columns"]) bar_in_ses = _session_bar_index(idx).to_numpy(dtype=np.int64) ses = _session_key(idx).to_numpy() spy_c = spy_close.reindex(idx).astype(np.float64).to_numpy() spy_o = spy_open.reindex(idx).astype(np.float64).to_numpy() t_rows, n_sym = close.shape[0], close.shape[1] port_r = np.zeros(t_rows, dtype=np.float64) trades: list[TradeRow] = [] state = np.zeros(n_sym, dtype=np.int8) pending = np.zeros(n_sym, dtype=bool) pending_limit = np.full(n_sym, np.nan) pending_atr = np.full(n_sym, np.nan) pending_exit_atr = np.full(n_sym, np.nan) pending_from = np.full(n_sym, -1, dtype=np.int32) entry_px = np.full(n_sym, np.nan) entry_atr = 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) slip = float(cfg.slippage_bps) move_mult = float(cfg.move_mult) rel_min = float(cfg.rel_spy_min) lim_mult = float(cfg.limit_atr_mult) prof_k = float(cfg.profit_atr_mult) book_n = min(int(cfg.top_n), int(cfg.max_concurrent)) prev_ses = None for d in range(t_rows): if prev_ses is None or ses[d] != prev_ses: pending[:] = False traded_session[:] = None prev_ses = ses[d] # drop stale pending stale = pending & (pending_from < d - 1) pending[stale] = False # fills from prior bar signal if d >= 1: fill_mask = pending & (pending_from == d - 1) for i in np.flatnonzero(fill_mask): lim = pending_limit[i] if ( np.isfinite(lim) and np.isfinite(low[d, i]) and low[d, i] <= lim and state[i] == 0 and traded_session[i] is None ): state[i] = 1 entry_px[i] = _slip(lim, slip, "buy") entry_atr[i] = pending_atr[i] entry_exit_atr[i] = pending_exit_atr[i] entry_bar[i] = d entry_time[i] = str(idx[d]) traded_session[i] = ses[d] pending[i] = False # 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 = entry_px[i], entry_exit_atr[i] eb = int(entry_bar[i]) reason = None if _can_exit(d, eb, cfg): if np.isfinite(ea) and c_d >= ep + prof_k * ea: reason = "profit_atr" elif cfg.exit_on_prior_high and d > eb: if np.isfinite(high[d - 1, i]) and c_d > high[d - 1, i]: reason = "prior_high" if d + 1 >= t_rows or ses[d + 1] != ses[d]: reason = reason or "session_close" if reason: xp = _slip(c_d, slip, "sell") trades.append( TradeRow( variant="relative_washout", 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=float(xp / ep - 1.0), bars_held=int(d - entry_bar[i]), exit_reason=reason, ) ) state[i] = 0 # new signals (after OR) if ( bar_in_ses[d] >= 6 and d + 1 < t_rows and ses[d + 1] == ses[d] and _regime_allows(d, regime, cfg) ): slots = max(0, book_n - _portfolio_open_count(state, pending)) if slots > 0: spy_ret = spy_c[d] / spy_o[d] - 1.0 if spy_o[d] > 0 and np.isfinite(spy_c[d]) else np.nan cand = [] for i in range(n_sym): if state[i] == 1 or pending[i] or traded_session[i] is not None: continue c_d, o_d, a_d = close[d, i], ses_open[d, i], atr[d, i] if not (np.isfinite(c_d) and np.isfinite(o_d) and o_d > 0 and np.isfinite(a_d) and a_d > 0): continue stock_ret = c_d / o_d - 1.0 drop_from_open = o_d - c_d if drop_from_open < move_mult * a_d: continue if not np.isfinite(spy_ret) or stock_ret - spy_ret > -rel_min: continue if not _entry_confirmed( d, i, cfg, open_=open_, high=high, low=low, close=close, volume=volume ): continue atr_pct = a_d / c_d cand.append((i, atr_pct)) cand.sort(key=lambda x: -x[1]) for i, _ in cand[:slots]: cl = close[d, i] a_v = atr[d, i] pending[i] = True pending_from[i] = d pending_limit[i] = cl - lim_mult * a_v pending_atr[i] = a_v pending_exit_atr[i] = _exit_atr(a_v, daily_atr[d, i], cfg) held = state == 1 if np.any(held): rets = np.zeros(n_sym, dtype=np.float64) 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 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 _simulate_variant_pdl_short( panels: dict[str, pd.DataFrame], cfg: CmIntradayConfig, regime: RegimeContext | None = None, ) -> tuple[list[TradeRow], np.ndarray]: """ Support breakdown + dead-cat short: 1. Session low breaks below prior-day low (support lost) 2. Price retests PDL from below and rejects (close < PDL) 3. Short; cover on ATR profit, PDL reclaim stop, or MOC """ close = panels["close"].to_numpy(dtype=np.float64) open_ = panels["open"].to_numpy(dtype=np.float64) low = panels["low"].to_numpy(dtype=np.float64) high = panels["high"].to_numpy(dtype=np.float64) atr = panels["atr"].to_numpy(dtype=np.float64) daily_atr = panels["daily_atr"].to_numpy(dtype=np.float64) pdl = panels["pdl"].to_numpy(dtype=np.float64) support = panels["support_intact"].to_numpy(dtype=np.float64) idx = panels["close"].index cols = list(panels["columns"]) bar_in_ses = _session_bar_index(idx).to_numpy(dtype=np.int64) ses = _session_key(idx).to_numpy() t_rows, n_sym = close.shape[0], close.shape[1] 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_pdl = 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) breakdown_seen = np.zeros(n_sym, dtype=bool) retest_used = np.zeros(n_sym, dtype=bool) session_low = np.full(n_sym, np.nan) slip = float(cfg.slippage_bps) prof_k = float(cfg.profit_atr_mult) stop_k = float(cfg.stop_atr_mult) break_k = float(cfg.min_break_atr) max_conc = int(cfg.max_concurrent) prev_ses = None for d in range(t_rows): if prev_ses is None or ses[d] != prev_ses: state[:] = 0 traded_session[:] = None breakdown_seen[:] = False retest_used[:] = False session_low[:] = np.nan entry_px[:] = np.nan entry_exit_atr[:] = np.nan entry_pdl[:] = np.nan entry_bar[:] = -1 prev_ses = ses[d] if bar_in_ses[d] < int(cfg.min_session_bars): continue for i in range(n_sym): 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]) p_i, a_i = pdl[d, i], atr[d, i] if ( not breakdown_seen[i] and np.isfinite(session_low[i]) and np.isfinite(p_i) and np.isfinite(a_i) and session_low[i] <= p_i - break_k * a_i ): breakdown_seen[i] = True # cover shorts 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, p_ref = entry_px[i], entry_exit_atr[i], entry_pdl[i] eb = int(entry_bar[i]) reason = None if _can_exit(d, eb, cfg): if np.isfinite(ea) and c_d <= ep - prof_k * ea: reason = "profit_atr" elif np.isfinite(p_ref) and np.isfinite(ea) and c_d >= p_ref + stop_k * ea: reason = "pdl_reclaim" elif np.isfinite(ea) and c_d >= ep + stop_k * ea: reason = "stop_atr" if d + 1 >= t_rows or ses[d + 1] != ses[d]: reason = reason or "session_close" if reason: xp = _slip(c_d, slip, "buy") # cover trades.append( TradeRow( variant="pdl_bounce_short", 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=float((ep - xp) / ep) if ep > 0 else 0.0, bars_held=int(d - entry_bar[i]), exit_reason=reason, ) ) state[i] = 0 # short entries on failed PDL retest cand: list[tuple[int, float]] = [] for i in range(n_sym): if state[i] == 1 or traded_session[i] is not None or retest_used[i]: continue if not breakdown_seen[i]: continue if cfg.require_support_intact and support[d, i] < 0.5: continue if not _regime_allows(d, regime, cfg): continue if not _pdl_rejection_bar( d, i, open_=open_, high=high, low=low, close=close, pdl=pdl, atr=atr, cfg=cfg ): continue a_d, c_d = atr[d, i], close[d, i] if np.isfinite(a_d) and np.isfinite(c_d) and c_d > 0: cand.append((i, a_d / c_d)) cand.sort(key=lambda x: -x[1]) slots = max(0, max_conc - _portfolio_open_count(state)) for i, _ in cand[:slots]: c_d = close[d, i] a_d = atr[d, i] ep = _slip(c_d, slip, "sell") state[i] = 1 entry_px[i] = ep entry_exit_atr[i] = _exit_atr(a_d, daily_atr[d, i], cfg) entry_pdl[i] = pdl[d, i] entry_bar[i] = d entry_time[i] = str(idx[d]) traded_session[i] = ses[d] retest_used[i] = True held = state == 1 if np.any(held): rets = np.zeros(n_sym, dtype=np.float64) 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): w = 1.0 / float(np.sum(active)) port_r[d] = float(np.sum(rets[active] * w)) return trades, port_r def run_backtest( intra: Dict[str, pd.DataFrame], daily: Dict[str, pd.DataFrame], *, cfg: CmIntradayConfig, spy_intra: pd.DataFrame | None = None, return_start: pd.Timestamp | None = None, ) -> tuple[pd.Series, pd.DataFrame]: panels = build_feature_panels(intra, daily, cfg=cfg) master = panels["close"].index regime = build_regime_context(master, spy_intra, cfg) if spy_intra is not None else None if cfg.variant == "vwap_reclaim": trades, port_r = _simulate_variant_a(panels, cfg, regime=regime) elif cfg.variant == "pdl_touch": trades, port_r = _simulate_variant_b(panels, cfg, regime=regime) elif cfg.variant == "relative_washout": if spy_intra is None or spy_intra.empty: raise ValueError("relative_washout requires spy_intra bars") spy_close = spy_intra["close"].astype(np.float64) spy_close.index = pd.to_datetime(spy_close.index).tz_localize(None) spy_open = spy_close.groupby(_session_key(spy_close.index)).transform("first") trades, port_r = _simulate_variant_c(panels, spy_close, spy_open, cfg, regime=regime) elif cfg.variant == "pdl_bounce_short": trades, port_r = _simulate_variant_pdl_short(panels, cfg, regime=regime) elif cfg.variant == "pdl_touch_short": trades, port_r = _simulate_variant_pdl_touch_short(panels, cfg, regime=regime) else: raise ValueError(f"unknown variant {cfg.variant}") port = pd.Series(port_r, index=master, dtype=np.float64) if return_start is not None: port = port.loc[port.index >= return_start] tdf = pd.DataFrame([t.__dict__ for t in trades]) return port, tdf def trade_stats(tdf: pd.DataFrame) -> dict: if tdf is None or tdf.empty: return {"n_trades": 0, "win_rate_pct": float("nan"), "avg_pnl_pct": float("nan"), "avg_bars_held": float("nan")} wins = (tdf["pnl_pct"] > 0).sum() losses = (tdf["pnl_pct"] < 0).sum() denom = wins + losses return { "n_trades": int(len(tdf)), "win_rate_pct": float(100.0 * wins / denom) if denom else float("nan"), "avg_pnl_pct": float(tdf["pnl_pct"].mean() * 100), "avg_bars_held": float(tdf["bars_held"].mean()), "median_pnl_pct": float(tdf["pnl_pct"].median() * 100), } def portfolio_metrics(port_r: pd.Series) -> dict: from RenTech.strategy_stack.alpaca_minute_loader import compound_intraday_to_daily ds = compound_intraday_to_daily(port_r).dropna() if len(ds) < 2: return {} eq = (1 + ds).cumprod() years = len(ds) / 252.0 sd = float(ds.std(ddof=1)) return { "total_return_pct": float((eq.iloc[-1] - 1) * 100), "cagr_pct": float((eq.iloc[-1] ** (1 / years) - 1) * 100) if years > 0 else float("nan"), "max_dd_pct": float((eq / eq.cummax() - 1).min() * 100), "sharpe": float(ds.mean() / sd * np.sqrt(252)) if sd > 1e-12 else float("nan"), "n_sessions": int(len(ds)), }