diff --git a/apps/intraday_bt/scripts/diag_orb_sector_etf.py b/apps/intraday_bt/scripts/diag_orb_sector_etf.py new file mode 100644 index 0000000..542106a --- /dev/null +++ b/apps/intraday_bt/scripts/diag_orb_sector_etf.py @@ -0,0 +1,440 @@ +""" +V45 Diagnostic: Sector ETF Gap Signal + +Hypothesis: On days when the stock's sector ETF (XLK, XLF, etc.) itself has a positive +morning gap (open > prior close), individual ORB breakouts in that sector have stronger +follow-through. The logic: if XLK gaps up while NVDA gaps up, it's a sector-wide +institutional move with more follow-through than an idiosyncratic NVDA gap. + +Features tested: + sector_etf_gap_pct : (sector_etf_open - sector_etf_prev_close) / sector_etf_prev_close + sector_etf_orb_return : sector ETF return during its own 5-min ORB window + sector_minus_qqq_gap : sector_etf_gap_pct - qqq_gap_pct (sector-relative strength) + +Source: V24 400d run JSON + fetched sector ETF parquet data. + +Sector → ETF mapping (SPDR Select Sectors): + Technology → XLK + Financial Services → XLF + Healthcare → XLV + Energy → XLE + Industrials → XLI + Basic Materials → XLB + Communication Services → XLC + Utilities → XLU + Real Estate → XLRE + Consumer Defensive → XLP + Consumer Cyclical → XLY +""" +from __future__ import annotations + +import concurrent.futures +import datetime as dt +import json +import os +import sys +from pathlib import Path + +import pyarrow.parquet as pq + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))) + +from zoneinfo import ZoneInfo +from libs.common.time_utils import trading_days_between + +_ET = ZoneInfo("America/New_York") +_MKT_OPEN = dt.time(9, 30) +_ORB_END = dt.time(9, 35) # 5-min ORB end +_MKT_CLOSE = dt.time(16, 0) + +INTRADAY_CACHE_DIR = "data/cache/intraday" +V24_400D_RUN = "runs/intraday_orb/intraday_20260422_011012_06f59ede.json" +SECTOR_CACHE = "data/cache/sector_cache.json" + +SECTOR_TO_ETF: dict[str, str] = { + "Technology": "XLK", + "Financial Services": "XLF", + "Healthcare": "XLV", + "Energy": "XLE", + "Industrials": "XLI", + "Basic Materials": "XLB", + "Communication Services": "XLC", + "Utilities": "XLU", + "Real Estate": "XLRE", + "Consumer Defensive": "XLP", + "Consumer Cyclical": "XLY", +} + +# Also track QQQ for sector-relative comparison +ETF_TICKERS = list(set(SECTOR_TO_ETF.values())) + ["QQQ"] + + +def _parse_ts(ts_raw: object) -> dt.datetime: + s = str(ts_raw) + if s.endswith("Z"): + s = s[:-1] + "+00:00" + return dt.datetime.fromisoformat(s).astimezone(_ET) + + +def fetch_and_cache_etf_bars(tickers: list[str], start_date: str, end_date: str) -> None: + """Fetch intraday bars for sector ETFs and save to parquet cache.""" + import httpx + import pyarrow as pa + + base_url = "http://localhost:18001" + + def _fetch_one(ticker: str) -> tuple[str, list[dict]]: + url = base_url + "/api/v1/alpaca/intraday" + try: + resp = httpx.get( + url, + params={"tickers": ticker, "interval": "5min", + "start_date": start_date, "end_date": end_date}, + timeout=120.0, + ) + resp.raise_for_status() + return ticker, resp.json().get("bars", {}).get(ticker, []) + except Exception as exc: + print(f" WARNING: failed to fetch {ticker}: {exc}") + return ticker, [] + + bars_by_ticker: dict[str, list[dict]] = {} + for ticker in tickers: + print(f" Fetching {ticker}...", end=" ", flush=True) + _, bars = _fetch_one(ticker) + if bars: + bars_by_ticker[ticker] = bars + print(f"{len(bars)} bars") + else: + print("FAILED") + + schema = pa.schema( + [ + ("timestamp", pa.string()), + ("open", pa.float64()), + ("high", pa.float64()), + ("low", pa.float64()), + ("close", pa.float64()), + ("volume", pa.int64()), + ] + ) + + for ticker, bars in bars_by_ticker.items(): + # Group bars by date + by_date: dict[str, list[dict]] = {} + for bar in bars: + try: + ts = _parse_ts(bar["timestamp"]) + except Exception: + continue + date_str = ts.date().isoformat() + by_date.setdefault(date_str, []).append(bar) + + saved = 0 + for date_str, day_bars in by_date.items(): + path = Path(INTRADAY_CACHE_DIR) / ticker.upper() / f"{date_str}.parquet" + if path.exists(): + continue + path.parent.mkdir(parents=True, exist_ok=True) + rows = { + "timestamp": [], + "open": [], + "high": [], + "low": [], + "close": [], + "volume": [], + } + for b in day_bars: + rows["timestamp"].append(str(b.get("timestamp", ""))) + rows["open"].append(float(b.get("open", 0) or 0)) + rows["high"].append(float(b.get("high", 0) or 0)) + rows["low"].append(float(b.get("low", 0) or 0)) + rows["close"].append(float(b.get("close", 0) or 0)) + rows["volume"].append(int(b.get("volume", 0) or 0)) + table = pa.table(rows, schema=schema) + pq.write_table(table, str(path)) + saved += 1 + print(f" {ticker}: {saved} new parquet files saved") + + missing = [t for t in tickers if t not in bars_by_ticker] + if missing: + print(f" WARNING: no data returned for {missing}") + + +def load_etf_daily_ohlcv(ticker: str, dates: list[str]) -> dict[str, dict]: + """Load OHLCV data for each trading day. Returns {date: {open, close, orb_high, orb_return}}.""" + result = {} + for date in dates: + path = Path(INTRADAY_CACHE_DIR) / ticker.upper() / f"{date}.parquet" + if not path.exists(): + continue + try: + table = pq.read_table(str(path)) + rows = table.to_pydict() + except Exception: + continue + + # Parse all bars, collect opens/closes and ORB bar + mkt_bars = [] + for i, ts_raw in enumerate(rows.get("timestamp", [])): + try: + ts = _parse_ts(ts_raw) + except Exception: + continue + if not (_MKT_OPEN <= ts.time() < _MKT_CLOSE): + continue + mkt_bars.append( + { + "ts": ts, + "open": float(rows["open"][i] or 0), + "high": float(rows["high"][i] or 0), + "low": float(rows["low"][i] or 0), + "close": float(rows["close"][i] or 0), + } + ) + + if not mkt_bars: + continue + mkt_bars.sort(key=lambda b: b["ts"]) + + day_open = mkt_bars[0]["open"] + day_close = mkt_bars[-1]["close"] + + # 5-min ORB bar (9:30-9:35): first bar + orb_bar = mkt_bars[0] + orb_high = orb_bar["high"] + orb_close = orb_bar["close"] + orb_open = orb_bar["open"] + orb_return = (orb_close - orb_open) / orb_open if orb_open > 0 else None + + result[date] = { + "open": day_open, + "close": day_close, + "orb_high": orb_high, + "orb_return": orb_return, + } + return result + + +def pearson(xs: list[float], ys: list[float]) -> float | None: + n = len(xs) + if n < 2: + return None + xm, ym = sum(xs) / n, sum(ys) / n + num = sum((xs[i] - xm) * (ys[i] - ym) for i in range(n)) + dx = sum((x - xm) ** 2 for x in xs) ** 0.5 + dy = sum((y - ym) ** 2 for y in ys) ** 0.5 + if dx <= 0 or dy <= 0: + return None + return num / (dx * dy) + + +def tercile_stats(vals: list[float], rs: list[float]) -> dict: + if len(vals) < 9: + return {} + pairs = sorted(zip(vals, rs), key=lambda p: p[0]) + n = len(pairs) + t = n // 3 + def stats(sub): + ys = [p[1] for p in sub] + return {"n": len(ys), "wr": sum(1 for y in ys if y > 0) / len(ys), "avg_r": sum(ys) / len(ys)} + return {"low": stats(pairs[:t]), "mid": stats(pairs[t:2*t]), "high": stats(pairs[2*t:])} + + +def report_feature(label: str, vals: list[float], rs: list[float], obv_vals: list[float] | None = None) -> None: + n = len(vals) + p = pearson(vals, rs) + ts = tercile_stats(vals, rs) + if not ts or p is None: + print(f" {label}: n={n}, insufficient data") + return + low, mid, high = ts["low"], ts["mid"], ts["high"] + avg_r_gap = abs(high["avg_r"] - low["avg_r"]) + wr_gap = abs(high["wr"] - low["wr"]) + g1 = abs(p) >= 0.07 and n >= 120 + g2 = avg_r_gap >= 0.30 + g3 = wr_gap >= 0.05 + + # G5a: redundancy with OBV slope + g5a = True + rho_obv = None + if obv_vals is not None and len(obv_vals) == n: + rho_obv = pearson(vals, obv_vals) + if rho_obv is not None: + g5a = abs(rho_obv) < 0.70 + + print(f"\n [{label}] n={n} Pearson={p:+.3f}") + print(f" G1: {'PASS' if g1 else 'FAIL'} (|{abs(p):.3f}| {'≥' if abs(p)>=0.07 else '<'} 0.07, n={n})") + print(f" G2: {'PASS' if g2 else 'FAIL'} (avg_R gap = {avg_r_gap:.3f}R [threshold 0.30R])") + print(f" G3: {'PASS' if g3 else 'FAIL'} (WR gap = {wr_gap*100:.1f}pp [threshold 5pp])") + if rho_obv is not None: + print(f" G5a: {'PASS' if g5a else 'FAIL'} (ρ(feature,OBV-slope)={rho_obv:+.3f} [threshold |<0.70|])") + print(f" Bottom tercile: WR {low['wr']*100:.1f}% avg_R {low['avg_r']:+.3f} n={low['n']}") + print(f" Middle tercile: WR {mid['wr']*100:.1f}% avg_R {mid['avg_r']:+.3f} n={mid['n']}") + print(f" Top tercile: WR {high['wr']*100:.1f}% avg_R {high['avg_r']:+.3f} n={high['n']}") + all_gates = g1 and g2 and g3 and g5a + print(f" → {'ALL GATES PASS ✓' if all_gates else 'FAIL'}") + + +def main() -> None: + print("=== V45 Sector ETF Gap Signal Diagnostic ===\n") + + # Load V24 400d run + with open(V24_400D_RUN) as f: + run_data = json.load(f) + trades = run_data.get("trades", []) + m = run_data.get("metrics", {}) + print(f"Loaded V24 400d: {m.get('total_trades')} trades, {m.get('start_date')} → {m.get('end_date')}") + + # Load sector cache + with open(SECTOR_CACHE) as f: + sector_cache = json.load(f) + + trade_records = [ + { + "ticker": t["ticker"], + "date": t["date"][:10], + "r_multiple": float(t["r_multiple_at_exit"]), + "sector": sector_cache.get(t["ticker"], "UNKNOWN"), + } + for t in trades + if t.get("r_multiple_at_exit") is not None + ] + + # Date range + all_dates_in_trades = sorted(set(r["date"] for r in trade_records)) + min_d = dt.date.fromisoformat(min(all_dates_in_trades)) + max_d = dt.date.fromisoformat(max(all_dates_in_trades)) + # Need prior-day data — go back 1 extra trading day + start_d = min_d - dt.timedelta(days=5) + all_td = trading_days_between(start_d, max_d) + all_dates = [d.isoformat() for d in all_td] + + # Check which ETF dates are missing from cache + missing_etfs = [] + for etf in ETF_TICKERS: + etf_dir = Path(INTRADAY_CACHE_DIR) / etf.upper() + cached_dates = set(f.stem for f in etf_dir.glob("*.parquet")) if etf_dir.exists() else set() + needed = [d for d in all_dates if d not in cached_dates] + if needed: + missing_etfs.append((etf, needed)) + + if missing_etfs: + print(f"\nFetching missing sector ETF data ({len(missing_etfs)} ETFs need data)...") + tickers_to_fetch = [e for e, _ in missing_etfs] + start_str = start_d.isoformat() + end_str = max_d.isoformat() + fetch_and_cache_etf_bars(tickers_to_fetch, start_str, end_str) + else: + print(f"\nAll sector ETF data already cached ({len(ETF_TICKERS)} ETFs).") + + # Load daily OHLCV for all sector ETFs + QQQ + print(f"\nLoading ETF daily data ({len(all_dates)} trading days)...") + etf_daily: dict[str, dict[str, dict]] = {} + for etf in ETF_TICKERS: + etf_daily[etf] = load_etf_daily_ohlcv(etf, all_dates) + print(f" {etf}: {len(etf_daily[etf])} days loaded") + + # Build OBV slope reference for G5a redundancy check + # We'll use the obv_slope_20 from V24 run's enrichment if available, otherwise skip G5a + obv_by_record: list[float | None] = [] + + # Compute features for each trade + sector_gap_vals: list[float] = [] + sector_orb_ret_vals: list[float] = [] + sector_minus_qqq_vals: list[float] = [] + r_mults: list[float] = [] + obv_paired: list[float] = [] + missing = 0 + + for rec in trade_records: + date = rec["date"] + sector = rec["sector"] + etf = SECTOR_TO_ETF.get(sector) + r = rec["r_multiple"] + + if etf is None: + missing += 1 + continue + + # Get prior trading day for sector ETF and QQQ + idx = all_dates.index(date) if date in all_dates else -1 + if idx <= 0: + missing += 1 + continue + prior_date = all_dates[idx - 1] + + etf_data = etf_daily.get(etf, {}) + qqq_data = etf_daily.get("QQQ", {}) + + today_etf = etf_data.get(date) + prior_etf = etf_data.get(prior_date) + today_qqq = qqq_data.get(date) + prior_qqq = qqq_data.get(prior_date) + + if not today_etf or not prior_etf or prior_etf["close"] <= 0: + missing += 1 + continue + + # Feature 1: sector ETF morning gap (today open vs prior close) + etf_gap = (today_etf["open"] - prior_etf["close"]) / prior_etf["close"] + + # Feature 2: sector ETF 5-min ORB return + etf_orb_ret = today_etf.get("orb_return") + if etf_orb_ret is None: + missing += 1 + continue + + # Feature 3: sector ETF gap relative to QQQ gap + if today_qqq and prior_qqq and prior_qqq["close"] > 0: + qqq_gap = (today_qqq["open"] - prior_qqq["close"]) / prior_qqq["close"] + sector_relative_gap = etf_gap - qqq_gap + else: + sector_relative_gap = None + + sector_gap_vals.append(etf_gap) + sector_orb_ret_vals.append(etf_orb_ret) + r_mults.append(r) + if sector_relative_gap is not None: + sector_minus_qqq_vals.append(sector_relative_gap) + + n_valid = len(r_mults) + print(f"\nValid trades: {n_valid} / {len(trade_records)} (missing/unmapped: {missing})") + + if sector_gap_vals: + mean_gap = sum(sector_gap_vals) / len(sector_gap_vals) + pct_positive = sum(1 for g in sector_gap_vals if g > 0) / len(sector_gap_vals) + print(f"Mean sector ETF gap: {mean_gap:+.2%} (positive: {pct_positive:.1%})") + + print("\n" + "=" * 60) + print("GATE RESULTS (G1: |P|≥0.07 & n≥120; G2: avg_R≥0.30R; G3: WR≥5pp; G5a: |ρ_OBV|<0.70)") + + report_feature("sector_etf_gap_pct (morning gap vs prior close)", sector_gap_vals, r_mults) + report_feature("sector_etf_orb_return (ORB bar return 9:30-9:35)", sector_orb_ret_vals, r_mults) + if len(sector_minus_qqq_vals) >= 100: + report_feature( + "sector_minus_qqq_gap (sector ETF gap - QQQ gap)", + sector_minus_qqq_vals, + r_mults[: len(sector_minus_qqq_vals)], + ) + + # Sector breakdown: which sector ETF dominates? + from collections import defaultdict + sector_r: dict[str, list[float]] = defaultdict(list) + for rec in trade_records: + sector_r[rec["sector"]].append(rec["r_multiple"]) + print("\n=== Per-sector breakdown ===") + for s in sorted(sector_r, key=lambda x: -len(sector_r[x])): + rs = sector_r[s] + etf = SECTOR_TO_ETF.get(s, "N/A") + wr = sum(1 for r in rs if r > 0) / len(rs) + avg_r = sum(rs) / len(rs) + print(f" {s:28s} → {etf:5s} n={len(rs):3d} WR={wr:.1%} avg_R={avg_r:+.3f}") + + print("\n=== Summary ===") + print("G2 ≥ 0.30R is the binding gate. OBV-slope 20d passed at 0.394R (V24).") + print("Any feature passing ALL gates warrants Phase 2 backtest.") + + +if __name__ == "__main__": + main()