diff --git a/scripts/v28_probe/wiki_signal_probe.py b/scripts/v28_probe/wiki_signal_probe.py new file mode 100644 index 0000000..d847387 --- /dev/null +++ b/scripts/v28_probe/wiki_signal_probe.py @@ -0,0 +1,183 @@ +""" +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()