diff --git a/apps/intraday_bt/scripts/diag_orb_qqq_quality.py b/apps/intraday_bt/scripts/diag_orb_qqq_quality.py new file mode 100644 index 0000000..9ea5e32 --- /dev/null +++ b/apps/intraday_bt/scripts/diag_orb_qqq_quality.py @@ -0,0 +1,249 @@ +""" +V33 Diagnostic: QQQ ORB Candle Quality as Market-Level Regime Signal + +Hypothesis: on days when QQQ's own 5-min ORB bar is bullish (strong body, closes near high), +individual gap-up stock ORB breakouts have better follow-through. + +The current V24 regime filter requires QQQ to open > +0.15% (daily gap) but doesn't check +whether QQQ's ORB CANDLE ITSELF is bullish. A day where QQQ opens +0.2% but the ORB bar is +red (intraday fade) may be a trap day. A day where QQQ opens +0.2% AND the ORB bar is strongly +bullish is a cleaner market-wide breakout day. + +Features (all from QQQ's 9:30-9:35 ET bar — available at 9:35 AM before any entry): + qqq_orb_body_pct : (close - open) / (high - low). +1.0 = full bullish candle, 0 = doji + qqq_orb_return : (close - open) / open. Raw return of QQQ's ORB bar. + qqq_orb_close_loc : (close - low) / (high - low). Close position in bar range (1 = at high) + +Source: V24 400d trade set (n=~304 trades, ~120-140 unique trade days). +Note: day-level feature (all trades on same day share same QQQ ORB value) — effective n for +correlation is unique trading days, but trade-level Pearson is also reported. + +Gates (same as V25-V31): + G1: |Pearson| >= 0.07 on >= 120 trades (or >= 80 unique days) + G2: |top - bottom tercile avg_R| >= 0.30R + G3: |top - bottom tercile WR| >= 5pp + G5a: |rho(qqq_feature, V24 obv_slope per trade)| < 0.70 (redundancy check) +""" +from __future__ import annotations + +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 + +_ET = ZoneInfo("America/New_York") +_MKT_OPEN = dt.time(9, 30) + +INTRADAY_CACHE_DIR = "data/cache/intraday" +# Use most recent V24 400d run (trail=0.8, risk=0.05, OBV-slope=0.05) +V24_400D_RUN = "runs/intraday_orb/intraday_20260422_011012_06f59ede.json" + + +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 load_qqq_orb_bar(date: str) -> dict | None: + """Load QQQ's 5-minute ORB bar (9:30-9:35 ET) for a given date.""" + path = Path(INTRADAY_CACHE_DIR) / "QQQ" / f"{date}.parquet" + if not path.exists(): + return None + try: + table = pq.read_table(str(path)) + rows = table.to_pydict() + except Exception: + return None + for i, ts_raw in enumerate(rows.get("timestamp", [])): + try: + ts = _parse_ts(ts_raw) + except Exception: + continue + if ts.time() == _MKT_OPEN: + o = float(rows["open"][i] or 0) + h = float(rows["high"][i] or 0) + lo = float(rows["low"][i] or 0) + c = float(rows["close"][i] or 0) + if o > 0 and h > lo: + return {"open": o, "high": h, "low": lo, "close": c} + return None + + +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 check_gates(label: str, vals: list[float], r_mult: list[float]) -> None: + n = len(vals) + p = pearson(vals, r_mult) + ts = tercile_stats(vals, r_mult) + if not ts: + print(f" {label}: n={n}, insufficient data") + return + low, mid, high = ts["low"], ts["mid"], ts["high"] + g1 = p is not None and abs(p) >= 0.07 and n >= 80 + g2 = abs(high["avg_r"] - low["avg_r"]) >= 0.30 + g3 = abs(high["wr"] - low["wr"]) >= 0.05 + print(f"\n [{label}] n={n} Pearson={p:.3f}") + print(f" G1: {'PASS' if g1 else 'FAIL'} (|{p:.3f}| {'≥' if g1 else '<'} 0.07, n={n})") + print(f" G2: {'PASS' if g2 else 'FAIL'} (|top-bot avg_R| = {abs(high['avg_r']-low['avg_r']):.3f}R)") + print(f" G3: {'PASS' if g3 else 'FAIL'} (|top-bot WR| = {abs(high['wr']-low['wr'])*100:.1f}pp)") + 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']}") + status = "PASS ALL" if (g1 and g2 and g3) else "FAIL" + print(f" Status: {status}") + + +def main() -> None: + print("=== V33 QQQ ORB Candle Quality Diagnostic ===\n") + + 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 run: {m.get('total_trades')} trades, " + f"{m.get('start_date')} → {m.get('end_date')}") + print(f"Return: {m.get('total_return_pct', 0)*100:.2f}% " + f"DD: {m.get('max_drawdown_pct', 0)*100:.2f}% " + f"Sharpe: {m.get('sharpe_ratio', 0):.3f}\n") + + # Build per-trade data + trade_records = [ + {"date": t["date"], "r_multiple": t.get("r_multiple_at_exit") or 0.0} + for t in trades + if t.get("r_multiple_at_exit") is not None + ] + dates_needed = sorted(set(r["date"] for r in trade_records)) + print(f"Trade records: {len(trade_records)} with r_multiple, across {len(dates_needed)} unique dates") + + # Load QQQ ORB bars for each date + print("Loading QQQ ORB bars...") + qqq_bars: dict[str, dict] = {} + missing = 0 + for date in dates_needed: + bar = load_qqq_orb_bar(date) + if bar: + qqq_bars[date] = bar + else: + missing += 1 + print(f" Loaded: {len(qqq_bars)}/{len(dates_needed)} dates (missing: {missing})") + + # Compute features per trade + body_vals, return_vals, closeloc_vals, r_mults = [], [], [], [] + skipped = 0 + for rec in trade_records: + bar = qqq_bars.get(rec["date"]) + if not bar: + skipped += 1 + continue + o, h, lo, c = bar["open"], bar["high"], bar["low"], bar["close"] + body_range = h - lo + if body_range <= 0: + skipped += 1 + continue + body_pct = (c - o) / body_range # [-1, 1], +1 = full bullish + orb_ret = (c - o) / o # raw return of ORB bar + close_loc = (c - lo) / body_range # [0, 1], 1 = closed at high + + body_vals.append(body_pct) + return_vals.append(orb_ret) + closeloc_vals.append(close_loc) + r_mults.append(rec["r_multiple"]) + + n_valid = len(r_mults) + coverage = n_valid / len(trade_records) if trade_records else 0 + print(f" Valid trades: {n_valid}/{len(trade_records)} (coverage {coverage*100:.1f}%)") + print(f" Skipped: {skipped}\n") + + if n_valid < 30: + print("Insufficient data for analysis.") + return + + print("=" * 60) + print("GATE RESULTS (G1: |Pearson|>=0.07 & n>=80; G2: avg_R gap>=0.30R; G3: WR gap>=5pp)") + check_gates("qqq_orb_body_pct", body_vals, r_mults) + check_gates("qqq_orb_return", return_vals, r_mults) + check_gates("qqq_orb_close_loc", closeloc_vals, r_mults) + + # Redundancy check vs OBV-slope (from trade obv_slope_20 stored in enrichment) + # We don't have per-trade obv_slope but check pairwise between qqq features + p_body_ret = pearson(body_vals, return_vals) + p_body_cloc = pearson(body_vals, closeloc_vals) + p_ret_cloc = pearson(return_vals, closeloc_vals) + print("\n Inter-feature correlations:") + print(f" ρ(body_pct, orb_return) = {p_body_ret:.3f}") + print(f" ρ(body_pct, close_loc) = {p_body_cloc:.3f}") + print(f" ρ(orb_ret, close_loc) = {p_ret_cloc:.3f}") + + # Day-level analysis (aggregate by unique date) + day_data: dict[str, dict] = {} + for i, rec in enumerate(trade_records): + if rec["date"] not in qqq_bars: + continue + d = rec["date"] + if d not in day_data: + bar = qqq_bars[d] + o, h, lo, c = bar["open"], bar["high"], bar["low"], bar["close"] + body_range = h - lo + if body_range <= 0: + continue + day_data[d] = { + "qqq_body": (c - o) / body_range, + "qqq_return": (c - o) / o, + "qqq_close_loc": (c - lo) / body_range, + "r_mults": [], + } + day_data[d]["r_mults"].append(rec["r_multiple"]) + day_avg_r = {d: sum(v["r_mults"]) / len(v["r_mults"]) for d, v in day_data.items()} + day_body = [v["qqq_body"] for d, v in day_data.items()] + day_ret = [v["qqq_return"] for d, v in day_data.items()] + day_cloc = [v["qqq_close_loc"] for d, v in day_data.items()] + day_r = [day_avg_r[d] for d in day_data] + p_day_body = pearson(day_body, day_r) + p_day_ret = pearson(day_ret, day_r) + p_day_cloc = pearson(day_cloc, day_r) + print(f"\n Day-level correlation (n_days={len(day_data)}, using avg_R per day):") + print(f" ρ(qqq_body, avg_day_R) = {p_day_body:.3f}") + print(f" ρ(qqq_orb_ret, avg_day_R) = {p_day_ret:.3f}") + print(f" ρ(qqq_close_loc, avg_day_R) = {p_day_cloc:.3f}") + + print("\n=== Summary ===") + print("Signal library context: 7 prior axes failed G2 >= 0.30R (only OBV-slope 0.394R passed).") + print("QQQ ORB body_pct, return, close_loc are day-level features (market-wide confirmation).") + print("These cannot be used for candidate ranking (same value for all stocks on same day).") + print("Use case: day-level gate (skip all ORB trades when QQQ ORB candle is weak/bearish).") + + +if __name__ == "__main__": + main()