yirugi
/
fithia2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Diagnose ORB tape ignition (V26): NULL result, V24 remains champion

Browse Source 
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 <noreply@anthropic.com>
main
I Luk Kim 4 months ago
parent 1e7fcde14d
commit 137b413084
1 changed files with 683 additions and 0 deletions
Show Stats Download Patch File Download Diff File

683
apps/intraday_bt/scripts/diag_orb_tape_ignition.py
Unescape Escape View File

@ -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()
Write Preview
Loading…
Cancel
Save