diff --git a/apps/intraday_bt/scripts/diag_orb_near_miss_composite_v46.py b/apps/intraday_bt/scripts/diag_orb_near_miss_composite_v46.py new file mode 100644 index 0000000..0ee4ee6 --- /dev/null +++ b/apps/intraday_bt/scripts/diag_orb_near_miss_composite_v46.py @@ -0,0 +1,370 @@ +""" +V47 Diagnostic: Near-Miss Composite Signal on V46 400d Trade Set + +Tests four near-miss signal candidates on V46's 284 trade set (400d window): + 1. momentum_20d — 20d price momentum before entry + 2. grav_pull_20_50 — distance from SMA20/SMA50 cluster (ATR-normalized) + 3. range_pos_52w — position in 52-week range [0=at_low, 1=at_high] + 4. obv_slope_20 — OBV accumulation slope (redundancy reference only) + +Pre-committed gates (set before seeing results): + G1: |Pearson| ≥ 0.07 AND n ≥ 120 + G2: |top − bottom tercile avg_R| ≥ 0.30R (binding gate) + G3: |top − bottom tercile WR| ≥ 5pp + G5a: |ρ(feature, obv_slope_20)| < 0.70 (redundancy check) + +Decision: + Any single feature passes G1+G2+G3+G5a → wire individually, backtest V47 + ≥2 features each ≥0.20R AND pairwise ρ < 0.40 AND G5a < 0.70 → composite + Neither → document "near-miss axis closed on V46 base" + +Data source: data/cache/daily/{ticker}.parquet (OHLCV daily bars) +Run source: runs/v46_correct_w12_400d.json/intraday_20260422_043215_63f03e64.json +""" +from __future__ import annotations + +import concurrent.futures +import itertools +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__), "../../.."))) + +DAILY_CACHE_DIR = "data/cache/daily" +V46_400D_RUN = "runs/v46_correct_w12_400d.json/intraday_20260422_043215_63f03e64.json" + + +def load_daily_bars(ticker: str) -> list[dict]: + path = Path(DAILY_CACHE_DIR) / f"{ticker}.parquet" + if not path.exists(): + return [] + try: + table = pq.read_table(str(path)) + rows = table.to_pydict() + except Exception: + return [] + bars = [] + for i, date in enumerate(rows.get("date", [])): + bars.append({ + "date": date, + "open": rows["open"][i], + "high": rows["high"][i], + "low": rows["low"][i], + "close": rows["close"][i], + "volume": rows["volume"][i], + }) + return sorted(bars, key=lambda b: b["date"]) + + +def compute_momentum_20d(prev_bars: list[dict]) -> float | None: + if len(prev_bars) < 21: + return None + c_now = prev_bars[-1]["close"] + c_past = prev_bars[-21]["close"] + if not c_past or c_past <= 0: + return None + return (c_now - c_past) / c_past + + +def compute_grav_pull_20_50(prev_bars: list[dict]) -> float | None: + if len(prev_bars) < 51: + return None + close = prev_bars[-1]["close"] + sma20 = sum(b["close"] for b in prev_bars[-20:]) / 20 + sma50 = sum(b["close"] for b in prev_bars[-50:]) / 50 + ma_center = (sma20 + sma50) / 2.0 + recent = prev_bars[-15:] + true_ranges = [] + for i in range(1, len(recent)): + curr = recent[i] + prev_b = recent[i - 1] + h = curr["high"] or 0 + l = curr["low"] or 0 + pc = prev_b["close"] or 0 + tr = max(h - l, abs(h - pc), abs(l - pc)) if pc > 0 else (h - l) + true_ranges.append(tr) + atr = sum(true_ranges) / len(true_ranges) if true_ranges else 0 + if atr <= 0: + return None + return abs(close - ma_center) / atr + + +def compute_range_pos_52w(prev_bars: list[dict]) -> float | None: + if len(prev_bars) < 50: + return None + window = prev_bars[-252:] if len(prev_bars) >= 252 else prev_bars + prices = [b["close"] for b in window if b["close"] and b["close"] > 0] + if len(prices) < 50: + return None + high_52w = max(prices) + low_52w = min(prices) + last_close = prev_bars[-1]["close"] + if high_52w == low_52w: + return 0.5 + return (last_close - low_52w) / (high_52w - low_52w) + + +def compute_obv_slope_20(prev_bars: list[dict], lookback: int = 20) -> float | None: + if len(prev_bars) < lookback + 2: + return None + recent = prev_bars[-(lookback + 1):] + obv_series = [0.0] + total_vol = 0.0 + for i in range(1, len(recent)): + pc = recent[i - 1]["close"] + cc = recent[i]["close"] + vol = float(recent[i]["volume"] or 0) + total_vol += vol + if not pc or pc <= 0: + continue + if cc > pc: + obv_series.append(obv_series[-1] + vol) + elif cc < pc: + obv_series.append(obv_series[-1] - vol) + else: + obv_series.append(obv_series[-1]) + avg_vol = total_vol / lookback if lookback > 0 else 1.0 + if avg_vol <= 0: + return None + n = len(obv_series) + x_mean = (n - 1) / 2.0 + y_mean = sum(obv_series) / n + numerator = sum((i - x_mean) * (obv_series[i] - y_mean) for i in range(n)) + denominator = sum((i - x_mean) ** 2 for i in range(n)) + if denominator <= 0: + return 0.0 + return (numerator / denominator) / avg_vol + + +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, idxs: list[int], vals: list[float], rs: list[float], + obv_by_idx: dict[int, float]) -> dict: + 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 {"pass_g2": False, "pass_all": False, "pearson": None, "g2_gap": None, "g5a": True} + + low, mid, high = ts["low"], ts["mid"], ts["high"] + raw_g2_gap = high["avg_r"] - low["avg_r"] + avg_r_gap = abs(raw_g2_gap) + 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 + + # Compute ρ(feature, obv_slope) on the intersection of trade indices + obv_aligned = [(vals[i], obv_by_idx[idx]) for i, idx in enumerate(idxs) if idx in obv_by_idx] + rho_obv = None + g5a = True + if len(obv_aligned) >= 20: + va = [x[0] for x in obv_aligned] + vo = [x[1] for x in obv_aligned] + rho_obv = pearson(va, vo) + if rho_obv is not None: + g5a = abs(rho_obv) < 0.70 + + print(f"\n [{label}] n={n} Pearson={p:+.3f} direction={'↑high' if raw_g2_gap > 0 else '↓low'}") + print(f" G1: {'PASS' if g1 else 'FAIL'} (|{abs(p):.3f}| {'≥' if abs(p) >= 0.07 else '<'} 0.07, n={n})") + print(f" G2: {'PASS' if g2 else 'FAIL'} (avg_R gap = {avg_r_gap:.3f}R [threshold 0.30R])") + print(f" G3: {'PASS' if g3 else 'FAIL'} (WR gap = {wr_gap * 100:.1f}pp [threshold 5pp])") + if rho_obv is not None: + print(f" G5a: {'PASS' if g5a else 'FAIL'} (|ρ_OBV| = {abs(rho_obv):.3f} [threshold < 0.70], n={len(obv_aligned)})") + 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']}") + pass_all = g1 and g2 and g3 and g5a + print(f" → {'ALL GATES PASS ✓' if pass_all else 'FAIL'}") + + return { + "pass_g2": g2, "pass_all": pass_all, "pearson": p, + "g2_gap": avg_r_gap, "raw_g2_gap": raw_g2_gap, + "g5a": g5a, "rho_obv": rho_obv, + "direction": "high" if raw_g2_gap > 0 else "low", + } + + +def pairwise_rho_by_idx(idxs_a: list[int], vals_a: list[float], + idxs_b: list[int], vals_b: list[float]) -> float | None: + map_b = {idx: vals_b[i] for i, idx in enumerate(idxs_b)} + aligned_a, aligned_b = [], [] + for i, idx in enumerate(idxs_a): + if idx in map_b: + aligned_a.append(vals_a[i]) + aligned_b.append(map_b[idx]) + if len(aligned_a) < 20: + return None + return pearson(aligned_a, aligned_b) + + +def main() -> None: + print("=== V47 Near-Miss Composite Diagnostic (V46 400d Base) ===\n") + + with open(V46_400D_RUN) as f: + run_data = json.load(f) + trades = run_data.get("trades", []) + m_meta = run_data.get("metrics", {}) + print(f"V46 400d: {m_meta.get('total_trades')} trades, " + f"{m_meta.get('start_date')} → {m_meta.get('end_date')}, " + f"return={m_meta.get('total_return_pct', 0) * 100:.1f}%") + + trade_records = [ + {"ticker": t["ticker"], "date": t["date"][:10], "r": float(t["r_multiple_at_exit"])} + for t in trades + if t.get("r_multiple_at_exit") is not None + ] + print(f"Valid r_multiple trades: {len(trade_records)}") + + tickers_needed = sorted(set(r["ticker"] for r in trade_records)) + print(f"Unique tickers: {len(tickers_needed)}") + print("Loading daily bars...") + + ticker_bars: dict[str, list[dict]] = {} + with concurrent.futures.ThreadPoolExecutor(max_workers=12) as ex: + def _load(ticker: str) -> tuple[str, list[dict]]: + return ticker, load_daily_bars(ticker) + 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") + + # Per-trade feature computation; track by trade index for cross-feature alignment + m20_idxs, m20_vals, m20_rs = [], [], [] + gp_idxs, gp_vals, gp_rs = [], [], [] + rp_idxs, rp_vals, rp_rs = [], [], [] + obv_idxs, obv_vals, obv_rs = [], [], [] + obv_by_idx: dict[int, float] = {} + + for idx, rec in enumerate(trade_records): + ticker = rec["ticker"] + date = rec["date"] + r = rec["r"] + + bars = ticker_bars.get(ticker, []) + prev_bars = [b for b in bars if b["date"] < date] + + m20 = compute_momentum_20d(prev_bars) + gp = compute_grav_pull_20_50(prev_bars) + rp = compute_range_pos_52w(prev_bars) + obv = compute_obv_slope_20(prev_bars) + + if m20 is not None: + m20_idxs.append(idx); m20_vals.append(m20); m20_rs.append(r) + if gp is not None: + gp_idxs.append(idx); gp_vals.append(gp); gp_rs.append(r) + if rp is not None: + rp_idxs.append(idx); rp_vals.append(rp); rp_rs.append(r) + if obv is not None: + obv_idxs.append(idx); obv_vals.append(obv); obv_rs.append(r) + obv_by_idx[idx] = obv + + total = len(trade_records) + print(f"Coverage: momentum_20d={len(m20_vals)}/{total} " + f"grav_pull={len(gp_vals)}/{total} " + f"range_52w={len(rp_vals)}/{total} " + f"obv_slope={len(obv_vals)}/{total}") + + print("\n" + "=" * 65) + print("GATE RESULTS (G1: |P|≥0.07 & n≥120; G2: avg_R≥0.30R; G3: WR≥5pp; G5a: ρ_OBV<0.70)") + + res_m = report_feature("momentum_20d", m20_idxs, m20_vals, m20_rs, obv_by_idx) + res_g = report_feature("grav_pull_20_50 (ATR-norm)", gp_idxs, gp_vals, gp_rs, obv_by_idx) + res_r = report_feature("range_pos_52w (0=low,1=high)", rp_idxs, rp_vals, rp_rs, obv_by_idx) + res_o = report_feature("obv_slope_20 (reference)", obv_idxs, obv_vals, obv_rs, {}) + + print("\n" + "=" * 65) + print("PAIRWISE CORRELATIONS (composite gate: ρ < 0.40 between candidates)") + + named = [ + ("momentum_20d", m20_idxs, m20_vals), + ("grav_pull_20_50", gp_idxs, gp_vals), + ("range_pos_52w", rp_idxs, rp_vals), + ("obv_slope_20", obv_idxs, obv_vals), + ] + pair_rhos: dict[tuple[str, str], float | None] = {} + for (na, ia, va), (nb, ib, vb) in itertools.combinations(named, 2): + rho = pairwise_rho_by_idx(ia, va, ib, vb) + pair_rhos[(na, nb)] = rho + print(f" ρ({na}, {nb}) = {rho:+.3f}" if rho is not None else f" ρ({na}, {nb}) = N/A") + + print("\n" + "=" * 65) + print("DECISION") + + candidate_results = [ + ("momentum_20d", res_m), + ("grav_pull_20_50", res_g), + ("range_pos_52w", res_r), + ] + single_pass = [name for name, res in candidate_results if res["pass_all"]] + + composite_cands = [ + (name, res) for name, res in candidate_results + if res["g2_gap"] is not None and res["g2_gap"] >= 0.20 and res["g5a"] + ] + + composite_pairs_ok = [] + if len(composite_cands) >= 2: + for (na, ra), (nb, rb) in itertools.combinations(composite_cands, 2): + rho = pair_rhos.get((na, nb)) or pair_rhos.get((nb, na)) + if rho is not None and abs(rho) < 0.40: + composite_pairs_ok.append((na, nb, ra["g2_gap"], rb["g2_gap"], rho)) + + if single_pass: + best = max(single_pass, key=lambda n: { + "momentum_20d": res_m, "grav_pull_20_50": res_g, "range_pos_52w": res_r + }[n]["g2_gap"] or 0) + best_res = {"momentum_20d": res_m, "grav_pull_20_50": res_g, "range_pos_52w": res_r}[best] + print(f"\n RESULT: SINGLE FEATURE PASS → {single_pass}") + print(f" Best: {best} G2={best_res['g2_gap']:.3f}R Pearson={best_res['pearson']:+.3f}") + print(" Action: wire feature, build V47 config, run 200d + 400d backtest") + elif composite_pairs_ok: + for na, nb, ga, gb, rho in composite_pairs_ok: + print(f"\n COMPOSITE CANDIDATE: {na} + {nb}") + print(f" G2 gap: {ga:.3f}R + {gb:.3f}R pairwise ρ={rho:.3f}") + print(" Action: wire composite, build V47 config, run 200d + 400d backtest") + else: + best_name, best_gap = "none", 0.0 + for name, res in candidate_results: + if res["g2_gap"] is not None and res["g2_gap"] > best_gap: + best_gap = res["g2_gap"] + best_name = name + print(f"\n RESULT: ALL FAIL — composite near-miss axis CLOSED on V46 base") + print(f" Best near-miss: {best_name} (G2={best_gap:.3f}R)") + print(" Next step: new data required (options flow, 13F institutional ownership)") + + +if __name__ == "__main__": + main()