""" V28 Stage A: Signal sanity probe. Does the existing sparse attention cache contain usable information about V23 trade outcomes? Usage: python scripts/v28_probe/wiki_signal_probe.py Gate (promote to Stage B): spread (WR_has_signal - WR_no_signal) > 3pp AND avg PnL on has_signal trades > 0 AND has_signal trade count >= 20 """ import gzip import json import os import numpy as np BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) RUNS_DIR = os.path.join(BASE_DIR, "runs", "intraday_orb") ATTENTION_DIR = os.path.join(BASE_DIR, "data", "cache", "orb_attention") V23_200D = os.path.join(RUNS_DIR, "intraday_20260419_195924_f325f91a.json") V23_400D = os.path.join(RUNS_DIR, "intraday_20260419_213436_27e7c928.json") # Loaded once per run _attention_cache: dict[str, dict] = {} def load_attention(ticker: str) -> dict: if ticker in _attention_cache: return _attention_cache[ticker] path = os.path.join(ATTENTION_DIR, f"{ticker}.json.gz") if not os.path.exists(path): _attention_cache[ticker] = {} return {} try: with gzip.open(path) as f: data = json.load(f) days = data.get("days", {}) except Exception: days = {} _attention_cache[ticker] = days return days def has_signal(entry: dict, wiki_thresh: float, article_thresh: int) -> bool: wiki = entry.get("attention_wiki_spike_10d") or 0.0 articles = entry.get("attention_article_count_3d") or 0 return wiki >= wiki_thresh or articles >= article_thresh def analyze_trades(trades: list, label: str, wiki_thresh: float = 2.0, article_thresh: int = 1) -> dict: signal_trades = [] no_signal_trades = [] missing = 0 for trade in trades: ticker = trade["ticker"] date = trade["date"] pnl = trade["pnl"] win = pnl > 0 days = load_attention(ticker) if date not in days: missing += 1 no_signal_trades.append({"pnl": pnl, "win": win, "ticker": ticker, "date": date}) elif has_signal(days[date], wiki_thresh, article_thresh): signal_trades.append({ "pnl": pnl, "win": win, "ticker": ticker, "date": date, "wiki": days[date].get("attention_wiki_spike_10d", 0), "articles": days[date].get("attention_article_count_3d", 0), }) else: no_signal_trades.append({"pnl": pnl, "win": win, "ticker": ticker, "date": date}) def stats(bucket): if not bucket: return {"count": 0, "wr": 0.0, "avg_pnl": 0.0, "total_pnl": 0.0} wins = sum(1 for t in bucket if t["win"]) return { "count": len(bucket), "wr": wins / len(bucket), "avg_pnl": float(np.mean([t["pnl"] for t in bucket])), "total_pnl": sum(t["pnl"] for t in bucket), } s = stats(signal_trades) n = stats(no_signal_trades) spread = s["wr"] - n["wr"] print(f"\n{'='*60}") print(f" {label} [wiki>={wiki_thresh}, articles>={article_thresh}]") print(f"{'='*60}") print(f" Total trades: {len(trades)} (cache miss: {missing})") print(f" HAS SIGNAL: count={s['count']} WR={s['wr']:.1%} avg_pnl=${s['avg_pnl']:.2f}") print(f" NO SIGNAL: count={n['count']} WR={n['wr']:.1%} avg_pnl=${n['avg_pnl']:.2f}") print(f" WR spread: {spread:+.1%}") gate_pass = spread > 0.03 and s["avg_pnl"] > 0 and s["count"] >= 20 print(f"\n Gate: spread>{0.03:.0%}:{spread*100:.1f}pp avg_pnl>0:${s['avg_pnl']:.2f} count>=20:{s['count']}") print(f" => {'PROMOTE to Stage B' if gate_pass else 'FAIL'}") return { "label": label, "signal": s, "no_signal": n, "spread": spread, "gate_pass": gate_pass, "signal_trades": signal_trades, } def analyze_2024_losing_days(run_400d: dict): """For each V23 2024 losing day, what % of cached tickers had wiki signal?""" print(f"\n{'='*60}") print(" 2024 LOSING DAY ANALYSIS (400d run)") print(f"{'='*60}") losing_2024 = [ d for d in run_400d.get("daily_summary", []) if d["date"].startswith("2024") and d["daily_pnl"] < 0 and d["trades"] > 0 ] print(f" V23 2024 losing days (with trades): {len(losing_2024)}") tickers = [f.replace(".json.gz", "") for f in os.listdir(ATTENTION_DIR) if f.endswith(".json.gz")][:200] rows = [] for day in losing_2024[:8]: date = day["date"] checked = with_signal = 0 for ticker in tickers: d = load_attention(ticker) if date in d: checked += 1 if has_signal(d[date], 2.0, 1): with_signal += 1 pct = (with_signal / checked * 100) if checked else 0 rows.append((date, day["daily_pnl"], checked, with_signal, pct)) print(f" {date}: V23 PnL=${day['daily_pnl']:+.0f} checked={checked} signal={with_signal} ({pct:.0f}%)") if rows: print(f"\n Avg signal coverage on V23 losing days: {np.mean([r[4] for r in rows]):.0f}%") def main(): print("V28 Stage A: Wiki signal sanity probe") _attention_cache.clear() with open(V23_200D) as f: run_200d = json.load(f) with open(V23_400D) as f: run_400d = json.load(f) r200 = analyze_trades(run_200d["trades"], "V23 200d", wiki_thresh=2.0) r400 = analyze_trades(run_400d["trades"], "V23 400d", wiki_thresh=2.0) print() r200_low = analyze_trades(run_200d["trades"], "V23 200d [sensitivity: wiki>=1.5]", wiki_thresh=1.5) r400_low = analyze_trades(run_400d["trades"], "V23 400d [sensitivity: wiki>=1.5]", wiki_thresh=1.5) analyze_2024_losing_days(run_400d) # Final verdict any_pass = r200["gate_pass"] or r400["gate_pass"] or r200_low["gate_pass"] or r400_low["gate_pass"] print(f"\n\n{'='*60}") print(" FINAL VERDICT") print(f"{'='*60}") print(f" 200d (thresh=2.0): {'PASS' if r200['gate_pass'] else 'FAIL'}") print(f" 400d (thresh=2.0): {'PASS' if r400['gate_pass'] else 'FAIL'}") print(f" 200d (thresh=1.5): {'PASS' if r200_low['gate_pass'] else 'FAIL'}") print(f" 400d (thresh=1.5): {'PASS' if r400_low['gate_pass'] else 'FAIL'}") print(f"\n Decision: {'PROMOTE to Stage B' if any_pass else 'ABORT V28 — signal is noise in existing cache'}") if any_pass: best = max([r200, r400, r200_low, r400_low], key=lambda r: r["spread"]) print(f"\n Best: {best['label']} spread={best['spread']:+.1%}") top = sorted(best["signal_trades"], key=lambda x: x["pnl"], reverse=True)[:10] print(" Top signal trades:") for t in top: print(f" {t['date']} {t['ticker']:6s} wiki={t['wiki']:.2f} articles={t['articles']} pnl=${t['pnl']:+.0f}") if __name__ == "__main__": main()