diff --git a/apps/intraday_bt/scripts/diag_orb_short_volume.py b/apps/intraday_bt/scripts/diag_orb_short_volume.py
new file mode 100644
index 0000000..1009de4
--- /dev/null
+++ b/apps/intraday_bt/scripts/diag_orb_short_volume.py
@@ -0,0 +1,596 @@
+"""
+V25 Short-Volume Feature Diagnostic: FINRA Short Ratio
+
+Hypothesis (Boehmer et al. 2020): For a LONG ORB, higher short-volume ratio entering
+a breakout day predicts weaker follow-through (informed bears leaning against the move).
+Direction: LOWER short ratio → HIGHER r_multiple. Confirmed empirically here.
+
+Method:
+  1. Run V24 simulation over 200d window (baseline)
+  2. For each completed trade, compute three short-volume features from FINRA CDN data:
+       - short_ratio_prior_day : ratio from the most-recent prior trading day
+       - short_ratio_avg_20d   : mean ratio over last 20 prior trading days
+       - short_ratio_zscore_20d: (prior_day − mean_20d) / std_20d (spike detector)
+  3. Report per-feature: Pearson vs r_multiple_at_exit, tercile WR/avg_R, coverage
+  4. Pairwise correlation vs obv_slope_20 (redundancy check)
+
+Gates (for promotion to Phase 2):
+  G1: |Pearson| ≥ 0.07 on ≥ 120 trades (relaxed to 60 if window coverage < 80%)
+  G2: |top − bottom tercile avg_R| ≥ 0.30R
+  G3: |top − bottom tercile WR| ≥ 5pp
+  G4: feature coverage ≥ 80% of trades with valid r_multiple
+  G5: |ρ(short_feature, obv_slope_20)| < 0.70 (redundancy gate)
+
+Data source: FINRA CDN daily files cached to data/cache/finra_daily/
+  URL template: https://cdn.finra.org/equity/regsho/daily/CNMSshvol{yyyymmdd}.txt
+"""
+from __future__ import annotations
+
+import concurrent.futures
+import datetime as dt
+import json
+import math
+import os
+import statistics
+import sys
+from pathlib import Path
+
+import requests
+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
+
+# ── 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"
+FINRA_CACHE_DIR = "data/cache/finra_daily"
+LOOKBACK_DAYS = 200
+FEATURE_LOOKBACK_TRADING = 25   # need 20 prior short-vol days; add small buffer
+
+FINRA_CDN_URL = "https://cdn.finra.org/equity/regsho/daily/CNMSshvol{yyyymmdd}.txt"
+
+_ET = ZoneInfo("America/New_York")
+_MKT_OPEN = dt.time(9, 30)
+_MKT_CLOSE = dt.time(16, 0)
+
+
+# ── FINRA Data Fetching ──────────────────────────────────────────────────────
+
+def _fetch_finra_daily(date: dt.date) -> dict[str, float]:
+    """Download and parse one FINRA daily file. Returns {ticker: short_ratio}."""
+    cache_path = Path(FINRA_CACHE_DIR) / f"{date.isoformat()}.json"
+    if cache_path.exists():
+        try:
+            with open(cache_path) as f:
+                return json.load(f)
+        except Exception:
+            pass
+
+    url = FINRA_CDN_URL.format(yyyymmdd=date.strftime("%Y%m%d"))
+    try:
+        resp = requests.get(url, timeout=15)
+    except requests.RequestException:
+        return {}
+    if resp.status_code != 200:
+        return {}
+
+    lines = [line.strip() for line in resp.text.splitlines() if line.strip()]
+    if not lines:
+        return {}
+
+    header = [p.strip() for p in lines[0].split("|")]
+    try:
+        sym_idx = header.index("Symbol")
+        short_idx = header.index("ShortVolume")
+        total_idx = header.index("TotalVolume")
+    except ValueError:
+        return {}
+
+    ratios: dict[str, float] = {}
+    for line in lines[1:]:
+        parts = [p.strip() for p in line.split("|")]
+        if len(parts) <= max(sym_idx, short_idx, total_idx):
+            continue
+        try:
+            total = float(parts[total_idx])
+            short = float(parts[short_idx])
+        except ValueError:
+            continue
+        if total > 0:
+            ratios[parts[sym_idx].upper()] = short / total
+
+    Path(FINRA_CACHE_DIR).mkdir(parents=True, exist_ok=True)
+    with open(cache_path, "w") as f:
+        json.dump(ratios, f)
+    return ratios
+
+
+def build_short_volume_dict(trading_days: list[str], workers: int = 16) -> dict[str, dict[str, float]]:
+    """Download FINRA CDN for each trading day → {ticker: {date: short_ratio}}.
+
+    Dates are trading-day strings. Files are cached to FINRA_CACHE_DIR.
+    """
+    def _load(d_str: str) -> tuple[str, dict[str, float]]:
+        d = dt.date.fromisoformat(d_str)
+        return d_str, _fetch_finra_daily(d)
+
+    print(f"  Fetching/loading {len(trading_days)} FINRA CDN files (parallel, cached)...")
+    results: list[tuple[str, dict[str, float]]] = []
+    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
+        for i, (d_str, ratios) in enumerate(ex.map(_load, trading_days)):
+            results.append((d_str, ratios))
+            if (i + 1) % 25 == 0 or (i + 1) == len(trading_days):
+                print(f"  {i+1}/{len(trading_days)} days loaded")
+
+    # Invert: {ticker: {date: ratio}}
+    by_ticker: dict[str, dict[str, float]] = {}
+    for d_str, ratios in results:
+        for ticker, ratio in ratios.items():
+            if ticker not in by_ticker:
+                by_ticker[ticker] = {}
+            by_ticker[ticker][d_str] = ratio
+    return by_ticker
+
+
+# ── Short-Ratio Feature Computations ────────────────────────────────────────
+
+def compute_short_ratio_prior_day(
+    ticker: str,
+    entry_date: str,
+    short_vol: dict[str, dict[str, float]],
+    trading_days: list[str],
+) -> float | None:
+    """Most-recent prior-day short ratio. Lookahead-free (strictly before entry_date)."""
+    ticker_data = short_vol.get(ticker, {})
+    if not ticker_data:
+        return None
+    if entry_date in trading_days:
+        idx = trading_days.index(entry_date)
+        for d in reversed(trading_days[:idx]):
+            r = ticker_data.get(d)
+            if r is not None:
+                return r
+    return None
+
+
+def compute_short_ratio_avg_20d(
+    ticker: str,
+    entry_date: str,
+    short_vol: dict[str, dict[str, float]],
+    trading_days: list[str],
+) -> float | None:
+    """Mean short ratio over the last 20 prior trading days."""
+    ticker_data = short_vol.get(ticker, {})
+    if not ticker_data:
+        return None
+    if entry_date not in trading_days:
+        return None
+    idx = trading_days.index(entry_date)
+    prior = trading_days[max(0, idx - 20):idx]
+    vals = [ticker_data[d] for d in prior if d in ticker_data]
+    if not vals:
+        return None
+    return sum(vals) / len(vals)
+
+
+def compute_short_ratio_zscore_20d(
+    ticker: str,
+    entry_date: str,
+    short_vol: dict[str, dict[str, float]],
+    trading_days: list[str],
+) -> float | None:
+    """Z-score of prior-day short ratio vs 20d history."""
+    prior_day = compute_short_ratio_prior_day(ticker, entry_date, short_vol, trading_days)
+    avg_20d = compute_short_ratio_avg_20d(ticker, entry_date, short_vol, trading_days)
+    if prior_day is None or avg_20d is None:
+        return None
+    if entry_date not in trading_days:
+        return None
+    idx = trading_days.index(entry_date)
+    prior = trading_days[max(0, idx - 20):idx]
+    ticker_data = short_vol.get(ticker, {})
+    vals = [ticker_data[d] for d in prior if d in ticker_data]
+    if len(vals) < 5:
+        return None
+    std = statistics.stdev(vals) if len(vals) >= 2 else 0.0
+    if std <= 0:
+        return 0.0
+    return (prior_day - avg_20d) / std
+
+
+# ── Daily Bar Builder (same as diag_orb_quality_features.py) ─────────────────
+
+import pyarrow.parquet as pq
+
+
+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:
+            if ts_raw.endswith("Z"):
+                ts_raw = ts_raw[:-1] + "+00:00"
+            ts = dt.datetime.fromisoformat(ts_raw).astimezone(_ET)
+        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]]]:
+    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
+
+
+# ── 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(pairs_sub):
+        ys = [p[1] for p in pairs_sub]
+        wins = [y for y in ys if y > 0]
+        wr = len(wins) / 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("=== V25 Short-Volume Feature Diagnostic ===\n")
+
+    # 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 range for daily bars (need feature_lookback for prev_bars)
+    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
+    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
+    print("Computing enrichment...")
+    enrichment = enrich_daily_bars(daily_bars, trading_days_list)
+    print(f"Enrichment for {len(enrichment)} tickers")
+
+    # 6. Pre-screen + load intraday
+    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 intraday bars...")
+    all_intraday = load_intraday_bulk(candidates)
+    intraday_pairs = sum(len(v) for v in all_intraday.values())
+    print(f"Loaded: {intraday_pairs} 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 — insufficient sample")
+        return
+
+    # 8. Load FINRA short-volume data (CDN, cached)
+    # Need data for extended window to compute 20d lookback before first trade
+    extended_short_days = [d.isoformat() for d in all_td[-(LOOKBACK_DAYS + 30):]]
+    print(f"\nLoading FINRA short-volume data for {len(extended_short_days)} trading days...")
+    short_vol = build_short_volume_dict(extended_short_days)
+    print(f"Short-vol dict: {len(short_vol)} tickers with data")
+
+    # Check coverage for unique trade tickers
+    trade_tickers_uniq = {t.ticker for t in trades_with_r}
+    covered_tickers = {t for t in trade_tickers_uniq if t in short_vol}
+    print(f"Trade tickers: {len(trade_tickers_uniq)} unique, {len(covered_tickers)} have short-vol data")
+
+    # 9. Build daily bars lookup for OBV (pairwise correlation)
+    sorted_daily: dict[str, list[dict]] = {
+        t: sorted(bars, key=lambda b: b["date"])
+        for t, bars in daily_bars.items()
+    }
+
+    # 10. Compute features per trade
+    print("\nComputing features per trade...")
+    annotated: list[dict] = []
+    missing = {"prior_day": 0, "avg_20d": 0, "zscore_20d": 0, "obv_slope": 0}
+
+    for trade in trades_with_r:
+        ticker = trade.ticker
+        date = trade.date
+        r = float(trade.r_multiple_at_exit)
+
+        # Short-ratio features
+        f_prior = compute_short_ratio_prior_day(ticker, date, short_vol, extended_short_days)
+        f_avg = compute_short_ratio_avg_20d(ticker, date, short_vol, extended_short_days)
+        f_zscore = compute_short_ratio_zscore_20d(ticker, date, short_vol, extended_short_days)
+
+        # 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)
+
+        if f_prior is None:
+            missing["prior_day"] += 1
+        if f_avg is None:
+            missing["avg_20d"] += 1
+        if f_zscore is None:
+            missing["zscore_20d"] += 1
+        if obv is None:
+            missing["obv_slope"] += 1
+
+        annotated.append({
+            "ticker": ticker, "date": date, "r": r,
+            "prior_day": f_prior,
+            "avg_20d": f_avg,
+            "zscore_20d": f_zscore,
+            "obv_slope": obv,
+            "win": r > 0,
+        })
+
+    total = len(annotated)
+    print(f"Annotated: {total} trades")
+    print(f"Missing: prior_day={missing['prior_day']}, avg_20d={missing['avg_20d']}, "
+          f"zscore_20d={missing['zscore_20d']}, obv_slope={missing['obv_slope']}")
+
+    # 11. Report per-feature
+    feature_defs = [
+        ("short_ratio_prior_day", "prior_day",
+         "Lower = fewer informed shorts = cleaner breakout (Boehmer 2020)", "low"),
+        ("short_ratio_avg_20d", "avg_20d",
+         "20d avg short ratio baseline", "low"),
+        ("short_ratio_zscore_20d", "zscore_20d",
+         "Spike detector: high z = unusual short surge before breakout", "low"),
+    ]
+
+    print("\n" + "=" * 90)
+    print("FEATURE ANALYSIS — V24 200d trade set")
+    print("=" * 90)
+
+    results = {}
+    for feat_name, feat_key, description, best_tercile in feature_defs:
+        valid = [(t[feat_key], t["r"]) for t in annotated if t[feat_key] 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]
+        wins = [v for v in valid if v[1] > 0]
+        overall_wr = len(wins) / len(valid)
+
+        rho = pearson(vals, rs)
+        tstat = tercile_stats(vals, rs)
+        coverage_pct = len(valid) / total
+
+        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}%")
+        if rho is not None:
+            print(f"  Pearson(feature, r_multiple) = {rho:.4f}")
+        else:
+            print("  Pearson = n/a")
+
+        # Determine which end is "high performance"
+        worst_tercile = "high" if best_tercile == "low" else "low"
+
+        if tstat:
+            h = tstat["high"]
+            m = tstat["mid"]
+            lo = tstat["low"]
+            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}")
+
+            best = tstat[best_tercile]
+            worst = tstat[worst_tercile]
+            rho_abs = abs(rho) if rho is not None else 0.0
+
+            # Adjust minimum trades gate based on coverage
+            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  # at least 50% of trades covered
+
+            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 ✗'}")
+            overall_pass = g1 and g2 and g3 and g4
+            print(f"  VERDICT: {'ALL GATES PASS → PROCEED TO PHASE 2' if overall_pass else f'FAIL ({sum([not g1, not g2, not g3, not g4])} gate(s) failed)'}")
+            results[feat_name] = {
+                "pass": overall_pass, "pearson": rho, "stats": tstat,
+                "n": len(valid), "coverage": coverage_pct, "best_tercile": best_tercile,
+            }
+
+    # 12. Pairwise correlations
+    print(f"\n{'─'*60}")
+    print("PAIRWISE CORRELATIONS (short features vs obv_slope_20):")
+    for feat_key, feat_name_short in [("prior_day", "prior_day"), ("avg_20d", "avg_20d"), ("zscore_20d", "zscore_20d")]:
+        combined = [(t[feat_key], t["obv_slope"]) for t in annotated
+                    if t[feat_key] is not None and t["obv_slope"] is not None]
+        if len(combined) >= 10:
+            fvals = [c[0] for c in combined]
+            ovals = [c[1] for c in combined]
+            rho_x = pearson(fvals, ovals)
+            print(f"  ρ({feat_name_short}, obv_slope): {rho_x:.4f}" if rho_x is not None else f"  ρ({feat_name_short}, obv_slope): n/a")
+            if rho_x is not None:
+                g5_pass = abs(rho_x) < 0.70
+                if feat_name_short in [k.split("short_ratio_")[-1] if k.startswith("short_ratio_") else k for k in results]:
+                    print(f"    G5 |ρ| < 0.70: {'PASS ✓' if g5_pass else 'FAIL (redundant with OBV) ✗'}")
+
+    # Inter-feature correlations
+    for f1, f2 in [("prior_day", "avg_20d"), ("prior_day", "zscore_20d"), ("avg_20d", "zscore_20d")]:
+        combined = [(t[f1], t[f2]) for t in annotated if t[f1] is not None and t[f2] is not None]
+        if len(combined) >= 10:
+            rho_x = pearson([c[0] for c in combined], [c[1] for c in combined])
+            print(f"  ρ({f1}, {f2}): {rho_x:.4f}" if rho_x is not None else f"  ρ({f1}, {f2}): n/a")
+
+    # 13. 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 (insufficient data): {skipped if skipped else 'NONE'}")
+
+    if passing:
+        # Pick strongest by |Pearson|
+        best = max(passing, key=lambda n: abs(results[n]["pearson"] or 0))
+        pearson_val = results[best]["pearson"]
+        obv_pearson_ref = 0.2349  # V24's OBV Pearson from Phase 1
+        weight_magnitude = round(0.05 * min(1.0, abs(pearson_val) / obv_pearson_ref), 2)
+        weight_magnitude = max(weight_magnitude, 0.02)
+
+        # Sign: "low" best tercile means NEGATIVE weight (lower ratio = better)
+        best_direction = results[best]["best_tercile"]
+        weight_sign = -1 if best_direction == "low" 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_short_ratio = {weight_sign * weight_magnitude:+.3f}")
+        print(f"  Suggested weight_short_ratio in V25 config: {weight_sign * weight_magnitude:+.3f}")
+        print(f"  (Magnitude: 0.05 × |{abs(pearson_val):.4f}| / 0.23 = {weight_magnitude:.3f}, floor 0.02)")
+        print(f"\n  → Wire '{best}' as 'short_ratio_signal' in enrich_daily_bars")
+        print(f"  → Add weight_short_ratio = {weight_sign * weight_magnitude:+.3f} to V25 config (parent V24)")
+    else:
+        print(f"\nVERDICT: ABORT — FINRA short-volume axis null on V24 200d trade set")
+        print("  → V24 remains champion. Short-volume axis exhausted on this set.")
+        print("  → Next Ralph iteration: RSI-14, BB %B, or gravitational pull from libs/features/market_features.py")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/scripts/audit_short_volume_coverage.py b/scripts/audit_short_volume_coverage.py
new file mode 100644
index 0000000..f073425
--- /dev/null
+++ b/scripts/audit_short_volume_coverage.py
@@ -0,0 +1,180 @@
+"""
+Phase 0: Audit FINRA short_sale_daily coverage for V25 short-volume overlay.
+
+Checks whether the DB has sufficient historical short-volume data for:
+  - The midlarge universe over the 200d backtest window (2025-07-07 → 2026-04-21)
+  - The V24 200d trade set (ticker, entry_date) pairs
+
+Gates:
+  G0a: Median universe coverage ≥ 80%
+  G0b: V24 trade-set coverage ≥ 80% (≥ 114/142 trades)
+"""
+from __future__ import annotations
+
+import asyncio
+import datetime as dt
+import json
+import os
+import sys
+from pathlib import Path
+
+import asyncpg
+import yaml
+
+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
+
+UNIVERSE_FILE = "configs/symbols_midlarge_snapshot_exact.yaml"
+V24_RUN_FILE = "runs/intraday_orb/intraday_20260421_205350_67d5361a.json"
+LOOKBACK_DAYS = 200
+END_DATE = dt.date(2026, 4, 21)
+
+
+async def main() -> None:
+    print("=== Phase 0: short_sale_daily Coverage Audit ===\n")
+
+    # 1. Compute 200d window
+    all_td = trading_days_between(END_DATE - dt.timedelta(days=400), END_DATE)
+    trading_days = [d.isoformat() for d in all_td[-LOOKBACK_DAYS:]]
+    start_date = trading_days[0]
+    end_date = trading_days[-1]
+    print(f"Window: {start_date} → {end_date} ({len(trading_days)} trading days)")
+
+    # 2. Load universe
+    with open(UNIVERSE_FILE) as f:
+        udata = yaml.safe_load(f)
+    universe: list[str] = udata.get("symbols", udata) if isinstance(udata, dict) else udata
+    print(f"Universe: {len(universe)} tickers")
+
+    # 3. Connect to DB (raw asyncpg, same pattern as enrich_tier2_features.py:263)
+    dsn = get_settings().postgres_dsn.replace("+asyncpg", "")
+    conn = await asyncpg.connect(dsn=dsn)
+
+    try:
+        # 4. Check overall DB date range for short_sale_daily
+        date_range_row = await conn.fetchrow(
+            "SELECT MIN(trade_date)::text AS min_date, MAX(trade_date)::text AS max_date, COUNT(*) AS total_rows FROM short_sale_daily"
+        )
+        print(f"\nDB table short_sale_daily:")
+        print(f"  Total rows: {date_range_row['total_rows']:,}")
+        print(f"  Date range: {date_range_row['min_date']} → {date_range_row['max_date']}")
+
+        # 5. Bulk fetch coverage for universe tickers × window
+        rows = await conn.fetch(
+            """
+            SELECT ticker_raw,
+                   COUNT(DISTINCT trade_date) AS row_count,
+                   MIN(trade_date)::text AS min_date,
+                   MAX(trade_date)::text AS max_date
+            FROM short_sale_daily
+            WHERE ticker_raw = ANY($1)
+              AND trade_date >= $2::date
+              AND trade_date <= $3::date
+              AND total_volume IS NOT NULL
+              AND total_volume > 0
+            GROUP BY ticker_raw
+            """,
+            universe, start_date, end_date,
+        )
+
+        coverage_by_ticker: dict[str, int] = {r["ticker_raw"]: r["row_count"] for r in rows}
+        n_trading = len(trading_days)
+
+        pcts = []
+        zero_tickers = []
+        for ticker in universe:
+            count = coverage_by_ticker.get(ticker, 0)
+            pct = count / n_trading
+            pcts.append(pct)
+            if count == 0:
+                zero_tickers.append(ticker)
+
+        pcts_sorted = sorted(pcts)
+        median_pct = pcts_sorted[len(pcts_sorted) // 2]
+        mean_pct = sum(pcts) / len(pcts)
+        tickers_above_80 = sum(1 for p in pcts if p >= 0.80)
+        tickers_above_60 = sum(1 for p in pcts if p >= 0.60)
+
+        print(f"\nUniverse coverage over window:")
+        print(f"  Median: {median_pct*100:.1f}%")
+        print(f"  Mean:   {mean_pct*100:.1f}%")
+        print(f"  Tickers ≥80% coverage: {tickers_above_80}/{len(universe)}")
+        print(f"  Tickers ≥60% coverage: {tickers_above_60}/{len(universe)}")
+        print(f"  Zero-coverage tickers: {len(zero_tickers)}")
+        if zero_tickers[:10]:
+            print(f"  (first 10): {zero_tickers[:10]}")
+
+        gate_g0a = median_pct >= 0.80
+        print(f"\n  G0a Median ≥ 80%: {median_pct*100:.1f}%  → {'PASS ✓' if gate_g0a else 'FAIL ✗'}")
+
+        # 6. V24 trade-set coverage
+        if Path(V24_RUN_FILE).exists():
+            with open(V24_RUN_FILE) as f:
+                v24_data = json.load(f)
+            v24_trades = v24_data.get("trades", [])
+            print(f"\nV24 trade set: {len(v24_trades)} trades")
+
+            # For each trade, check if short volume exists for the entry date and prev 20 days
+            # Use a single bulk query for all (ticker, date) combos within window
+            trade_tickers = list({t["ticker"] for t in v24_trades})
+            short_rows = await conn.fetch(
+                """
+                SELECT ticker_raw, trade_date::text AS date
+                FROM short_sale_daily
+                WHERE ticker_raw = ANY($1)
+                  AND trade_date >= $2::date
+                  AND trade_date <= $3::date
+                  AND total_volume IS NOT NULL
+                  AND total_volume > 0
+                """,
+                trade_tickers, start_date, end_date,
+            )
+
+            # Build set of (ticker, date) with data
+            short_set: set[tuple[str, str]] = {(r["ticker_raw"], r["date"]) for r in short_rows}
+
+            # For each trade, check if the prior-day short volume is available
+            # (any short data in the 10 trading days before entry = sufficient for prior-day feature)
+            trade_idx = {d: i for i, d in enumerate(trading_days)}
+            covered = 0
+            missing_trades: list[str] = []
+            for trade in v24_trades:
+                ticker = trade["ticker"]
+                entry_date = trade["date"]
+                # Find prior trading days (look back up to 10)
+                if entry_date in trade_idx:
+                    idx = trade_idx[entry_date]
+                    prior_days = trading_days[max(0, idx - 10):idx]
+                else:
+                    prior_days = []
+                has_prior = any((ticker, d) in short_set for d in prior_days)
+                if has_prior:
+                    covered += 1
+                else:
+                    missing_trades.append(f"{trade['date']}:{ticker}")
+
+            trade_coverage_pct = covered / len(v24_trades)
+            gate_g0b = covered >= int(0.80 * len(v24_trades))
+            print(f"  Trades with prior-day short data: {covered}/{len(v24_trades)} ({trade_coverage_pct*100:.1f}%)")
+            print(f"  G0b V24 trade coverage ≥ 80%: {trade_coverage_pct*100:.1f}%  → {'PASS ✓' if gate_g0b else 'FAIL ✗'}")
+            if missing_trades[:10]:
+                print(f"  Missing trades (first 10): {missing_trades[:10]}")
+
+            both_pass = gate_g0a and gate_g0b
+        else:
+            print(f"\nWARNING: V24 run file not found: {V24_RUN_FILE}")
+            both_pass = gate_g0a
+
+        print(f"\n{'='*60}")
+        print(f"PHASE 0 VERDICT: {'PASS — proceed to Phase 1' if both_pass else 'FAIL — backfill required before Phase 1'}")
+        if not both_pass:
+            print("  Action: run `apps/sync/short_volume_sync/main.py --days 300` then re-audit")
+
+    finally:
+        await conn.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())