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

Diagnose FINRA short-volume axis for V25: NULL result, V24 remains champion

Browse Source 
Phase 1 diagnostic (diag_orb_short_volume.py) tests 3 short-ratio features
(prior-day, 20d avg, 20d z-score) on V24 200d trade set via FINRA CDN data.
All features fail all gates (|Pearson| ≤ 0.062, avg_R gap < 0.30R, WR gap < 5pp).

Direction inverts Boehmer 2020 hypothesis: high short ratio correlates positively
with ORB r_multiple (short-squeeze dynamics dominate at intraday ORB timescale).
V25 short-volume overlay not wired. V24 remains live champion.

Also adds scripts/audit_short_volume_coverage.py (Phase 0 DB coverage audit).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
main
I Luk Kim 4 months ago
parent 4b8a157a67
commit 1e7fcde14d
2 changed files with 776 additions and 0 deletions
Show Stats Download Patch File Download Diff File

596
apps/intraday_bt/scripts/diag_orb_short_volume.py
Unescape Escape View File

@ -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()

180
scripts/audit_short_volume_coverage.py
Unescape Escape View File

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