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

Add V47 near-miss composite diagnostic on V46 400d base

Browse Source 
Tests momentum_20d, grav_pull_20_50, range_pos_52w, obv_slope_20 on V46's
284-trade set (400d). All signals collapse vs V24 base: best G2=0.118R
(momentum_20d) vs 0.291R on V24 and pre-committed gate of 0.30R.
Composite gate (2+ features ≥ 0.20R) also fails. Axis permanently closed.

V46's PEAD event catalyst selection fundamentally changes the ticker/entry
universe vs V24's pure momentum base — momentum signals non-transferable.
V47 requires new data sources (options flow, 13F institutional ownership).

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

370
apps/intraday_bt/scripts/diag_orb_near_miss_composite_v46.py
Unescape Escape View File

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