From 137b413084d2d0feaf49070134511f685d9f29db Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Tue, 21 Apr 2026 22:58:35 -0700 Subject: [PATCH] Diagnose ORB tape ignition (V26): NULL result, V24 remains champion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 5 1-min microstructure features tested on V24 200d trade set (n=101). range_coil_orb shows directional signal (Pearson=-0.128, WR gap +9.9pp) but fails G1 (n=101 < 120) and G2 (avg_R gap 0.195R < 0.30R threshold). p-value ~0.10 — insufficient for promotion. V24 remains champion. Script fetches 1-min bars from Oracle and caches to data/cache/intraday_1min/. Co-Authored-By: Claude Sonnet 4.6 --- .../scripts/diag_orb_tape_ignition.py | 683 ++++++++++++++++++ 1 file changed, 683 insertions(+) create mode 100644 apps/intraday_bt/scripts/diag_orb_tape_ignition.py diff --git a/apps/intraday_bt/scripts/diag_orb_tape_ignition.py b/apps/intraday_bt/scripts/diag_orb_tape_ignition.py new file mode 100644 index 0000000..3439036 --- /dev/null +++ b/apps/intraday_bt/scripts/diag_orb_tape_ignition.py @@ -0,0 +1,683 @@ +""" +V26 ORB Tape Ignition Diagnostic — Opening-Range Microstructure Features + +Hypothesis (Zarattini/Stucchi SSRN 4729284, 2024): ORB quality is predictable from +the microstructure of the opening-range itself: + (a) Range compression: a tighter ORB relative to ATR → more explosive breakout + (b) Volume slope within ORB: increasing volume → conviction/tape ignition + (c) Volume back-half ratio: second half of ORB bars heavier than first half + (d) Range coil: per-bar range DECREASING during ORB → coiling before explosion + (e) RVOL-normalized volume slope: volume shape independent of magnitude (deconfounds + pre-screen filters already in V24 — min_rvol:1.5 + min_premarket_dollar_vol) + +Data: 1-min bars fetched from Oracle (Alpaca endpoint) for each trade (ticker, date) + and cached to data/cache/intraday_1min/{ticker}/{date}.parquet. + The 5-min simulation cache has only one bar per ORB window — 1-min is required. + +Method: + 1. Run V24 simulation over 200d window → 101 trades with r_multiple + 2. Batch-fetch 1-min bars from Oracle for each unique (ticker, date) + 3. Extract 5 × 1-min bars in [9:30, 9:35) ET (the ORB window) + 4. Compute 5 features (all lookahead-free — ORB closes at 9:35, entry is after 9:35): + orb_range_ratio : (orb_high − orb_low) / atr_14 [compressed = low = good?] + range_coil_orb : slope of [r1..r5] (ri=high_i−low_i) / mean_r [neg = coiling] + vol_slope_orb : slope of [v1..v5] / mean_v [pos = accelerating] + vol_rvol_slope_orb : slope of [v1..v5] / (avg_daily_vol_14d/390) [magnitude-independent] + vol_back_half_ratio : (v4+v5) / (v1+v2) [>1 = tape ignition in second half] + 5. Report per-feature: Pearson vs r_multiple, tercile WR/avg_R, coverage + 6. Pairwise correlations vs obv_slope_20 and avg_daily_vol_14d (G5a/G5b redundancy) + +Gates: + G1: |Pearson| ≥ 0.07 on ≥ 120 trades (relaxed to 60 if coverage < 80%) + G2: |top − bottom tercile avg_R| ≥ 0.30R + G3: |top − bottom tercile WR| ≥ 5pp + G4: feature coverage ≥ 50% of trades + G5a: |ρ(feature, obv_slope_20)| < 0.70 (redundancy vs OBV) + G5b: |ρ(feature, avg_daily_vol_14d)| < 0.70 (redundancy vs magnitude filter) +""" +from __future__ import annotations + +import asyncio +import concurrent.futures +import datetime as dt +import json +import os +import sys +from pathlib import Path + +import pyarrow as pa +import pyarrow.parquet as pq +import yaml +from zoneinfo import ZoneInfo + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.."))) + +from libs.common.config import get_settings +from libs.common.time_utils import trading_days_between +from libs.intraday.domain import ORBStrategyParams +from libs.intraday.features import compute_obv_slope_approx, enrich_daily_bars +from libs.intraday.orb_simulator import ORBSimulationState, run_orb_simulation_with_state +from libs.intraday.screener import orb_pre_screen_candidates +from libs.oracle_client.client import OracleClient + +# ── Config ────────────────────────────────────────────────────────────────── +V24_CONFIG = "configs/intraday/strategies/orb_gainers_v24_quality_overlay.yaml" +UNIVERSE_FILE = "configs/symbols_midlarge_snapshot_exact.yaml" +INTRADAY_CACHE_DIR = "data/cache/intraday" # 5-min cache (for simulation) +INTRADAY_1MIN_CACHE = "data/cache/intraday_1min" # 1-min cache (for ORB features) +LOOKBACK_DAYS = 200 +FEATURE_LOOKBACK_TRADING = 25 + +MIN_ORB_BARS = 3 # require at least 3 of 5 1-min ORB bars + +_ET = ZoneInfo("America/New_York") +_MKT_OPEN = dt.time(9, 30) +_MKT_CLOSE = dt.time(16, 0) +_ORB_END = dt.time(9, 35) # 5-min ORB: bars at 9:30, 9:31, 9:32, 9:33, 9:34 + + +# ── Daily Bar Builder (for enrichment — uses 5-min cache) ──────────────────── + +def _parse_ts(ts_raw: str | object) -> dt.datetime: + s = str(ts_raw) + if s.endswith("Z"): + s = s[:-1] + "+00:00" + return dt.datetime.fromisoformat(s).astimezone(_ET) + + +def _build_daily_bar_from_intraday(path: Path, date: str) -> dict | None: + try: + table = pq.read_table(str(path)) + rows = table.to_pydict() + except Exception: + return None + opens, highs, lows, closes, vols = [], [], [], [], [] + for i, ts_raw in enumerate(rows.get("timestamp", [])): + try: + ts = _parse_ts(ts_raw) + except Exception: + continue + if _MKT_OPEN <= ts.time() < _MKT_CLOSE: + opens.append(float(rows["open"][i] or 0)) + highs.append(float(rows["high"][i] or 0)) + lows.append(float(rows["low"][i] or 0)) + closes.append(float(rows["close"][i] or 0)) + vols.append(float(rows["volume"][i] or 0)) + if not opens: + return None + return { + "date": date, "open": opens[0], "high": max(highs), + "low": min(lows), "close": closes[-1], "volume": sum(vols), + } + + +def build_daily_bars(tickers: list[str], dates: list[str], workers: int = 8) -> dict[str, list[dict]]: + root = Path(INTRADAY_CACHE_DIR) + + def _load(ticker: str) -> tuple[str, list[dict]]: + d_path = root / ticker + if not d_path.is_dir(): + return ticker, [] + bars: list[dict] = [] + for date in dates: + p = d_path / f"{date}.parquet" + if not p.exists(): + continue + bar = _build_daily_bar_from_intraday(p, date) + if bar and bar["close"] > 0: + bars.append(bar) + return ticker, bars + + result: dict[str, list[dict]] = {} + with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex: + for ticker, bars in ex.map(_load, tickers): + if bars: + result[ticker] = bars + return result + + +def load_intraday_bulk(candidates: dict[str, list[str]]) -> dict[str, dict[str, list[dict]]]: + """Load 5-min bars for simulation from cache.""" + import pandas as pd + result: dict[str, dict[str, list[dict]]] = {} + for date, tickers in candidates.items(): + day_bars: dict[str, list[dict]] = {} + for ticker in tickers: + p = Path(INTRADAY_CACHE_DIR) / ticker / f"{date}.parquet" + if not p.exists(): + continue + try: + df = pd.read_parquet(str(p)) + if not df.empty and len(df) >= 5: + day_bars[ticker] = df.to_dict("records") + except Exception: + pass + if day_bars: + result[date] = day_bars + return result + + +# ── 1-Min Oracle Fetcher ────────────────────────────────────────────────────── + +async def _fetch_1min_batch( + pairs: list[tuple[str, str]], # [(ticker, date), ...] + oracle_url: str, + concurrency: int = 8, +) -> dict[str, dict[str, list[dict]]]: + """Fetch 1-min bars for each (ticker, date) from Oracle, cache to disk. + + Uses multi-ticker endpoint: GET /api/v1/alpaca/intraday?tickers=A,B&interval=1min&... + Returns {date: {ticker: [bar_dicts]}}. + """ + cache_root = Path(INTRADAY_1MIN_CACHE) + cache_root.mkdir(parents=True, exist_ok=True) + + def _cache_path(ticker: str, date: str) -> Path: + return cache_root / ticker / f"{date}.parquet" + + def _save_bars(ticker: str, date: str, bars: list[dict]) -> None: + p = _cache_path(ticker, date) + p.parent.mkdir(parents=True, exist_ok=True) + tbl = pa.table({ + "timestamp": [str(b["timestamp"]) for b in bars], + "open": [float(b["open"]) for b in bars], + "high": [float(b["high"]) for b in bars], + "low": [float(b["low"]) for b in bars], + "close": [float(b["close"]) for b in bars], + "volume": [float(b["volume"]) for b in bars], + }) + pq.write_table(tbl, str(p)) + + def _load_bars(ticker: str, date: str) -> list[dict]: + p = _cache_path(ticker, date) + table = pq.read_table(str(p)) + rows = table.to_pydict() + return [ + {k: rows[k][i] for k in ("timestamp", "open", "high", "low", "close", "volume")} + for i in range(len(rows["timestamp"])) + ] + + # Split into cache hits and misses; group misses by date for batch calls + hits: list[tuple[str, str]] = [] + misses_by_date: dict[str, list[str]] = {} + for ticker, date in pairs: + if _cache_path(ticker, date).exists(): + hits.append((ticker, date)) + else: + misses_by_date.setdefault(date, []).append(ticker) + + n_miss_pairs = sum(len(v) for v in misses_by_date.values()) + print(f" 1-min cache: {len(hits)} hits, {n_miss_pairs} Oracle fetches needed ({len(misses_by_date)} dates)") + + result: dict[str, dict[str, list[dict]]] = {} + sem = asyncio.Semaphore(concurrency) + fetched_count = 0 + + async def _fetch_date(date: str, tickers: list[str], client: OracleClient) -> dict[str, list[dict]]: + async with sem: + raw = await client.get( + "/api/v1/alpaca/intraday", + params={ + "tickers": ",".join(tickers), + "interval": "1min", + "start_date": date, + "end_date": date, + }, + ) + bars_by_ticker = raw.get("bars", {}) + day_result: dict[str, list[dict]] = {} + for ticker in tickers: + bars = [ + { + "timestamp": b.get("timestamp", ""), + "open": float(b.get("open", 0)), + "high": float(b.get("high", 0)), + "low": float(b.get("low", 0)), + "close": float(b.get("close", 0)), + "volume": float(b.get("volume", 0)), + } + for b in bars_by_ticker.get(ticker, []) + ] + if bars: + day_result[ticker] = bars + _save_bars(ticker, date, bars) + return day_result + + async with OracleClient(base_url=oracle_url) as client: + if misses_by_date: + tasks = {date: _fetch_date(date, tickers, client) for date, tickers in misses_by_date.items()} + dates = list(tasks.keys()) + coros = list(tasks.values()) + results_list = await asyncio.gather(*coros, return_exceptions=True) + for date, res in zip(dates, results_list): + fetched_count += 1 + if isinstance(res, Exception): + print(f" WARN: {date} fetch error: {res}") + continue + if res: + result[date] = result.get(date, {}) + result[date].update(res) + if fetched_count % 20 == 0 or fetched_count == len(dates): + print(f" Fetched {fetched_count}/{len(dates)} dates from Oracle") + + # Load cache hits + for ticker, date in hits: + try: + bars = _load_bars(ticker, date) + result.setdefault(date, {})[ticker] = bars + except Exception: + pass + + return result + + +# ── ORB Bar Extraction ──────────────────────────────────────────────────────── + +def extract_orb_bars(raw_bars: list[dict]) -> list[dict]: + """Return 1-min bars with timestamps in [9:30, 9:35) ET, sorted by time.""" + orb: list[tuple[dt.datetime, dict]] = [] + for bar in raw_bars: + ts_raw = bar.get("timestamp", "") + if not ts_raw: + continue + try: + ts = _parse_ts(ts_raw) + except Exception: + continue + if _MKT_OPEN <= ts.time() < _ORB_END: + orb.append((ts, bar)) + orb.sort(key=lambda x: x[0]) + return [b for _, b in orb] + + +# ── Feature Computations ────────────────────────────────────────────────────── + +def _linreg_slope_normalized(vals: list[float]) -> float | None: + """Linear regression slope over [0..n-1], normalized by mean(vals). Scale-free.""" + n = len(vals) + if n < 2: + return None + mean_v = sum(vals) / n + if mean_v <= 0: + return None + x_mean = (n - 1) / 2.0 + num = sum((i - x_mean) * (vals[i] - mean_v) for i in range(n)) + denom = sum((i - x_mean) ** 2 for i in range(n)) + if denom <= 0: + return None + return (num / denom) / mean_v + + +def compute_orb_range_ratio(orb_bars: list[dict], atr_14: float | None) -> float | None: + """(orb_high − orb_low) / atr_14. Low = compressed = hypothesis: explosive breakout.""" + if len(orb_bars) < MIN_ORB_BARS or atr_14 is None or atr_14 <= 0: + return None + highs = [float(b.get("high", 0) or 0) for b in orb_bars] + lows = [float(b.get("low", 0) or 0) for b in orb_bars] + orb_range = max(highs) - min(lows) + if orb_range <= 0: + return None + return orb_range / atr_14 + + +def compute_range_coil_orb(orb_bars: list[dict]) -> float | None: + """Normalized slope of per-bar ranges. Negative = bars coiling (tightening).""" + if len(orb_bars) < MIN_ORB_BARS: + return None + ranges = [float((b.get("high") or 0) - (b.get("low") or 0)) for b in orb_bars] + return _linreg_slope_normalized(ranges) + + +def compute_vol_slope_orb(orb_bars: list[dict]) -> float | None: + """Normalized slope of 1-min volumes during ORB. Positive = volume increasing.""" + if len(orb_bars) < MIN_ORB_BARS: + return None + vols = [float(b.get("volume", 0) or 0) for b in orb_bars] + return _linreg_slope_normalized(vols) + + +def compute_vol_rvol_slope_orb( + orb_bars: list[dict], avg_daily_vol_14d: float | None +) -> float | None: + """Volume slope normalized by avg 14d bar volume (deconfounds magnitude filters).""" + if len(orb_bars) < MIN_ORB_BARS or avg_daily_vol_14d is None or avg_daily_vol_14d <= 0: + return None + avg_bar = avg_daily_vol_14d / 390.0 + if avg_bar <= 0: + return None + vols = [float(b.get("volume", 0) or 0) / avg_bar for b in orb_bars] + return _linreg_slope_normalized(vols) + + +def compute_vol_back_half_ratio(orb_bars: list[dict]) -> float | None: + """(v4+v5) / (v1+v2) — tape ignition: second half volume heavier than first.""" + if len(orb_bars) < 4: + return None + vols = [float(b.get("volume", 0) or 0) for b in orb_bars] + first = vols[0] + vols[1] + last = vols[-2] + vols[-1] + if first <= 0: + return None + return last / first + + +# ── Stats Helpers ───────────────────────────────────────────────────────────── + +def pearson(xs: list[float], ys: list[float]) -> float | None: + if len(xs) != len(ys) or len(xs) < 2: + return None + n = len(xs) + xm = sum(xs) / n + ym = 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], outcomes_r: list[float]) -> dict: + if len(vals) < 6: + return {} + pairs = sorted(zip(vals, outcomes_r), key=lambda p: p[0]) + n = len(pairs) + t = n // 3 + + def stats(sub): + ys = [p[1] for p in sub] + wr = sum(1 for y in ys if y > 0) / len(ys) if ys else 0.0 + avg = sum(ys) / len(ys) if ys else 0.0 + return {"n": len(ys), "wr": wr, "avg_r": avg} + + return {"low": stats(pairs[:t]), "mid": stats(pairs[t:2*t]), "high": stats(pairs[2*t:])} + + +# ── Main ───────────────────────────────────────────────────────────────────── + +def main() -> None: + print("=== V26 ORB Tape Ignition Diagnostic (1-min bars) ===\n") + + settings = get_settings() + oracle_url = settings.stock_oracle_url + + # 1. Load V24 config + with open(V24_CONFIG) as f: + raw = yaml.safe_load(f) + params = ORBStrategyParams(**raw["orb_strategy"]) + print(f"V24 config loaded. weight_obv_slope={params.weight_obv_slope}") + + # 2. Determine 200d trading window + today = dt.date(2026, 4, 21) + all_td = trading_days_between(today - dt.timedelta(days=400), today) + trading_days_list = [d.isoformat() for d in all_td[-LOOKBACK_DAYS:]] + print(f"Window: {trading_days_list[0]} → {trading_days_list[-1]} ({len(trading_days_list)} trading days)") + + extended_td = [d.isoformat() for d in all_td[-(LOOKBACK_DAYS + FEATURE_LOOKBACK_TRADING + 10):]] + first_cal = dt.date.fromisoformat(extended_td[0]) + last_cal = dt.date.fromisoformat(trading_days_list[-1]) + needed_dates: list[str] = [] + d = first_cal + while d <= last_cal: + needed_dates.append(d.isoformat()) + d += dt.timedelta(days=1) + + # 3. Load universe + with open(UNIVERSE_FILE) as f: + udata = yaml.safe_load(f) + universe = udata.get("symbols", udata) if isinstance(udata, dict) else udata + if "QQQ" not in universe: + universe = list(universe) + ["QQQ"] + print(f"Universe: {len(universe)} tickers") + + # 4. Build daily bars (5-min cache → daily OHLCV) + print(f"\nBuilding daily bars ({len(needed_dates)} calendar days)...") + daily_bars = build_daily_bars(universe, needed_dates) + print(f"Built daily bars for {len(daily_bars)} tickers") + + # 5. Enrichment (provides atr_14, avg_daily_vol_14d, obv_slope_20) + print("Computing enrichment...") + enrichment = enrich_daily_bars(daily_bars, trading_days_list) + print(f"Enrichment for {len(enrichment)} tickers") + + # 6. Pre-screen + load 5-min bars for simulation + candidates = orb_pre_screen_candidates( + daily_bars, trading_days_list, enrichment, + min_price=params.min_price, + min_atr=params.min_atr_14, + min_avg_dollar_vol=params.min_avg_dollar_volume, + max_per_day=None, + ) + total_pairs = sum(len(v) for v in candidates.values()) + print(f"Pre-screened: {total_pairs} ticker-days") + print("Loading 5-min intraday bars for simulation...") + all_intraday = load_intraday_bulk(candidates) + print(f"Loaded: {sum(len(v) for v in all_intraday.values())} ticker-days") + + # 7. Run V24 simulation + print("\nRunning V24 simulation...") + state = ORBSimulationState(equity=params.initial_capital) + day_results, _ = run_orb_simulation_with_state( + all_intraday, trading_days_list, params, enrichment, state=state, + ) + all_trades = [t for dr in day_results for t in dr.trades] + trades_with_r = [t for t in all_trades if getattr(t, "r_multiple_at_exit", None) is not None] + print(f"Total trades: {len(all_trades)}, with r_multiple: {len(trades_with_r)}") + + if len(trades_with_r) < 20: + print("ABORT: fewer than 20 trades with r_multiple") + return + + # 8. Fetch 1-min bars from Oracle for each unique (ticker, date) + trade_pairs = list({(str(t.ticker), str(t.date)[:10]) for t in trades_with_r}) + print(f"\nFetching 1-min bars for {len(trade_pairs)} unique (ticker, date) pairs...") + bars_1min = asyncio.run(_fetch_1min_batch(trade_pairs, oracle_url)) + n_pairs_loaded = sum(len(v) for v in bars_1min.values()) + print(f"1-min data available: {n_pairs_loaded} ticker-days") + + # 9. Compute ORB tape ignition features per trade + print("\nComputing ORB tape ignition features...") + sorted_daily: dict[str, list[dict]] = { + t: sorted(bars, key=lambda b: b["date"]) + for t, bars in daily_bars.items() + } + + annotated: list[dict] = [] + missing_counts: dict[str, int] = { + "orb_range_ratio": 0, "range_coil_orb": 0, "vol_slope_orb": 0, + "vol_rvol_slope_orb": 0, "vol_back_half_ratio": 0, "obv_slope": 0, + "no_1min_data": 0, + } + + for trade in trades_with_r: + ticker = trade.ticker + date = str(trade.date)[:10] + r = float(trade.r_multiple_at_exit) + + # Get 1-min bars for this (ticker, date) + raw_1min = bars_1min.get(date, {}).get(ticker, []) + if not raw_1min: + missing_counts["no_1min_data"] += 1 + # Can't compute any ORB features — fill None + orb_bars = [] + else: + orb_bars = extract_orb_bars(raw_1min) + + # ATR and avg_vol from enrichment + enrich_day = enrichment.get(ticker, {}).get(date, {}) + atr_14 = enrich_day.get("atr_14") + avg_daily_vol = enrich_day.get("avg_daily_vol_14d") + + f_range_ratio = compute_orb_range_ratio(orb_bars, atr_14) + f_coil = compute_range_coil_orb(orb_bars) + f_vol_slope = compute_vol_slope_orb(orb_bars) + f_vol_rvol_slope = compute_vol_rvol_slope_orb(orb_bars, avg_daily_vol) + f_back_half = compute_vol_back_half_ratio(orb_bars) + + # OBV slope for pairwise correlation + bars_t = sorted_daily.get(ticker, []) + prev_bars = [b for b in bars_t if b["date"][:10] < date] + obv = compute_obv_slope_approx(prev_bars, lookback=20) + + for fname, fval in [ + ("orb_range_ratio", f_range_ratio), ("range_coil_orb", f_coil), + ("vol_slope_orb", f_vol_slope), ("vol_rvol_slope_orb", f_vol_rvol_slope), + ("vol_back_half_ratio", f_back_half), + ]: + if fval is None: + missing_counts[fname] += 1 + if obv is None: + missing_counts["obv_slope"] += 1 + + annotated.append({ + "ticker": ticker, "date": date, "r": r, "win": r > 0, + "orb_range_ratio": f_range_ratio, + "range_coil_orb": f_coil, + "vol_slope_orb": f_vol_slope, + "vol_rvol_slope_orb": f_vol_rvol_slope, + "vol_back_half_ratio": f_back_half, + "obv_slope": obv, + "avg_daily_vol": avg_daily_vol, + "n_orb_bars": len(orb_bars), + }) + + total = len(annotated) + print(f"Annotated: {total} trades") + orb_bar_counts = [a["n_orb_bars"] for a in annotated] + avg_orb = sum(orb_bar_counts) / len(orb_bar_counts) if orb_bar_counts else 0 + n_full = sum(1 for c in orb_bar_counts if c >= 5) + print(f"ORB bars: avg={avg_orb:.1f}, full(≥5)={n_full}/{total}") + print(f"No 1-min data: {missing_counts['no_1min_data']}/{total}") + for fname in ["orb_range_ratio", "range_coil_orb", "vol_slope_orb", "vol_rvol_slope_orb", "vol_back_half_ratio"]: + print(f" Missing {fname}: {missing_counts[fname]}/{total}") + + # 10. Feature analysis + feature_defs = [ + ("orb_range_ratio", "ORB range / ATR_14 — compressed ORB before breakout", "low"), + ("range_coil_orb", "Slope of per-bar ranges — negative = coiling", "low"), + ("vol_slope_orb", "Volume slope (shape) during ORB — positive = building conviction", "high"), + ("vol_rvol_slope_orb", "Volume slope / avg 14d bar vol — magnitude-independent shape", "high"), + ("vol_back_half_ratio", "Vol(last 2 bars) / Vol(first 2 bars) — tape ignition proxy", "high"), + ] + + print("\n" + "=" * 90) + print("FEATURE ANALYSIS — V24 200d trade set") + print("=" * 90) + + results: dict[str, dict | None] = {} + for feat_name, description, best_tercile in feature_defs: + valid = [(a[feat_name], a["r"]) for a in annotated if a[feat_name] is not None] + if len(valid) < 20: + print(f"\n{feat_name}: SKIP — only {len(valid)} valid trades (need ≥20)") + results[feat_name] = None + continue + + vals = [v[0] for v in valid] + rs = [v[1] for v in valid] + overall_wr = sum(1 for v in valid if v[1] > 0) / len(valid) + rho = pearson(vals, rs) + tstat = tercile_stats(vals, rs) + coverage_pct = len(valid) / total + worst_tercile = "high" if best_tercile == "low" else "low" + + print(f"\n{'─'*60}") + print(f"FEATURE: {feat_name}") + print(f" Description: {description}") + print(f" n={len(valid)}/{total} ({coverage_pct*100:.0f}% coverage), overall WR={overall_wr*100:.1f}%") + rho_abs = abs(rho) if rho is not None else 0.0 + print(f" Pearson(feature, r_multiple) = {rho:.4f}" if rho is not None else " Pearson = n/a") + + if tstat: + h, m, lo = tstat["high"], tstat["mid"], tstat["low"] + best = tstat[best_tercile] + worst = tstat[worst_tercile] + print(f" Tercile breakdown (low→high feature value):") + print(f" Bottom: n={lo['n']}, WR={lo['wr']*100:.1f}%, avg_R={lo['avg_r']:+.3f}") + print(f" Middle: n={m['n']}, WR={m['wr']*100:.1f}%, avg_R={m['avg_r']:+.3f}") + print(f" Top: n={h['n']}, WR={h['wr']*100:.1f}%, avg_R={h['avg_r']:+.3f}") + + min_trades = 120 if coverage_pct >= 0.80 else 60 + g1 = rho_abs >= 0.07 and len(valid) >= min_trades + g2 = best["avg_r"] - worst["avg_r"] >= 0.30 + g3 = best["wr"] >= worst["wr"] + 0.05 + g4 = coverage_pct >= 0.50 + n_failed = sum([not g1, not g2, not g3, not g4]) + overall_pass = n_failed == 0 + + print(f" Gates (best='{best_tercile}' tercile):") + print(f" G1 |Pearson|≥0.07 + n≥{min_trades}: {rho_abs:.4f}, n={len(valid)} → {'PASS ✓' if g1 else 'FAIL ✗'}") + print(f" G2 avg_R gap ≥ 0.30R: {best['avg_r'] - worst['avg_r']:+.3f} → {'PASS ✓' if g2 else 'FAIL ✗'}") + print(f" G3 WR gap ≥ 5pp: {(best['wr'] - worst['wr'])*100:+.1f}pp → {'PASS ✓' if g3 else 'FAIL ✗'}") + print(f" G4 coverage ≥ 50%: {coverage_pct*100:.0f}% → {'PASS ✓' if g4 else 'FAIL ✗'}") + print(f" VERDICT: {'ALL GATES PASS → PROCEED TO PHASE 2' if overall_pass else f'FAIL ({n_failed} gate(s) failed)'}") + + results[feat_name] = { + "pass": overall_pass, "pearson": rho, "stats": tstat, + "n": len(valid), "coverage": coverage_pct, "best_tercile": best_tercile, + } + + # 11. Pairwise correlations (G5a vs OBV, G5b vs avg_daily_vol) + print(f"\n{'─'*60}") + print("PAIRWISE CORRELATIONS (G5a: vs obv_slope_20, G5b: vs avg_daily_vol_14d):") + for feat_name in [fd[0] for fd in feature_defs]: + for ref_key, ref_label, gate_name in [ + ("obv_slope", "obv_slope_20", "G5a"), + ("avg_daily_vol", "avg_daily_vol_14d", "G5b"), + ]: + combined = [ + (a[feat_name], a[ref_key]) for a in annotated + if a[feat_name] is not None and a[ref_key] is not None + ] + if len(combined) >= 10: + rho_x = pearson([c[0] for c in combined], [c[1] for c in combined]) + if rho_x is not None: + gate_pass = abs(rho_x) < 0.70 + print(f" ρ({feat_name[:24]:24s}, {ref_label}): {rho_x:+.4f} {gate_name}: {'PASS ✓' if gate_pass else 'FAIL (redundant) ✗'}") + + # Inter-feature correlations + feat_keys = [fd[0] for fd in feature_defs] + print("\n Inter-feature correlations:") + for i, f1 in enumerate(feat_keys): + for f2 in feat_keys[i+1:]: + combined = [(a[f1], a[f2]) for a in annotated if a[f1] is not None and a[f2] is not None] + if len(combined) >= 10: + rho_x = pearson([c[0] for c in combined], [c[1] for c in combined]) + if rho_x is not None: + print(f" ρ({f1[:22]:22s}, {f2[:22]:22s}): {rho_x:+.4f}") + + # 12. Summary + passing = [name for name, r in results.items() if r is not None and r["pass"]] + failed = [name for name, r in results.items() if r is not None and not r["pass"]] + skipped = [name for name, r in results.items() if r is None] + + print(f"\n{'='*90}") + print("SUMMARY") + print(f"{'='*90}") + print(f"Features passing all gates: {passing if passing else 'NONE'}") + print(f"Features failing gates: {failed if failed else 'NONE'}") + print(f"Features skipped: {skipped if skipped else 'NONE'}") + + if passing: + best = max(passing, key=lambda n: abs(results[n]["pearson"] or 0)) + pearson_val = results[best]["pearson"] + obv_pearson_ref = 0.2349 + weight_magnitude = round(0.05 * min(1.0, abs(pearson_val) / obv_pearson_ref), 2) + weight_magnitude = max(weight_magnitude, 0.02) + best_direction = results[best]["best_tercile"] + weight_sign = +1 if best_direction == "high" else -1 + + print(f"\nVERDICT: PROCEED TO PHASE 2") + print(f" Best feature: {best}") + print(f" Pearson: {pearson_val:.4f}") + print(f" Direction: '{best_direction}' tercile is best → weight = {weight_sign * weight_magnitude:+.3f}") + print(f" Suggested weight_tape_ignition in V26 config: {weight_sign * weight_magnitude:+.3f}") + print(f"\n → Wire '{best}' as intraday feature in ORB simulator") + print(f" → Add weight_tape_ignition = {weight_sign * weight_magnitude:+.3f} to V26 config (parent V24)") + else: + print(f"\nVERDICT: ABORT — Opening-range tape ignition axis null on V24 200d trade set") + print(" → V24 remains champion. Microstructure axis exhausted (1-min data confirmed).") + print(" → Next Ralph iteration: RSI-14, BB %B, or gravitational pull from libs/features/market_features.py") + + +if __name__ == "__main__": + main()