From 16f49411cb855ae1f04908269b1a53ac719d12db Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Wed, 22 Apr 2026 02:54:11 -0700 Subject: [PATCH] Add V34-V44 ORB diagnostic scripts; wire obv_slope_5 + min_obv_slope_20d infra MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signal axes tested (V34-V44, all failed G2 ≥ 0.30R gate): - V34 obv_slope_5 (5d): G2=0.083R (null) - V35 obv_slope composite (5d+20d): regime artifact (200d +18pp, 400d -12pp) - V36 RSI-14: G2=0.148R, G5a=0.742 (redundant with OBV) - V37 BB %B / BB width: G2=0.186R - V38 dollar_vol_trend / sleep_streak / prior_day / vol_trend: all fail G2 - V39 premarket acceleration + hold ratio: G2=0.013R (null) - V40 prior-day market breadth: G2=0.008R (null) - V41 min_obv_slope_20d=0.0 hard filter: -15pp (OBV as gate too aggressive) - V42 52w-high proximity + range position: G2=0.200R (best near-miss, fails) - V43 30-min ORB window: -14.79% (catastrophic) - V44 trailing multiplier sweep 0.6-1.0: 0.80 confirmed global optimum Infrastructure added (backward-compatible, V24 parity preserved): - features.py: obv_slope_5 enrichment key - domain.py: weight_obv_slope_5=0.0, min_obv_slope_20d=None - orb_simulator.py: 5 wiring sites for obv_slope_5; min_obv_slope_20d gate V24 remains live champion. 20 signal axes exhausted. Co-Authored-By: Claude Sonnet 4.6 --- apps/intraday_bt/scripts/diag_orb_52w_high.py | 215 +++++++++++++ apps/intraday_bt/scripts/diag_orb_bb_pctb.py | 275 +++++++++++++++++ apps/intraday_bt/scripts/diag_orb_breadth.py | 283 ++++++++++++++++++ .../scripts/diag_orb_obv_acceleration.py | 283 ++++++++++++++++++ .../scripts/diag_orb_premarket_accel.py | 206 +++++++++++++ .../intraday_bt/scripts/diag_orb_remaining.py | 263 ++++++++++++++++ apps/intraday_bt/scripts/diag_orb_rsi14.py | 263 ++++++++++++++++ .../orb_gainers_v41_obv_filter.yaml | 115 +++++++ .../strategies/orb_gainers_v43_30min.yaml | 114 +++++++ libs/intraday/domain.py | 8 + libs/intraday/features.py | 4 + libs/intraday/orb_simulator.py | 9 + 12 files changed, 2038 insertions(+) create mode 100644 apps/intraday_bt/scripts/diag_orb_52w_high.py create mode 100644 apps/intraday_bt/scripts/diag_orb_bb_pctb.py create mode 100644 apps/intraday_bt/scripts/diag_orb_breadth.py create mode 100644 apps/intraday_bt/scripts/diag_orb_obv_acceleration.py create mode 100644 apps/intraday_bt/scripts/diag_orb_premarket_accel.py create mode 100644 apps/intraday_bt/scripts/diag_orb_remaining.py create mode 100644 apps/intraday_bt/scripts/diag_orb_rsi14.py create mode 100644 configs/intraday/strategies/orb_gainers_v41_obv_filter.yaml create mode 100644 configs/intraday/strategies/orb_gainers_v43_30min.yaml diff --git a/apps/intraday_bt/scripts/diag_orb_52w_high.py b/apps/intraday_bt/scripts/diag_orb_52w_high.py new file mode 100644 index 0000000..51b4ae3 --- /dev/null +++ b/apps/intraday_bt/scripts/diag_orb_52w_high.py @@ -0,0 +1,215 @@ +""" +V42 Diagnostic: 52-Week High Proximity Signal + +Hypothesis: Stocks trading near their 52-week high on the ORB entry day have: + (a) confirmed long-term uptrend momentum + (b) no overhead price resistance (buyers that are "underwater" don't sell) + (c) higher institutional confidence = better ORB follow-through + +Features: + pct_from_52w_high : (last_close - 52w_high) / 52w_high [≤ 0; 0 = at high] + is_near_52w_high : 1 if within 5% of 52w high, else 0 (binary version) + dist_from_52w_low : (last_close - 52w_low) / (52w_high - 52w_low) [0-1, 1 = at high] + "position in 52-week range" — independent from proximity to high + +Source: V24 400d run JSON + daily parquet cache (need 252 prior trading days). +""" +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) +_MKT_CLOSE = dt.time(16, 0) + +INTRADAY_CACHE_DIR = "data/cache/intraday" +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 get_daily_close(ticker: str, dates: list[str]) -> dict[str, float]: + closes = {} + for date in dates: + path = Path(INTRADAY_CACHE_DIR) / ticker / f"{date}.parquet" + if not path.exists(): + continue + try: + table = pq.read_table(str(path)) + rows = table.to_pydict() + except Exception: + continue + c_list = [] + 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: + c_list.append(float(rows["close"][i] or 0)) + if c_list and c_list[-1] > 0: + closes[date] = c_list[-1] + return closes + + +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]) -> 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 + 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])") + 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']}") + print(f" → {'ALL GATES PASS ✓' if (g1 and g2 and g3) else 'FAIL'}") + + +def main() -> None: + print("=== V42 52-Week High Proximity 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: {m.get('total_trades')} trades, " + f"{m.get('start_date')} → {m.get('end_date')}") + + trade_records = [ + {"ticker": t["ticker"], "date": t["date"][:10], + "r_multiple": float(t["r_multiple_at_exit"])} + for t in trades + if t.get("r_multiple_at_exit") is not None + ] + tickers_needed = sorted(set(r["ticker"] for r in trade_records)) + + min_d = dt.date.fromisoformat(min(r["date"] for r in trade_records)) + max_d = dt.date.fromisoformat(max(r["date"] for r in trade_records)) + # Need 252 prior trading days = ~1 year lookback + start_d = min_d - dt.timedelta(days=400) + all_td = trading_days_between(start_d, max_d) + all_dates = [d.isoformat() for d in all_td] + + print(f"Loading closes for {len(tickers_needed)} tickers across {len(all_dates)} trading days...") + + ticker_closes: dict[str, dict[str, float]] = {} + with concurrent.futures.ThreadPoolExecutor(max_workers=10) as ex: + def _load(ticker: str) -> tuple[str, dict[str, float]]: + return ticker, get_daily_close(ticker, all_dates) + for ticker, closes in ex.map(_load, tickers_needed): + if closes: + ticker_closes[ticker] = closes + + print(f"Loaded: {len(ticker_closes)} / {len(tickers_needed)} tickers\n") + + pct_from_high_vals, range_pos_vals, r_mults = [], [], [] + missing = 0 + + for rec in trade_records: + ticker = rec["ticker"] + date = rec["date"] + r = rec["r_multiple"] + + closes = ticker_closes.get(ticker, {}) + # Get all closes STRICTLY before trade date + prev_closes = [(d, c) for d, c in closes.items() if d < date] + if len(prev_closes) < 50: + missing += 1 + continue + + # Sort by date and take last 252 trading days + prev_closes = sorted(prev_closes, key=lambda x: x[0]) + window = prev_closes[-252:] + prices = [c for _, c in window] + last_close = prices[-1] + high_52w = max(prices) + low_52w = min(prices) + + if high_52w <= 0: + missing += 1 + continue + + pct_from_high = (last_close - high_52w) / high_52w # ≤ 0 + if high_52w == low_52w: + range_pos = 0.5 + else: + range_pos = (last_close - low_52w) / (high_52w - low_52w) # [0, 1] + + pct_from_high_vals.append(pct_from_high) + range_pos_vals.append(range_pos) + r_mults.append(r) + + n_valid = len(r_mults) + print(f"Valid trades: {n_valid} / {len(trade_records)} (missing: {missing})") + if pct_from_high_vals: + avg_pfh = sum(pct_from_high_vals) / len(pct_from_high_vals) + print(f"Mean pct_from_52w_high: {avg_pfh:.1%}") + print(f"Mean range_pos: {sum(range_pos_vals)/len(range_pos_vals):.2f}") + + print("\n" + "=" * 60) + print("GATE RESULTS (G1: |P|≥0.07 & n≥120; G2: avg_R≥0.30R; G3: WR≥5pp)") + + report_feature("pct_from_52w_high (0 = AT high, negative = below)", pct_from_high_vals, r_mults) + report_feature("52w_range_position (0=at_low, 1=at_high)", range_pos_vals, r_mults) + + p_corr = pearson(pct_from_high_vals, range_pos_vals) + print(f"\n ρ(pct_from_high, range_pos) = {p_corr:.3f}") + + print("\n=== Summary ===") + print("Target: G2 ≥ 0.30R. Any value approaching this warrants further investigation.") + + +if __name__ == "__main__": + main() diff --git a/apps/intraday_bt/scripts/diag_orb_bb_pctb.py b/apps/intraday_bt/scripts/diag_orb_bb_pctb.py new file mode 100644 index 0000000..930f8ba --- /dev/null +++ b/apps/intraday_bt/scripts/diag_orb_bb_pctb.py @@ -0,0 +1,275 @@ +""" +V37 Diagnostic: Bollinger Band %B and BB Width pre-breakout signal + +Hypothesis A (compression): stocks with narrowing Bollinger Bands (low BB width) before a +gap-up may be in a coiling phase — explosive breakout, better follow-through. + +Hypothesis B (position): stocks near the upper band (%B > 0.8) have confirmed momentum +and continue higher after the gap. + +Features: + bb_pct_b : (close - lower_band) / (upper_band - lower_band), last prev_bar [0-1+] + bb_width : (upper_band - lower_band) / close — normalized band width (compression) + bb_width_pct: percentile rank of bb_width vs own 60d history (0=tightest compression) + +Source: V24 400d run JSON + daily parquet cache. +""" +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 + +_ET = ZoneInfo("America/New_York") +_MKT_OPEN = dt.time(9, 30) +_MKT_CLOSE = dt.time(16, 0) + +INTRADAY_CACHE_DIR = "data/cache/intraday" +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 _build_daily_bar(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)) + closes.append(float(rows["close"][i] or 0)) + if not closes or closes[-1] <= 0: + return None + return {"date": date, "open": opens[0] if opens else 0, "close": closes[-1]} + + +def load_ticker_bars(ticker: str, all_dates: list[str]) -> list[dict]: + root = Path(INTRADAY_CACHE_DIR) / ticker + if not root.is_dir(): + return [] + bars = [] + for date in all_dates: + p = root / f"{date}.parquet" + if not p.exists(): + continue + bar = _build_daily_bar(p, date) + if bar and bar["close"] > 0: + bars.append(bar) + return sorted(bars, key=lambda b: b["date"]) + + +def compute_bb(bars: list[dict], window: int = 20, num_std: float = 2.0) -> tuple[float, float] | None: + """Returns (pct_b, width) from last `window` bars. bars sorted oldest→newest.""" + if len(bars) < window: + return None + tail = bars[-window:] + closes = [b["close"] for b in tail] + mean = sum(closes) / window + variance = sum((c - mean) ** 2 for c in closes) / window + std = variance ** 0.5 + if std == 0: + return None + upper = mean + num_std * std + lower = mean - num_std * std + last_close = closes[-1] + band_width = upper - lower + if band_width <= 0: + return None + pct_b = (last_close - lower) / band_width + width = band_width / last_close # normalized + return pct_b, width + + +def compute_bb_width_percentile(bars: list[dict], window: int = 20, history: int = 60) -> float | None: + """Percentile rank of current BB width vs own last `history` days.""" + if len(bars) < window + history: + return None + widths = [] + for i in range(history): + end_idx = len(bars) - history + i + 1 + tail = bars[max(0, end_idx - window):end_idx] + if len(tail) < window: + continue + closes = [b["close"] for b in tail] + mean = sum(closes) / len(closes) + std = (sum((c - mean) ** 2 for c in closes) / len(closes)) ** 0.5 + if std == 0: + continue + widths.append((mean + 2 * std - (mean - 2 * std)) / closes[-1]) + if len(widths) < 20: + return None + current = widths[-1] + rank = sum(1 for w in widths if w <= current) / len(widths) + return rank + + +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], obv20: list[float] | None = None) -> None: + n = len(vals) + p = pearson(vals, rs) + ts = tercile_stats(vals, rs) + rho_obv20 = pearson(vals, obv20) if obv20 else None + if not ts or p is None: + print(f" {label}: insufficient data n={n}") + 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 = rho_obv20 is None or abs(rho_obv20) < 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_obv20 is not None: + print(f" G5a: {'PASS' if g5a else 'FAIL'} (|ρ(feature, obv_slope_20)| = {abs(rho_obv20):.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']}") + print(f" → {'ALL GATES PASS ✓' if (g1 and g2 and g3 and g5a) else 'FAIL'}") + + +def main() -> None: + print("=== V37 BB %B and Width Pre-Breakout 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}% DD: {m.get('max_drawdown_pct', 0)*100:.2f}% Sharpe: {m.get('sharpe_ratio', 0):.3f}\n") + + trade_records = [ + {"ticker": t["ticker"], "date": t["date"][:10], + "r_multiple": float(t["r_multiple_at_exit"])} + for t in trades + if t.get("r_multiple_at_exit") is not None + ] + print(f"Trades with r_multiple: {len(trade_records)}") + + tickers_needed = sorted(set(r["ticker"] for r in trade_records)) + min_date = min(r["date"] for r in trade_records) + max_date = max(r["date"] for r in trade_records) + start_cal = (dt.date.fromisoformat(min_date) - dt.timedelta(days=150)).isoformat() + all_dates = [] + d = dt.date.fromisoformat(start_cal) + end_d = dt.date.fromisoformat(max_date) + while d <= end_d: + all_dates.append(d.isoformat()) + d += dt.timedelta(days=1) + + print(f"Loading bars for {len(tickers_needed)} tickers ({start_cal} → {max_date})...") + ticker_bars: dict[str, list[dict]] = {} + with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex: + def _load(ticker: str) -> tuple[str, list[dict]]: + return ticker, load_ticker_bars(ticker, all_dates) + for ticker, bars in ex.map(_load, tickers_needed): + if bars: + ticker_bars[ticker] = bars + print(f"Loaded: {len(ticker_bars)} / {len(tickers_needed)} tickers\n") + + pctb_vals, width_vals, width_pct_vals, obv20_vals, r_mults = [], [], [], [], [] + missing_bb, missing_wp = 0, 0 + + from libs.intraday.features import compute_obv_slope_approx + for rec in trade_records: + ticker = rec["ticker"] + date = rec["date"] + r = rec["r_multiple"] + + bars = ticker_bars.get(ticker, []) + prev_bars = [b for b in bars if b["date"] < date] + + bb = compute_bb(prev_bars, window=20) if len(prev_bars) >= 20 else None + if bb is None: + missing_bb += 1 + continue + + pct_b, width = bb + wp = compute_bb_width_percentile(prev_bars, window=20, history=60) + if wp is None: + missing_wp += 1 + + obv20 = compute_obv_slope_approx(prev_bars, lookback=20) if len(prev_bars) >= 22 else None + + pctb_vals.append(pct_b) + width_vals.append(width) + if wp is not None: + width_pct_vals.append(wp) + obv20_vals.append(obv20 if obv20 is not None else 0.0) + r_mults.append(r) + + n_valid = len(r_mults) + print(f"Valid trades (BB computable): {n_valid} / {len(trade_records)}") + print(f"Missing BB: {missing_bb} Missing width_pct: {missing_wp} Valid width_pct: {len(width_pct_vals)}") + + print("\n" + "=" * 60) + print("GATE RESULTS (G1: |P|≥0.07 & n≥120; G2: avg_R≥0.30R; G3: WR≥5pp; G5a: ρ<0.70 vs obv_slope_20)") + + obv20_aligned_pctb = obv20_vals[:len(pctb_vals)] + report_feature("bb_pct_b", pctb_vals, r_mults, obv20_aligned_pctb) + report_feature("bb_width (compression)", width_vals, r_mults, obv20_aligned_pctb) + if len(width_pct_vals) >= 120: + report_feature("bb_width_percentile_60d", width_pct_vals, r_mults[:len(width_pct_vals)], obv20_aligned_pctb[:len(width_pct_vals)]) + else: + print(f"\n bb_width_percentile_60d: insufficient n={len(width_pct_vals)} (need 120)") + + p_pctb_width = pearson(pctb_vals, width_vals) + print(f"\n Inter-feature: ρ(bb_pct_b, bb_width) = {p_pctb_width:.3f}") + mean_pctb = sum(pctb_vals)/len(pctb_vals) if pctb_vals else 0 + print(f" %B distribution: mean={mean_pctb:.2f} (0=lower, 0.5=mid, 1=upper, >1=above band)") + + print("\n=== Summary ===") + print(f"G2 target: ≥ 0.30R. OBV-slope = 0.394R (only axis that cleared).") + + +if __name__ == "__main__": + main() diff --git a/apps/intraday_bt/scripts/diag_orb_breadth.py b/apps/intraday_bt/scripts/diag_orb_breadth.py new file mode 100644 index 0000000..041d19b --- /dev/null +++ b/apps/intraday_bt/scripts/diag_orb_breadth.py @@ -0,0 +1,283 @@ +""" +V40 Diagnostic: Prior-Day Market Breadth Signal + +Hypothesis: On days when broad market breadth was positive the day before an ORB gap-up, +follow-through should be better because the gap is "with the market" rather than isolated. + +Features: + breadth_up_pct : fraction of universe that closed UP the prior day [0-1] + breadth_adv_dec : (advancers - decliners) / total — net breadth [-1 to +1] + +This is a DAY-LEVEL signal (same value for all stocks on same day). +Unlike QQQ gap (already in V24) which is a single-stock measure, breadth captures +the DISTRIBUTION of market participation. + +Source: daily bars for full midlarge universe, computed from parquet cache. +""" +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 +import yaml + +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) +_MKT_CLOSE = dt.time(16, 0) + +INTRADAY_CACHE_DIR = "data/cache/intraday" +UNIVERSE_FILE = "configs/symbols_midlarge_snapshot_exact.yaml" +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 get_daily_close(ticker: str, date: str) -> float | None: + path = Path(INTRADAY_CACHE_DIR) / ticker / f"{date}.parquet" + if not path.exists(): + return None + try: + table = pq.read_table(str(path)) + rows = table.to_pydict() + except Exception: + return None + closes = [] + 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: + closes.append(float(rows["close"][i] or 0)) + return closes[-1] if closes and closes[-1] > 0 else None + + +def compute_breadth(universe: list[str], date: str, prev_date: str) -> dict | None: + """Compute breadth on `date` using closes from `prev_date` and day before.""" + def _load(ticker: str) -> tuple[str, float | None, float | None]: + return ticker, get_daily_close(ticker, prev_date), get_daily_close(ticker, _two_days_prior(date, prev_date)) + # Simplified: just compare prev_date vs prev-prev-date close + return None + + +def get_two_closes(ticker: str, dates: list[str]) -> dict[str, float]: + """Get close prices for a list of dates.""" + result = {} + for d in dates: + c = get_daily_close(ticker, d) + if c is not None: + result[d] = c + 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]) -> 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 + 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])") + 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']}") + print(f" → {'ALL GATES PASS ✓' if (g1 and g2 and g3) else 'FAIL'}") + + +def main() -> None: + print("=== V40 Prior-Day Market Breadth 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: {m.get('total_trades')} trades, " + f"{m.get('start_date')} → {m.get('end_date')}") + + trade_records = [ + {"ticker": t["ticker"], "date": t["date"][:10], + "r_multiple": float(t["r_multiple_at_exit"])} + for t in trades + if t.get("r_multiple_at_exit") is not None + ] + + with open(UNIVERSE_FILE) as f: + udata = yaml.safe_load(f) + universe = udata.get("symbols", udata) if isinstance(udata, dict) else list(udata) + print(f"Universe: {len(universe)} tickers") + + # Get all unique trade dates and their prior trading days + all_dates_set = set(r["date"] for r in trade_records) + min_d = dt.date.fromisoformat(min(all_dates_set)) + max_d = dt.date.fromisoformat(max(all_dates_set)) + all_td = trading_days_between(min_d - dt.timedelta(days=30), max_d) + td_list = [d.isoformat() for d in all_td] + + # Map each trade date to its prior trading day + prior_day_map: dict[str, str] = {} + for i, d in enumerate(td_list): + if d in all_dates_set and i > 0: + prior_day_map[d] = td_list[i - 1] + + print(f"Trade dates with prior day: {len(prior_day_map)} / {len(all_dates_set)}") + + # For each unique prior day, compute market breadth + prior_days_needed = sorted(set(prior_day_map.values())) + print(f"Loading daily closes for universe across {len(prior_days_needed)} prior days...") + + # Build date range for loading + all_needed_dates: list[str] = [] + for pd in prior_days_needed: + all_needed_dates.append(pd) + + def _load_ticker(ticker: str) -> tuple[str, dict[str, float]]: + return ticker, get_two_closes(ticker, all_needed_dates) + + ticker_closes: dict[str, dict[str, float]] = {} + with concurrent.futures.ThreadPoolExecutor(max_workers=12) as ex: + for ticker, closes in ex.map(_load_ticker, universe): + if closes: + ticker_closes[ticker] = closes + + # Compute breadth per prior day + # Breadth = fraction of universe UP on that day vs the day before + # We need TWO prior days: prior_day and prior_prior_day + all_dates_calendar = [] + d = min_d - dt.timedelta(days=40) + while d <= max_d: + all_dates_calendar.append(d.isoformat()) + d += dt.timedelta(days=1) + + # Get prior-prior days + prior_prior_map: dict[str, str] = {} + for i, d in enumerate(td_list): + if d in prior_days_needed and i > 0: + prior_prior_map[d] = td_list[i - 1] + + # Load prior-prior day closes + prior_prior_days = sorted(set(prior_prior_map.values())) + def _load_ticker2(ticker: str) -> tuple[str, dict[str, float]]: + return ticker, get_two_closes(ticker, prior_prior_days) + + ticker_pp_closes: dict[str, dict[str, float]] = {} + with concurrent.futures.ThreadPoolExecutor(max_workers=12) as ex: + for ticker, closes in ex.map(_load_ticker2, universe): + if closes: + ticker_pp_closes[ticker] = closes + + # Compute breadth for each prior day + breadth_by_day: dict[str, dict] = {} + for pd in prior_days_needed: + ppd = prior_prior_map.get(pd) + if ppd is None: + continue + adv, dec, total = 0, 0, 0 + for ticker in universe: + c_pd = ticker_closes.get(ticker, {}).get(pd) + c_ppd = ticker_pp_closes.get(ticker, {}).get(ppd) + if c_pd is None or c_ppd is None or c_ppd <= 0: + continue + total += 1 + if c_pd > c_ppd: + adv += 1 + elif c_pd < c_ppd: + dec += 1 + if total >= 50: + breadth_by_day[pd] = { + "up_pct": adv / total, + "adv_dec": (adv - dec) / total, + "n": total, + } + + print(f"Breadth computed for {len(breadth_by_day)} prior days") + if breadth_by_day: + sample = list(breadth_by_day.values())[:5] + up_pcts = [v["up_pct"] for v in breadth_by_day.values()] + print(f"Breadth range: up_pct {min(up_pcts):.1%} - {max(up_pcts):.1%} mean={sum(up_pcts)/len(up_pcts):.1%}") + + # Match to trades + up_pct_vals, adv_dec_vals, r_mults = [], [], [] + missing = 0 + for rec in trade_records: + pd = prior_day_map.get(rec["date"]) + if pd is None: + missing += 1 + continue + bdata = breadth_by_day.get(pd) + if bdata is None: + missing += 1 + continue + up_pct_vals.append(bdata["up_pct"]) + adv_dec_vals.append(bdata["adv_dec"]) + r_mults.append(rec["r_multiple"]) + + n_valid = len(r_mults) + print(f"\nValid trades: {n_valid} / {len(trade_records)} (missing: {missing})") + + print("\n" + "=" * 60) + print("GATE RESULTS (G1: |P|≥0.07 & n≥120; G2: avg_R≥0.30R; G3: WR≥5pp)") + print("NOTE: Day-level signal — effective n is unique trading days, not trades.") + unique_days = len(set(r["date"] for r in trade_records if prior_day_map.get(r["date"]) in breadth_by_day)) + print(f"Unique trade days: {unique_days}\n") + + report_feature("breadth_up_pct (prior day)", up_pct_vals, r_mults) + report_feature("breadth_adv_dec (adv-dec/total)", adv_dec_vals, r_mults) + + p_up_adv = pearson(up_pct_vals, adv_dec_vals) + print(f"\n ρ(up_pct, adv_dec) = {p_up_adv:.3f}") + + print("\n=== Summary ===") + print("V24's QQQ gap filter already captures macro regime. This tests BREADTH depth.") + + +if __name__ == "__main__": + main() diff --git a/apps/intraday_bt/scripts/diag_orb_obv_acceleration.py b/apps/intraday_bt/scripts/diag_orb_obv_acceleration.py new file mode 100644 index 0000000..a95a81c --- /dev/null +++ b/apps/intraday_bt/scripts/diag_orb_obv_acceleration.py @@ -0,0 +1,283 @@ +""" +V34 Diagnostic: OBV Slope Acceleration + +Tests whether SHORT-TERM OBV accumulation (5-day slope) provides signal orthogonal to +V24's existing 20-day OBV slope. Hypothesis: the most recent accumulation (last 5 days) +before a gap event captures fresher institutional positioning than the 20-day average. + +Features: + obv_slope_5 : 5-day OBV accumulation slope (same formula as obv_slope_20, shorter window) + obv_slope_accel: obv_slope_5 - obv_slope_20 (positive = recent acceleration of accumulation) + +The key test: does obv_slope_5 pass G5a (|ρ(slope_5, slope_20)| < 0.70)? +If highly correlated (ρ > 0.70), it's redundant with V24's signal. If orthogonal, it could +complement V24's score. + +Context: V24 already uses obv_slope_20 at weight=0.05. V25-V33 all failed. +This is the final attempt using existing data (obv_slope_5 uses same formula, different window). +""" +from __future__ import annotations + +import concurrent.futures +import datetime as dt +import os +import sys +from pathlib import Path + +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.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 + +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" +LOOKBACK_DAYS = 400 + +_ET = ZoneInfo("America/New_York") +_MKT_OPEN = dt.time(9, 30) +_MKT_CLOSE = dt.time(16, 0) + + +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 _build_daily_bar(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(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]]]: + 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 + + +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], obv20: list[float]) -> tuple[bool, bool, bool]: + n = len(vals) + p = pearson(vals, rs) + ts = tercile_stats(vals, rs) + rho_obv20 = pearson(vals, obv20) if obv20 else None + if not ts or p is None: + print(f" {label}: insufficient data") + return False, False, False + 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 = rho_obv20 is None or abs(rho_obv20) < 0.70 + print(f"\n [{label}] n={n} Pearson={p:+.3f}") + print(f" G1: {'PASS' if g1 else 'FAIL'} (|{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])") + print(f" G5a: {'PASS' if g5a else 'FAIL'} (|ρ(feature, obv_slope_20)| = {abs(rho_obv20):.3f} [threshold 0.70])" if rho_obv20 is not None else " G5a: N/A") + 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_pass = g1 and g2 and g3 and g5a + print(f" → {'ALL GATES PASS ✓' if all_pass else 'FAIL'}") + return g1, g2, g3 + + +def main() -> None: + print("=== V34 OBV Slope Acceleration Diagnostic ===\n") + + with open(V24_CONFIG) as f: + raw = yaml.safe_load(f) + params = ORBStrategyParams(**raw["orb_strategy"]) + + today = dt.date(2026, 4, 21) + all_td = trading_days_between(today - dt.timedelta(days=700), today) + trading_days_list = [d.isoformat() for d in all_td[-LOOKBACK_DAYS:]] + print(f"Window: {trading_days_list[0]} → {trading_days_list[-1]} ({LOOKBACK_DAYS} trading days)") + + extended_td = [d.isoformat() for d in all_td[-(LOOKBACK_DAYS + 35):]] + 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) + + 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") + + print(f"Building daily bars ({len(needed_dates)} calendar days)...") + daily_bars = build_daily_bars(universe, needed_dates) + print(f"Built: {len(daily_bars)} tickers") + print("Computing enrichment...") + enrichment = enrich_daily_bars(daily_bars, trading_days_list) + + 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, + ) + print(f"Pre-screened: {sum(len(v) for v in candidates.values())} ticker-days") + all_intraday = load_intraday_bulk(candidates) + + print("Running 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)}\n") + + sorted_daily: dict[str, list[dict]] = { + t: sorted(bars, key=lambda b: b["date"]) for t, bars in daily_bars.items() + } + + # Compute per-trade features + slope5_vals, slope20_vals, accel_vals, r_mults = [], [], [], [] + missing5, missing20, missingaccel = 0, 0, 0 + + for trade in trades_with_r: + ticker = trade.ticker + date = str(trade.date)[:10] + r = float(trade.r_multiple_at_exit) + + bars_t = sorted_daily.get(ticker, []) + prev_bars = [b for b in bars_t if b["date"][:10] < date] + + s5 = compute_obv_slope_approx(prev_bars, lookback=5) if len(prev_bars) >= 7 else None + s20 = compute_obv_slope_approx(prev_bars, lookback=20) if len(prev_bars) >= 22 else None + + if s5 is None: + missing5 += 1 + if s20 is None: + missing20 += 1 + if s5 is None or s20 is None: + missingaccel += 1 + continue + + slope5_vals.append(s5) + slope20_vals.append(s20) + accel_vals.append(s5 - s20) + r_mults.append(r) + + n_valid = len(r_mults) + print(f"Valid trades (both slope5 & slope20): {n_valid} / {len(trades_with_r)}") + print(f"Missing slope5: {missing5} slope20: {missing20}") + + print("\n" + "=" * 60) + print("GATE RESULTS (G1: |P|≥0.07, n≥120; G2: avg_R≥0.30R; G3: WR≥5pp; G5a: ρ<0.70 vs obv_slope_20)") + + report_feature("obv_slope_5", slope5_vals, r_mults, slope20_vals) + report_feature("obv_slope_accel (slope5 - slope20)", accel_vals, r_mults, slope20_vals) + + p_5_20 = pearson(slope5_vals, slope20_vals) + print(f"\n ρ(slope_5, slope_20) = {p_5_20:.3f} (G5a threshold: 0.70)") + + print("\n=== Summary ===") + if p_5_20 is not None and abs(p_5_20) >= 0.70: + print("G5a: obv_slope_5 is REDUNDANT with obv_slope_20 (ρ ≥ 0.70). Cannot add independent signal.") + print("Conclusion: V24's obv_slope_20 captures the full OBV axis. No improvement possible via window change.") + else: + print("G5a: obv_slope_5 has orthogonal component vs obv_slope_20. Check individual feature gates above.") + + +if __name__ == "__main__": + main() diff --git a/apps/intraday_bt/scripts/diag_orb_premarket_accel.py b/apps/intraday_bt/scripts/diag_orb_premarket_accel.py new file mode 100644 index 0000000..6b09461 --- /dev/null +++ b/apps/intraday_bt/scripts/diag_orb_premarket_accel.py @@ -0,0 +1,206 @@ +""" +V39 Diagnostic: Pre-Market Volume Acceleration Signal + +Hypothesis: The QUALITY of pre-market activity matters beyond its magnitude. +A stock where pre-market volume is BUILDING into the open (acceleration) suggests +fresh sustained demand vs. a single early spike that fades. + +The existing weight_premarket_dollar_vol captures the QUANTITY of pre-market activity. +This tests the SHAPE — specifically whether volume is accelerating toward the open. + +Features (computed from pre-market intraday bars 4:00-9:29 ET): + pm_accel_ratio : last_30min_vol / first_half_vol — >1.0 means volume building + pm_vol_per_bar : premarket_dollar_vol / pm_bar_count — density measure + pm_hold_ratio : last_premarket_close / session_premarket_high — is price holding near high? + +Context: The V24 400d run has trade dates with full intraday data in cache. +""" +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 + +_ET = ZoneInfo("America/New_York") +_PRE_OPEN = dt.time(4, 0) +_MKT_OPEN = dt.time(9, 30) + +INTRADAY_CACHE_DIR = "data/cache/intraday" +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_premarket_bars(ticker: str, date: str) -> list[dict]: + """Load all pre-market bars (4:00-9:29 ET) for a ticker on a date.""" + path = Path(INTRADAY_CACHE_DIR) / ticker / f"{date}.parquet" + if not path.exists(): + return [] + try: + table = pq.read_table(str(path)) + rows = table.to_pydict() + except Exception: + return [] + bars = [] + for i, ts_raw in enumerate(rows.get("timestamp", [])): + try: + ts = _parse_ts(ts_raw) + except Exception: + continue + if _PRE_OPEN <= 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) + v = float(rows["volume"][i] or 0) + if v > 0 and c > 0: + bars.append({ + "time": ts.time(), "open": o, "high": h, + "low": lo, "close": c, "volume": v, "dv": c * v, + }) + return bars + + +def compute_pm_features(pm_bars: list[dict]) -> dict | None: + if len(pm_bars) < 4: + return None + total_dv = sum(b["dv"] for b in pm_bars) + if total_dv <= 0: + return None + + n = len(pm_bars) + # Acceleration: last 30 min (last ~6 bars) vs first half + last_6 = pm_bars[-6:] + first_half = pm_bars[:n // 2] + last_vol = sum(b["dv"] for b in last_6) + first_vol = sum(b["dv"] for b in first_half) + if first_vol <= 0: + accel_ratio = None + else: + accel_ratio = last_vol / first_vol + + # Price hold: last close vs session high + pm_high = max(b["high"] for b in pm_bars) + last_close = pm_bars[-1]["close"] + if pm_high <= 0: + hold_ratio = None + else: + hold_ratio = last_close / pm_high + + return { + "accel_ratio": accel_ratio, + "hold_ratio": hold_ratio, + "bar_count": n, + } + + +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]) -> 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 + 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])") + 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']}") + print(f" → {'ALL GATES PASS ✓' if (g1 and g2 and g3) else 'FAIL'}") + + +def main() -> None: + print("=== V39 Pre-Market Volume Acceleration 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: {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}% Sharpe: {m.get('sharpe_ratio', 0):.3f}\n") + + trade_records = [ + {"ticker": t["ticker"], "date": t["date"][:10], + "r_multiple": float(t["r_multiple_at_exit"])} + for t in trades + if t.get("r_multiple_at_exit") is not None + ] + print(f"Trades with r_multiple: {len(trade_records)}") + + accel_vals, hold_vals, r_mults = [], [], [] + missing = 0 + + for rec in trade_records: + pm_bars = load_premarket_bars(rec["ticker"], rec["date"]) + feats = compute_pm_features(pm_bars) + if feats is None or feats["accel_ratio"] is None or feats["hold_ratio"] is None: + missing += 1 + continue + accel_vals.append(feats["accel_ratio"]) + hold_vals.append(feats["hold_ratio"]) + r_mults.append(rec["r_multiple"]) + + n_valid = len(r_mults) + print(f"Valid trades (pm data): {n_valid} / {len(trade_records)} (missing: {missing})") + + print("\n" + "=" * 60) + print("GATE RESULTS (G1: |P|≥0.07 & n≥120; G2: avg_R≥0.30R; G3: WR≥5pp)") + + report_feature("pm_accel_ratio (last_30min_dv / first_half_dv)", accel_vals, r_mults) + report_feature("pm_hold_ratio (last_close / pm_high)", hold_vals, r_mults) + + print("\n=== Summary ===") + print("OBV-slope 20d: Pearson=+0.235, G2=0.394R — the only passing axis.") + print("This tests INTRADAY pre-market shape, not captured by existing premarket_dollar_vol.") + + +if __name__ == "__main__": + main() diff --git a/apps/intraday_bt/scripts/diag_orb_remaining.py b/apps/intraday_bt/scripts/diag_orb_remaining.py new file mode 100644 index 0000000..bbc1c24 --- /dev/null +++ b/apps/intraday_bt/scripts/diag_orb_remaining.py @@ -0,0 +1,263 @@ +""" +V38 Diagnostic: Remaining unexplored daily feature axes + +Tests 4 features not yet explored against V24 trade set (400d, n=304): + + 1. dollar_vol_trend : avg_dollar_vol_10d / avg_dollar_vol_30d — is dollar volume + ACCELERATING? If yes, recent activity surge above baseline. + 2. sleep_streak : consecutive prior days with |daily_return| < 1.5% — + "coiled spring" hypothesis: long quiet period before explosive gap. + 3. prior_day_return : yesterday's close vs. day-before close (1-day momentum/extension check). + Hypothesis: stocks that were flat/down yesterday have better ORB + follow-through (less extension) than stocks already up yesterday. + 4. vol_trend_ratio : avg_volume_5d / avg_volume_20d — is volume accelerating (short-term)? +""" +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 + +_ET = ZoneInfo("America/New_York") +_MKT_OPEN = dt.time(9, 30) +_MKT_CLOSE = dt.time(16, 0) + +INTRADAY_CACHE_DIR = "data/cache/intraday" +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 _build_daily_bar(path: Path, date: str) -> dict | None: + try: + table = pq.read_table(str(path)) + rows = table.to_pydict() + except Exception: + return None + opens, 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)) + closes.append(float(rows["close"][i] or 0)) + vols.append(float(rows["volume"][i] or 0)) + if not closes or closes[-1] <= 0: + return None + close = closes[-1] + vol = sum(vols) + return {"date": date, "open": opens[0] if opens else 0, "close": close, + "volume": vol, "dollar_vol": close * vol} + + +def load_ticker_bars(ticker: str, all_dates: list[str]) -> list[dict]: + root = Path(INTRADAY_CACHE_DIR) / ticker + if not root.is_dir(): + return [] + bars = [] + for date in all_dates: + p = root / f"{date}.parquet" + if not p.exists(): + continue + bar = _build_daily_bar(p, date) + if bar and bar["close"] > 0: + bars.append(bar) + return sorted(bars, key=lambda b: b["date"]) + + +def _avg(vals: list[float]) -> float: + return sum(vals) / len(vals) if vals else 0.0 + + +def compute_dollar_vol_trend(prev_bars: list[dict]) -> float | None: + if len(prev_bars) < 30: + return None + dv_5 = _avg([b["dollar_vol"] for b in prev_bars[-10:]]) + dv_30 = _avg([b["dollar_vol"] for b in prev_bars[-30:]]) + if dv_30 <= 0: + return None + return dv_5 / dv_30 + + +def compute_sleep_streak(prev_bars: list[dict], threshold: float = 0.015) -> float | None: + """Count consecutive prior days with |return| < threshold.""" + if len(prev_bars) < 2: + return None + streak = 0 + for i in range(len(prev_bars) - 1, 0, -1): + c = prev_bars[i]["close"] + pc = prev_bars[i - 1]["close"] + if pc <= 0: + break + ret = abs((c - pc) / pc) + if ret < threshold: + streak += 1 + else: + break + return float(streak) + + +def compute_prior_day_return(prev_bars: list[dict]) -> float | None: + if len(prev_bars) < 2: + return None + c = prev_bars[-1]["close"] + pc = prev_bars[-2]["close"] + if pc <= 0: + return None + return (c - pc) / pc + + +def compute_vol_trend_ratio(prev_bars: list[dict]) -> float | None: + if len(prev_bars) < 20: + return None + v5 = _avg([b["volume"] for b in prev_bars[-5:]]) + v20 = _avg([b["volume"] for b in prev_bars[-20:]]) + if v20 <= 0: + return None + return v5 / v20 + + +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]) -> 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 + 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])") + 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']}") + print(f" → {'ALL GATES PASS ✓' if (g1 and g2 and g3) else 'FAIL'}") + + +def main() -> None: + print("=== V38 Remaining Daily Features 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}% DD: {m.get('max_drawdown_pct', 0)*100:.2f}% Sharpe: {m.get('sharpe_ratio', 0):.3f}\n") + + trade_records = [ + {"ticker": t["ticker"], "date": t["date"][:10], + "r_multiple": float(t["r_multiple_at_exit"])} + for t in trades + if t.get("r_multiple_at_exit") is not None + ] + print(f"Trades with r_multiple: {len(trade_records)}") + + tickers_needed = sorted(set(r["ticker"] for r in trade_records)) + min_date = min(r["date"] for r in trade_records) + max_date = max(r["date"] for r in trade_records) + start_cal = (dt.date.fromisoformat(min_date) - dt.timedelta(days=120)).isoformat() + all_dates = [] + d = dt.date.fromisoformat(start_cal) + end_d = dt.date.fromisoformat(max_date) + while d <= end_d: + all_dates.append(d.isoformat()) + d += dt.timedelta(days=1) + + print(f"Loading bars for {len(tickers_needed)} tickers...") + ticker_bars: dict[str, list[dict]] = {} + with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex: + def _load(ticker: str) -> tuple[str, list[dict]]: + return ticker, load_ticker_bars(ticker, all_dates) + for ticker, bars in ex.map(_load, tickers_needed): + if bars: + ticker_bars[ticker] = bars + print(f"Loaded: {len(ticker_bars)} / {len(tickers_needed)} tickers\n") + + dv_trend_all, sleep_all, prior_ret_all, vol_trend_all, r_mults_all = [], [], [], [], [] + + for rec in trade_records: + ticker = rec["ticker"] + date = rec["date"] + r = rec["r_multiple"] + bars = ticker_bars.get(ticker, []) + prev = [b for b in bars if b["date"] < date] + + dv = compute_dollar_vol_trend(prev) + sl = compute_sleep_streak(prev) + pr = compute_prior_day_return(prev) + vt = compute_vol_trend_ratio(prev) + + if dv is None or sl is None or pr is None or vt is None: + continue + + dv_trend_all.append(dv) + sleep_all.append(sl) + prior_ret_all.append(pr) + vol_trend_all.append(vt) + r_mults_all.append(r) + + print(f"Valid trades (all 4 features): {len(r_mults_all)} / {len(trade_records)}") + + print("\n" + "=" * 60) + print("GATE RESULTS (G1: |P|≥0.07 & n≥120; G2: avg_R≥0.30R; G3: WR≥5pp)") + + report_feature("dollar_vol_trend (dvol10d/dvol30d)", dv_trend_all, r_mults_all) + report_feature("sleep_streak (consec. low-vol days)", sleep_all, r_mults_all) + report_feature("prior_day_return", prior_ret_all, r_mults_all) + report_feature("vol_trend_ratio (vol5d/vol20d)", vol_trend_all, r_mults_all) + + print("\n=== Benchmark ===") + print("OBV-slope 20d: Pearson=+0.235, G2=0.394R — only passing signal.") + print("Any G2 ≥ 0.30R here → advance to wiring. Otherwise → V24 is terminal for daily OHLCV axis.") + + +if __name__ == "__main__": + main() diff --git a/apps/intraday_bt/scripts/diag_orb_rsi14.py b/apps/intraday_bt/scripts/diag_orb_rsi14.py new file mode 100644 index 0000000..da25e60 --- /dev/null +++ b/apps/intraday_bt/scripts/diag_orb_rsi14.py @@ -0,0 +1,263 @@ +""" +V36 Diagnostic: RSI-14 Pre-Breakout Signal + +Hypothesis: RSI-14 in the days before an ORB gap-up predicts follow-through quality. +Two competing sub-hypotheses: + (A) Momentum: high RSI (>60) = confirmed uptrend, clean breakout. Positive Pearson. + (B) Mean-reversion: high RSI = overbought, gap sells off. Negative Pearson. + (C) Sweet spot: moderate RSI (40-60) = breakout from accumulation, not extended. + +Features: + rsi_14 : raw RSI-14 [0, 100] on last prev_bar + rsi_momentum: (RSI - 50) / 50, centered at midline [-1, +1] + +Source: V24 400d run JSON, daily bars from parquet cache. +Uses Cutler's RSI (simple average, same formula as libs/features/market_features.py). +""" +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 + +_ET = ZoneInfo("America/New_York") +_MKT_OPEN = dt.time(9, 30) +_MKT_CLOSE = dt.time(16, 0) + +INTRADAY_CACHE_DIR = "data/cache/intraday" +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 _build_daily_bar_from_parquet(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 or closes[-1] <= 0: + return None + return { + "date": date, "open": opens[0], "high": max(highs), + "low": min(lows), "close": closes[-1], "volume": sum(vols), + } + + +def load_ticker_daily_bars(ticker: str, needed_dates: list[str]) -> list[dict]: + root = Path(INTRADAY_CACHE_DIR) / ticker + if not root.is_dir(): + return [] + bars = [] + for date in needed_dates: + p = root / f"{date}.parquet" + if not p.exists(): + continue + bar = _build_daily_bar_from_parquet(p, date) + if bar: + bars.append(bar) + return sorted(bars, key=lambda b: b["date"]) + + +def compute_rsi_14(bars: list[dict], period: int = 14) -> float | None: + """Cutler's RSI from daily close prices. bars sorted oldest→newest.""" + if len(bars) < period + 2: + return None + tail = bars[-(period + 1):] + gains, losses = [], [] + for i in range(period): + change = tail[i + 1]["close"] - tail[i]["close"] + if change >= 0: + gains.append(change) + losses.append(0.0) + else: + gains.append(0.0) + losses.append(abs(change)) + avg_gain = sum(gains) / period + avg_loss = sum(losses) / period + if avg_loss == 0: + return 100.0 + return 100.0 - (100.0 / (1.0 + avg_gain / avg_loss)) + + +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], obv20: list[float] | None = None) -> None: + n = len(vals) + p = pearson(vals, rs) + ts = tercile_stats(vals, rs) + rho_obv20 = pearson(vals, obv20) if obv20 else None + if not ts or p is None: + print(f" {label}: 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 = rho_obv20 is None or abs(rho_obv20) < 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} {'≥' if n>=120 else '<'} 120)") + 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_obv20 is not None: + print(f" G5a: {'PASS' if g5a else 'FAIL'} (|ρ(feature, obv_slope_20)| = {abs(rho_obv20):.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_pass = g1 and g2 and g3 and g5a + print(f" → {'ALL GATES PASS ✓' if all_pass else 'FAIL'}") + + +def main() -> None: + print("=== V36 RSI-14 Pre-Breakout 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") + + trade_records = [ + {"ticker": t["ticker"], "date": t["date"][:10], + "r_multiple": float(t["r_multiple_at_exit"])} + for t in trades + if t.get("r_multiple_at_exit") is not None + ] + print(f"Trades with r_multiple: {len(trade_records)}") + + # Gather all unique tickers and dates needed (extended window for RSI lookback) + tickers_needed = sorted(set(r["ticker"] for r in trade_records)) + dates_by_ticker: dict[str, set[str]] = {t: set() for t in tickers_needed} + for rec in trade_records: + dates_by_ticker[rec["ticker"]].add(rec["date"]) + + print(f"Loading daily bars for {len(tickers_needed)} tickers...") + + # Determine needed calendar range per ticker (need 30+ prior trading days) + import datetime as dt + min_date = min(r["date"] for r in trade_records) + max_date = max(r["date"] for r in trade_records) + # Build full calendar range with buffer + start_cal = (dt.date.fromisoformat(min_date) - dt.timedelta(days=90)).isoformat() + all_dates = [] + d = dt.date.fromisoformat(start_cal) + end_d = dt.date.fromisoformat(max_date) + while d <= end_d: + all_dates.append(d.isoformat()) + d += dt.timedelta(days=1) + + def _load(ticker: str) -> tuple[str, list[dict]]: + return ticker, load_ticker_daily_bars(ticker, all_dates) + + ticker_bars: dict[str, list[dict]] = {} + with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex: + for ticker, bars in ex.map(_load, tickers_needed): + if bars: + ticker_bars[ticker] = bars + + print(f"Loaded bars for {len(ticker_bars)} / {len(tickers_needed)} tickers\n") + + # Compute features per trade + rsi_vals, rsi_mom_vals, obv20_vals, r_mults = [], [], [], [] + missing = 0 + + for rec in trade_records: + ticker = rec["ticker"] + date = rec["date"] + r = rec["r_multiple"] + + bars = ticker_bars.get(ticker, []) + prev_bars = [b for b in bars if b["date"] < date] + + rsi = compute_rsi_14(prev_bars, period=14) if len(prev_bars) >= 16 else None + if rsi is None: + missing += 1 + continue + + # OBV slope 20d for redundancy check + from libs.intraday.features import compute_obv_slope_approx + obv20 = compute_obv_slope_approx(prev_bars, lookback=20) if len(prev_bars) >= 22 else None + + rsi_vals.append(rsi) + rsi_mom_vals.append((rsi - 50.0) / 50.0) + obv20_vals.append(obv20 if obv20 is not None else 0.0) + r_mults.append(r) + + n_valid = len(r_mults) + print(f"Valid trades (RSI computable): {n_valid} / {len(trade_records)} (missing: {missing})") + + print("\n" + "=" * 60) + print("GATE RESULTS (G1: |P|≥0.07 & n≥120; G2: avg_R≥0.30R; G3: WR≥5pp; G5a: ρ<0.70 vs obv_slope_20)") + + report_feature("rsi_14", rsi_vals, r_mults, obv20_vals) + report_feature("rsi_momentum (RSI-50)/50", rsi_mom_vals, r_mults, obv20_vals) + + # Distribution stats + mean_rsi = sum(rsi_vals) / len(rsi_vals) if rsi_vals else 0 + print(f"\n RSI distribution: mean={mean_rsi:.1f} min={min(rsi_vals):.1f} max={max(rsi_vals):.1f}") + + print("\n=== Summary ===") + p = pearson(rsi_vals, r_mults) + if p is not None: + direction = "Momentum (high RSI = better)" if p > 0 else "Mean-reversion (low RSI = better)" + print(f"Signal direction: {direction} (Pearson={p:+.3f})") + print("Context: V24's only passing axis: OBV-slope G2=0.394R. Target: G2 ≥ 0.30R.") + + +if __name__ == "__main__": + main() diff --git a/configs/intraday/strategies/orb_gainers_v41_obv_filter.yaml b/configs/intraday/strategies/orb_gainers_v41_obv_filter.yaml new file mode 100644 index 0000000..bdbd8b1 --- /dev/null +++ b/configs/intraday/strategies/orb_gainers_v41_obv_filter.yaml @@ -0,0 +1,115 @@ +_meta: + id: 109 + name: "ORB Gainers V41 OBV Filter" + status: candidate + live_readiness: experimental + parent: orb_gainers_v24_quality_overlay + description: > + V24 → V41: converts the OBV-slope weight into a HARD FILTER. + + V24 uses OBV-slope as a continuous scoring weight (weight_obv_slope: 0.05). + V41 keeps the weight AND adds min_obv_slope_20d: 0.0 to filter out candidates + in active distribution (negative OBV slope = institutions selling net). + + Hypothesis: stocks with OBV_slope_20 < 0 represent institutional distribution. + Even if they rank well on rvol/gap/premarket, entering a distributing stock on + a gap-up is a lower-quality setup. Filtering them out improves trade quality + at the cost of some volume reduction. + + The filter is additive to (not replacing) the weight. + All other V24 params unchanged. + + Gates (200d): Return ≥ +98.8% (V24 +94.8% + 4pp), DD ≥ -11.79%, Sharpe ≥ 2.93 + Gates (400d): Return ≥ +166.1% (V24 +162.1% + 4pp), DD ≥ -14.20%, Sharpe ≥ 2.52 + +strategy_mode: orb + +orb_strategy: + engine_family: gainers_leader + live_readiness: experimental + orb_minutes: 5 + sim_bar_minutes: 5 + + entry_direction: long_only + order_timeout_minutes: 45 + + allow_doji_breakout: true + allow_red_to_green_breakout: true + + min_price: 10.0 + min_avg_dollar_volume: 25000000 + min_atr_14: 0.50 + min_atr_pct: 0.04 + + min_rvol: 1.5 + min_abs_gap_pct: 0.02 + min_premarket_dollar_vol: 1500000 + max_candidates: 20 + max_candidates_per_sector: 3 + min_candidates_to_trade: 1 + ticker_cooldown_days: 0 + max_gap_pct: 0.04 + + min_candidate_breadth: 0.60 + market_regime_spy_threshold: 0.0015 + market_regime_ticker: QQQ + rolling_loss_days: 7 + rolling_loss_threshold: -0.07 + max_simultaneous_entries: 3 + min_breakout_rel_vol: 1.2 + + weight_rvol: 0.35 + weight_gap: 0.20 + weight_dollar_vol: 0.05 + weight_premarket_dollar_vol: 0.25 + weight_body_ratio: 0.0 + weight_momentum: 0.15 + weight_obv_slope: 0.05 + + # V41 change: hard filter — only trade stocks in accumulation (positive OBV slope) + min_obv_slope_20d: 0.0 + + atr_stop_multiplier: 0.75 + breakeven_at_r: 1.0 + trailing_at_r: 1.0 + trailing_stop_atr_multiplier: 0.8 + trailing_tighten_at_r: 2.0 + trailing_stop_atr_multiplier_tight: 0.3 + + partial_exit_at_r: 99.0 + partial_exit_pct: 0.50 + + risk_per_trade_pct: 0.05 + max_position_pct: 0.70 + daily_max_loss_pct: 0.05 + max_stops_per_day: 5 + exit_minutes_before_close: 5 + + slippage_bps: 5.0 + initial_capital: 10000 + + compound_returns: false + daily_budget_reset: true + settlement_days: 1 + + drawdown_governor_threshold: 0.025 + drawdown_governor_min_scale: 0.30 + + streak_sizing_win_bonus: 0.70 + streak_sizing_max: 2.5 + +universe: + source: midlarge + +backtest: + start_date: null + end_date: null + lookback_trading_days: 200 + +cache: + enabled: true + dir: data/cache/intraday + +output: + dir: runs/intraday_orb + verbose: false diff --git a/configs/intraday/strategies/orb_gainers_v43_30min.yaml b/configs/intraday/strategies/orb_gainers_v43_30min.yaml new file mode 100644 index 0000000..cf8ae20 --- /dev/null +++ b/configs/intraday/strategies/orb_gainers_v43_30min.yaml @@ -0,0 +1,114 @@ +_meta: + id: 110 + name: "ORB Gainers V43 30-Minute ORB" + status: candidate + live_readiness: experimental + parent: orb_gainers_v24_quality_overlay + description: > + V24 → V43: changes the ORB window from 5-min to 30-min. + + V24 uses the first 5-min candle (9:30-9:35 ET) as the opening range. + V43 uses the first 30-min window (9:30-10:00 ET) as the opening range. + + Hypothesis: a 30-min ORB gives more time for: + (a) False breakouts to resolve — early spikes and fades complete within the window + (b) Institutional orderflow to participate — large orders execute over 30 min + (c) The true range to establish — less noise in the high/low + + Risk: entering later (at 10:00+) rather than 9:35 means: + (a) Less time in trade, exit pressure before 4 PM + (b) May miss some early morning momentum + + All other V24 params unchanged. + + Gates (200d): Return ≥ +98.8% (V24 +94.8% + 4pp), DD ≥ -11.79%, Sharpe ≥ 2.93 + Gates (400d): Return ≥ +166.1% (V24 +162.1% + 4pp), DD ≥ -14.20%, Sharpe ≥ 2.52 + +strategy_mode: orb + +orb_strategy: + engine_family: gainers_leader + live_readiness: experimental + orb_minutes: 30 + sim_bar_minutes: 5 + + entry_direction: long_only + order_timeout_minutes: 45 + + allow_doji_breakout: true + allow_red_to_green_breakout: true + + min_price: 10.0 + min_avg_dollar_volume: 25000000 + min_atr_14: 0.50 + min_atr_pct: 0.04 + + min_rvol: 1.5 + min_abs_gap_pct: 0.02 + min_premarket_dollar_vol: 1500000 + max_candidates: 20 + max_candidates_per_sector: 3 + min_candidates_to_trade: 1 + ticker_cooldown_days: 0 + max_gap_pct: 0.04 + + min_candidate_breadth: 0.60 + market_regime_spy_threshold: 0.0015 + market_regime_ticker: QQQ + rolling_loss_days: 7 + rolling_loss_threshold: -0.07 + max_simultaneous_entries: 3 + min_breakout_rel_vol: 1.2 + + weight_rvol: 0.35 + weight_gap: 0.20 + weight_dollar_vol: 0.05 + weight_premarket_dollar_vol: 0.25 + weight_body_ratio: 0.0 + weight_momentum: 0.15 + weight_obv_slope: 0.05 + + atr_stop_multiplier: 0.75 + breakeven_at_r: 1.0 + trailing_at_r: 1.0 + trailing_stop_atr_multiplier: 0.8 + trailing_tighten_at_r: 2.0 + trailing_stop_atr_multiplier_tight: 0.3 + + partial_exit_at_r: 99.0 + partial_exit_pct: 0.50 + + risk_per_trade_pct: 0.05 + max_position_pct: 0.70 + daily_max_loss_pct: 0.05 + max_stops_per_day: 5 + exit_minutes_before_close: 5 + + slippage_bps: 5.0 + initial_capital: 10000 + + compound_returns: false + daily_budget_reset: true + settlement_days: 1 + + drawdown_governor_threshold: 0.025 + drawdown_governor_min_scale: 0.30 + + streak_sizing_win_bonus: 0.70 + streak_sizing_max: 2.5 + +universe: + source: midlarge + +backtest: + start_date: null + end_date: null + lookback_trading_days: 200 + +cache: + enabled: true + dir: data/cache/intraday + +output: + dir: runs/intraday_orb + verbose: false diff --git a/libs/intraday/domain.py b/libs/intraday/domain.py index 0e6ae30..891d71e 100644 --- a/libs/intraday/domain.py +++ b/libs/intraday/domain.py @@ -1127,6 +1127,9 @@ class ORBStrategyParams(BaseModel): weight_obv_slope: float = 0.0 """OBV accumulation slope (20d) ranking weight. Positive OBV = smart-money accumulation pre-breakout.""" + weight_obv_slope_5: float = 0.0 + """OBV accumulation slope (5d) ranking weight. Short-term accumulation signal, orthogonal to 20d slope.""" + weight_gap_zscore: float = 0.0 """Opening-gap z-score ranking weight relative to prior 20 sessions.""" @@ -1213,6 +1216,11 @@ class ORBStrategyParams(BaseModel): where short-sellers are already leaning against the name. Low gap_zscore = routine gap = better ORB. None disables the filter. Gainers_leader only.""" + min_obv_slope_20d: float | None = None + """Minimum allowed OBV slope (20d). 0.0 = require positive accumulation (net up-volume days). + Negative OBV slope = institutional distribution — filters those out when set. + None disables the filter. Gainers_leader only.""" + # ATR-based stop management atr_stop_multiplier: float = 0.10 """Initial stop distance = ATR(14) × this multiplier. Paper uses 10% (0.10).""" diff --git a/libs/intraday/features.py b/libs/intraday/features.py index d8e0fc4..0ce0f06 100644 --- a/libs/intraday/features.py +++ b/libs/intraday/features.py @@ -332,6 +332,10 @@ def enrich_daily_bars( compute_obv_slope_approx(prev_bars, lookback=20) if len(prev_bars) >= 22 else None ), + "obv_slope_5": ( + compute_obv_slope_approx(prev_bars, lookback=5) + if len(prev_bars) >= 7 else None + ), "atr_ratio_10_60": _compute_ratio( compute_average_true_range(prev_bars, lookback=10), compute_average_true_range(prev_bars, lookback=60), diff --git a/libs/intraday/orb_simulator.py b/libs/intraday/orb_simulator.py index 19f0f36..9d8f9a1 100644 --- a/libs/intraday/orb_simulator.py +++ b/libs/intraday/orb_simulator.py @@ -501,6 +501,7 @@ def compute_orb_candidates( entropy_20d = ticker_enrich.get("entropy_20d") obv_slope_20 = ticker_enrich.get("obv_slope_20") + obv_slope_5 = ticker_enrich.get("obv_slope_5") atr_ratio_10_60 = ticker_enrich.get("atr_ratio_10_60") range_compression_10_60 = ticker_enrich.get("range_compression_10_60") gap_zscore_20d = ticker_enrich.get("gap_zscore_20d") @@ -542,6 +543,10 @@ def compute_orb_candidates( if max_gzs is not None and (gap_zscore_20d is None or gap_zscore_20d > max_gzs): _f_rvol += 1 continue + min_obs = getattr(params, "min_obv_slope_20d", None) + if min_obs is not None and (obv_slope_20 is None or obv_slope_20 < min_obs): + _f_rvol += 1 + continue if engine_family == "stocks_in_play_dual_regime": if getattr(params, "require_event_flag", False) and not event_flag: @@ -631,6 +636,7 @@ def compute_orb_candidates( "momentum": momentum, "entropy_20d": entropy_20d or 0.0, "obv_slope_20": obv_slope_20 if obv_slope_20 is not None else 0.0, + "obv_slope_5": obv_slope_5 if obv_slope_5 is not None else 0.0, "atr_ratio_10_60": atr_ratio_10_60 or 0.0, "range_compression_10_60": range_compression_10_60, "gap_zscore_20d": gap_zscore_20d or 0.0, @@ -712,6 +718,7 @@ def compute_orb_candidates( ] entropy_vals = [c["entropy_20d"] for c in raw_candidates] obv_slope_vals = [c["obv_slope_20"] for c in raw_candidates] + obv_slope5_vals = [c["obv_slope_5"] for c in raw_candidates] atr_ratio_vals = [c["atr_ratio_10_60"] for c in raw_candidates] gap_zscore_vals = [c["gap_zscore_20d"] for c in raw_candidates] structure_vals = [ @@ -732,6 +739,7 @@ def compute_orb_candidates( norm_attention_news = _normalize_scores(attention_news_vals) norm_entropy = _normalize_scores(entropy_vals) norm_obv_slope = _normalize_scores(obv_slope_vals) + norm_obv_slope5 = _normalize_scores(obv_slope5_vals) norm_atr_ratio = _normalize_scores(atr_ratio_vals) norm_gap_zscore = _normalize_scores(gap_zscore_vals) @@ -761,6 +769,7 @@ def compute_orb_candidates( }: score += norm_entropy[i] * params.weight_entropy score += norm_obv_slope[i] * params.weight_obv_slope + score += norm_obv_slope5[i] * params.weight_obv_slope_5 score += norm_atr_ratio[i] * params.weight_atr_ratio if engine_family == "compression_breakout": # gap_zscore only added here for compression_breakout;