"""Earnings gap diagnostic (2a) + SUE IC (2b). Local research only. Requires ``earnings_events`` on the snapshot (see backfill_earnings_events.py). Example ------- python scripts/run_earnings_research.py \\ --snapshot backtest_snapshots/prod.sqlite --workers 6 --allow-spawn """ from __future__ import annotations import argparse import asyncio import json import math import os import sys from collections import defaultdict from datetime import date, datetime, timedelta from pathlib import Path from typing import Any from sqlalchemy import create_engine, text from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from app.ssl_bootstrap import bootstrap_ssl # noqa: E402 bootstrap_ssl() IRON_IC_BAR = 0.03 MIN_RELIABLE = 12 SUE_CARRY_DAYS = 63 SUE_TRAIL = 8 def _sqlite_url(path: Path) -> str: return f"sqlite+aiosqlite:///{path.resolve().as_posix()}" def _parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description=__doc__) p.add_argument("--snapshot", default="backtest_snapshots/prod.sqlite") p.add_argument("--workers", type=int, default=6) p.add_argument("--allow-spawn", action="store_true") p.add_argument("--skip-2a", action="store_true") p.add_argument("--skip-2b", action="store_true") p.add_argument("--quiet", action="store_true") p.add_argument("--out", default=None) return p.parse_args() def _load_earnings(snapshot: Path) -> list[dict]: engine = create_engine( f"sqlite:///{snapshot.resolve().as_posix()}", future=True, ) try: with engine.connect() as conn: # Table must exist. tables = { r[0] for r in conn.execute( text("SELECT name FROM sqlite_master WHERE type='table'") ) } if "earnings_events" not in tables: raise SystemExit( "earnings_events table missing — run scripts/backfill_earnings_events.py" ) rows = conn.execute( text( """ SELECT symbol, announce_date, announce_time, eps_estimate, eps_actual, revenue_estimate, revenue_actual FROM earnings_events ORDER BY symbol, announce_date """ ) ).fetchall() meta = {} if "earnings_backfill_meta" in tables: meta = { "done": int( conn.execute( text( "SELECT COUNT(*) FROM earnings_backfill_meta " "WHERE status='done'" ) ).scalar_one() ), "universe_tickers": int( conn.execute(text("SELECT COUNT(*) FROM tickers")).scalar_one() ), } finally: engine.dispose() events = [ { "symbol": str(r[0]).upper(), "announce_date": date.fromisoformat(str(r[1])[:10]), "announce_time": r[2], "eps_estimate": r[3], "eps_actual": r[4], "revenue_estimate": r[5], "revenue_actual": r[6], } for r in rows ] return events, meta def _percentile(xs: list[float], q: float) -> float | None: if not xs: return None s = sorted(xs) if len(s) == 1: return s[0] idx = q * (len(s) - 1) lo = int(math.floor(idx)) hi = int(math.ceil(idx)) if lo == hi: return s[lo] w = idx - lo return s[lo] * (1 - w) + s[hi] * w def _r_dist(rs: list[float]) -> dict[str, Any]: if not rs: return {"n": 0} return { "n": len(rs), "mean": round(sum(rs) / len(rs), 4), "win_rate": round(sum(1 for r in rs if r > 0) / len(rs), 4), "p05": round(_percentile(rs, 0.05), 4), "p25": round(_percentile(rs, 0.25), 4), "p50": round(_percentile(rs, 0.50), 4), "p75": round(_percentile(rs, 0.75), 4), "p95": round(_percentile(rs, 0.95), 4), "min": round(min(rs), 4), "max": round(max(rs), 4), } def _trading_days_between( entry: date, exit_: date, calendar: set[date] ) -> list[date]: """Inclusive trading dates in [entry, exit_] present on the union calendar.""" out = [] d = entry while d <= exit_: if d in calendar: out.append(d) d += timedelta(days=1) return out def _nth_trading_day_after( start: date, n: int, ordered_calendar: list[date] ) -> date | None: """First calendar date strictly after ``start``, then + (n-1) more sessions. announce+1 trading day: n=1 → first session after announce date (if announce is a trading day, still use the *next* session for PIT). """ # Sessions strictly after start. after = [d for d in ordered_calendar if d > start] if len(after) < n: return None return after[n - 1] def _build_sue_series( events_by_symbol: dict[str, list[dict]], prices: dict[str, tuple], ) -> dict[str, dict[date, float]]: """symbol → {asof_date: sue_value} for days when SUE is live (announce+1 .. +63).""" out: dict[str, dict[date, float]] = {} for sym, cols in prices.items(): ords = cols[0] closes = cols[4] dates = [date.fromordinal(int(o)) for o in ords] if not dates: continue ordered = dates # already chronological cal_set = set(ordered) events = events_by_symbol.get(sym.upper(), []) # Chronological surprises with actual+estimate. surprises: list[tuple[date, float, float]] = [] # announce, surprise, close_for_scale for ev in events: act, est = ev.get("eps_actual"), ev.get("eps_estimate") if act is None or est is None: continue ad = ev["announce_date"] # Close on/before announce for price fallback scale. close_px = None for d, c in zip(reversed(dates), reversed(closes)): if d <= ad and float(c) > 0: close_px = float(c) break surprises.append((ad, float(act) - float(est), close_px or 1.0)) surprises.sort(key=lambda x: x[0]) sue_on_day: dict[date, float] = {} for i, (ad, surprise, px) in enumerate(surprises): trail = [surprises[j][1] for j in range(max(0, i - SUE_TRAIL), i)] # Need history of surprises; include current only for value, stdev from prior 8. if len(trail) >= 3: mean_t = sum(trail) / len(trail) var = sum((x - mean_t) ** 2 for x in trail) / (len(trail) - 1) sd = math.sqrt(var) if var > 0 else None else: sd = None if sd is not None and sd > 1e-9: sue = surprise / sd else: # Fallback: scale by price (EPS surprise / price). sue = surprise / px if px > 0 else None if sue is None or not math.isfinite(sue): continue usable_from = _nth_trading_day_after(ad, 1, ordered) if usable_from is None: continue # Carry for SUE_CARRY_DAYS trading sessions starting at usable_from. try: start_idx = ordered.index(usable_from) except ValueError: # usable_from not in this symbol's calendar (halted etc.) start_idx = next( (k for k, d in enumerate(ordered) if d >= usable_from), None ) if start_idx is None: continue end_idx = min(len(ordered) - 1, start_idx + SUE_CARRY_DAYS - 1) for k in range(start_idx, end_idx + 1): # Later announcements overwrite earlier carry (latest SUE wins). sue_on_day[ordered[k]] = sue if sue_on_day: out[sym.upper()] = sue_on_day return out async def _run_2a( snapshot: Path, events: list[dict], *, quiet: bool, workers: int, ) -> dict[str, Any]: from app.config import settings from app.services import backtest_service as bt from app.services.admin_service import get_activation_config from app.services.recommendation_service import get_recommendation_config from app.services.paper_trade_service import get_exit_policy from app.services.benchmark_service import load_benchmark_closes from app.models.ticker import Ticker from sqlalchemy import select os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1" settings.backtest_workers = workers engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True) Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) try: async with Session() as db: config = await get_recommendation_config(db) activation = await get_activation_config(db) exit_config = await get_exit_policy(db) tickers = list( (await db.execute(select(Ticker).order_by(Ticker.symbol))).scalars() ) spy = await load_benchmark_closes(db, "SPY") prices: dict[str, tuple] = {} candidates: list[dict] = [] for idx, t in enumerate(tickers): if not quiet and idx % 50 == 0: print(f" 2a fetch {idx}/{len(tickers)}", end="\r", flush=True) cols = await bt._fetch_columns(db, t.symbol) if cols is None: continue prices[t.symbol] = cols cands, _ = bt._replay_and_signals( t.symbol, cols, config, activation, spy, bt.PRODUCTION_GTL_TARGET_MODEL, "weekly", False, ) candidates.extend(cands) finally: await engine.dispose() if not quiet: print() # Production ranks + qualify. bt._assign_momentum_percentiles(candidates) bt._assign_residual_momentum_percentiles(candidates) bt._assign_low_volatility_percentiles(candidates) bt._assign_activation_momentum_percentiles(candidates) bt._assign_residual_high_vol_blend(candidates) for c in candidates: c["qualified"] = bt._momentum_qualifies(c, 80.0) longs = [ c for c in candidates if c.get("qualified") and c.get("direction") == "long" ] strategy = next(s for s in bt.PORTFOLIO_MONITOR_STRATEGIES if s.get("is_production")) entry_cfg = bt._entry_variant_config(str(strategy["entry_variant"])) assert entry_cfg is not None ranking_key = str(entry_cfg.get("ranking_key") or entry_cfg["percentile_key"]) exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get( str(exit_config.get("mode", "atr_trailing")), "atr_trail3" ) hold_days = int(exit_config.get("hold_days", 30)) trail = float(exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER)) reentry = bt._make_gate_reset_reentry_fn( longs, prices, cadence="weekly", ranking_key=ranking_key ) sim = bt._simulate_portfolio( longs, prices, spy, exit_policy, hold_days, ranking_key=ranking_key, max_positions=int(entry_cfg["max_positions"]), risk_per_trade=float(entry_cfg["risk_per_trade"]), atr_trail_multiplier=trail, post_stop_reentry_fn=reentry, fill_mode=bt.FILL_MODE_CLOSE, include_trades=True, ) if sim is None: return {"error": "no_trades"} details = sim.get("trade_details") or [] # Build per-symbol earnings announce dates. earns_by_sym: dict[str, list[date]] = defaultdict(list) for ev in events: earns_by_sym[ev["symbol"]].append(ev["announce_date"]) for sym in earns_by_sym: earns_by_sym[sym].sort() # Union trading calendar from prices. cal: set[date] = set() for cols in prices.values(): for o in cols[0]: cal.add(date.fromordinal(int(o))) ordered_cal = sorted(cal) # Map entry date → list of announce dates for symbol (for pre-entry lookback). trades_parsed: list[dict] = [] for t in details: sym = str(t.get("symbol") or "").upper() # Field names from simulator. entry_s = t.get("entry_date") or t.get("open_date") or t.get("date") exit_s = t.get("exit_date") or t.get("close_date") r = t.get("realized_r") if r is None: r = t.get("r") if entry_s is None or exit_s is None or r is None: continue entry_d = date.fromisoformat(str(entry_s)[:10]) exit_d = date.fromisoformat(str(exit_s)[:10]) announces = earns_by_sym.get(sym, []) # Earnings between entry and exit (exclusive of entry day? inclusive hold). # "between entry and exit" — any announce with entry < announce <= exit # (gap often overnight after entry). Also count announce on entry day. in_hold = [ a for a in announces if entry_d <= a <= exit_d ] # Entries within 3 trading days BEFORE an announcement: # exists announce such that entry is in the 3 sessions immediately before announce. pre_earn = False for a in announces: # trading sessions in (a-lookback, a) sessions_before = [d for d in ordered_cal if d < a] last3 = sessions_before[-3:] if len(sessions_before) >= 3 else sessions_before if entry_d in last3: pre_earn = True break trades_parsed.append({ "symbol": sym, "entry": entry_d.isoformat(), "exit": exit_d.isoformat(), "r": float(r), "earnings_in_hold": len(in_hold) > 0, "n_earnings_in_hold": len(in_hold), "entry_within_3d_before_earn": pre_earn, }) all_r = [t["r"] for t in trades_parsed] loss_lt_1r = [t for t in trades_parsed if t["r"] < -1.0] loss_with_earn = [t for t in loss_lt_1r if t["earnings_in_hold"]] pre = [t["r"] for t in trades_parsed if t["entry_within_3d_before_earn"]] other = [t["r"] for t in trades_parsed if not t["entry_within_3d_before_earn"]] return { "sim_summary": { k: sim.get(k) for k in ( "sharpe", "sharpe_se", "cagr_pct", "max_drawdown_pct", "trades", "total_return_pct", ) }, "n_trades_parsed": len(trades_parsed), "q1_losses_worse_than_minus_1r": { "n_losses_lt_minus_1r": len(loss_lt_1r), "n_with_earnings_in_hold": len(loss_with_earn), "fraction_with_earnings": ( round(len(loss_with_earn) / len(loss_lt_1r), 4) if loss_lt_1r else None ), "all_trades_with_earnings_in_hold": sum( 1 for t in trades_parsed if t["earnings_in_hold"] ), "fraction_all_trades_with_earnings": ( round( sum(1 for t in trades_parsed if t["earnings_in_hold"]) / len(trades_parsed), 4, ) if trades_parsed else None ), }, "q2_entry_within_3d_before_announce": { "pre_earn_entries": _r_dist(pre), "other_entries": _r_dist(other), "all_entries": _r_dist(all_r), "tail_trim_note": ( "Compare p95/max and mean of pre_earn vs other. " "Rising win_rate with falling mean/p95 = right-tail trim red flag." ), }, "note": "REPORT-ONLY — no filter shipped.", } async def _run_2b_ic( snapshot: Path, events: list[dict], *, quiet: bool, workers: int, ) -> dict[str, Any]: """SUE IC via harness on identical cross-sections as momentum baselines.""" from app.config import settings from app.services import backtest_service as bt from app.services.benchmark_service import load_benchmark_closes from app.models.ticker import Ticker from sqlalchemy import select from collections import defaultdict as dd os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1" os.environ["BACKTEST_SIGNAL_EVAL_ONLY"] = "1" settings.backtest_workers = workers engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True) Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) # Collect base signals + attach SUE. collected: dict = dd(lambda: dd(list)) try: async with Session() as db: tickers = list( (await db.execute(select(Ticker).order_by(Ticker.symbol))).scalars() ) spy = await load_benchmark_closes(db, "SPY") prices: dict[str, tuple] = {} for idx, t in enumerate(tickers): if not quiet and idx % 50 == 0: print(f" 2b fetch {idx}/{len(tickers)}", end="\r", flush=True) cols = await bt._fetch_columns(db, t.symbol) if cols is None: continue prices[t.symbol] = cols series = bt._signal_series( [ type( "R", (), { "date": date.fromordinal(int(cols[0][i])), "close": cols[4][i], "high": cols[2][i], "volume": cols[5][i] if len(cols) > 5 else 0, }, )() for i in range(len(cols[0])) ], spy, symbol=t.symbol, ) for name, weeks in series.items(): for wk, pairs in weeks.items(): collected[name][wk].extend(pairs) finally: await engine.dispose() if not quiet: print() # SUE series. events_by_sym: dict[str, list[dict]] = defaultdict(list) for ev in events: events_by_sym[ev["symbol"]].append(ev) sue_map = _build_sue_series(events_by_sym, prices) # Inject sue_latest into collected using mom_12_1 observations as the # weekly as-of skeleton (same weeks / symbols). sue_collected: dict = dd(list) mom_weeks = collected.get("mom_12_1") or {} for week_key, recs in mom_weeks.items(): for rec in recs: pair = bt._obs_val_fwd(rec) if pair is None: continue _val, fwd = pair sym = None if isinstance(rec, dict): sym = rec.get("symbol") if not sym: continue # Need as-of date: recover from week — use Friday of ISO week as proxy # is weak. Better: re-derive from prices weekly indices. # Store asof on rich recs? Current rich rows lack asof date. # Fall back: compute SUE observations directly from prices weekly as-ofs. pass # Direct weekly as-of SUE + forward return (authoritative). for sym, cols in prices.items(): ords, _o, highs, _l, closes, _v = cols dates = [date.fromordinal(int(o)) for o in ords] sue_days = sue_map.get(sym.upper()) or {} if not sue_days: continue n = len(dates) # weekly as-of indices: reuse harness helper via fake records. records = [ type("R", (), {"date": dates[i], "close": closes[i], "high": highs[i]})() for i in range(n) ] for i in bt._weekly_asof_indices(records): j = i + bt.HORIZON if j >= n or closes[i] <= 0: continue asof = dates[i] sue = sue_days.get(asof) if sue is None: continue fwd = float(closes[j]) / float(closes[i]) - 1.0 iso = asof.isocalendar() week_key = (iso[0], iso[1]) # Also grab mom for conditional. mom = None if i >= 252 and closes[i - 252] > 0: mom = float(closes[i - 21]) / float(closes[i - 252]) - 1.0 sue_collected[week_key].append({ "val": float(sue), "fwd": fwd, "symbol": sym, "mom_12_1": mom, }) collected["sue_latest"] = sue_collected signal_eval = bt._signal_evaluation(collected) # Fair side-by-side: re-evaluate mom baselines on the *same* (symbol, week) # observations where SUE is present (incomplete backfill otherwise inflates # mom N relative to SUE). sue_pairs_by_week = sue_collected restricted: dict = dd(lambda: dd(list)) for week_key, recs in sue_pairs_by_week.items(): syms = {str(r.get("symbol")).upper() for r in recs if r.get("symbol")} for base_name in ("mom_12_1", "mom_12_1_resid"): base_recs = (collected.get(base_name) or {}).get(week_key) or [] for rec in base_recs: pair = bt._obs_val_fwd(rec) if pair is None: continue sym = None if isinstance(rec, dict): sym = rec.get("symbol") if not sym or str(sym).upper() not in syms: continue restricted[base_name][week_key].append(rec) restricted["sue_latest"][week_key].extend(recs) restricted_eval = bt._signal_evaluation(restricted) # Momentum-conditional: IC of SUE within top mom quintile each week. cond_ics: list[float] = [] stride = max(1, round(bt.HORIZON / 5)) usable = [wk for wk, recs in sue_collected.items() if len(recs) >= bt.MIN_CROSS_SECTION] kept = bt._nonoverlapping_weeks(usable, stride) for wk in kept: recs = sue_collected[wk] with_mom = [r for r in recs if r.get("mom_12_1") is not None] if len(with_mom) < bt.MIN_CROSS_SECTION: continue ordered = sorted(with_mom, key=lambda r: float(r["mom_12_1"])) k = max(1, len(ordered) // 5) top = ordered[-k:] if len(top) < 5: continue ic = bt._spearman( [float(r["val"]) for r in top], [float(r["fwd"]) for r in top], ) if ic is not None: cond_ics.append(ic) if cond_ics: mean_c = sum(cond_ics) / len(cond_ics) if len(cond_ics) > 1: std = math.sqrt( sum((x - mean_c) ** 2 for x in cond_ics) / (len(cond_ics) - 1) ) t_c = mean_c / std * math.sqrt(len(cond_ics)) if std > 0 else None else: t_c = None mom_cond = { "mean_ic": round(mean_c, 4), "ic_t_stat": round(t_c, 2) if t_c is not None else None, "weeks": len(cond_ics), "note": "IC of sue_latest within top mom_12_1 quintile (non-overlapping weeks)", } else: mom_cond = {"mean_ic": None, "weeks": 0} def _find(name: str) -> dict | None: for row in signal_eval: if row.get("signal") == name: return row return None sue = _find("sue_latest") grade = { "green": False, "reason": "sue_latest missing", } if sue: mean_ic = sue.get("mean_ic") t = sue.get("ic_t_stat") reliable = bool(sue.get("reliable")) sign_ok = mean_ic is not None and float(mean_ic) > 0 mag_ok = mean_ic is not None and abs(float(mean_ic)) >= IRON_IC_BAR grade = { "green": bool(sign_ok and mag_ok and reliable), "checks": { "mean_ic": mean_ic, "sign_positive": sign_ok, "abs_ge_0_03": mag_ok, "reliable": reliable, "ic_t_stat": t, "weeks": sue.get("weeks"), }, "reason": ( "iron rule cleared — STOP; book-integration is a separate human step" if (sign_ok and mag_ok and reliable) else "iron rule not met" ), "row": sue, } def _find_r(name: str) -> dict | None: for row in restricted_eval: if row.get("signal") == name: return row return None # Side-by-side baselines from same evaluation. side = { name: _find(name) for name in ( "mom_12_1", "mom_12_1_resid", "sue_latest", "fip_id", ) } side_restricted = { name: _find_r(name) for name in ("mom_12_1", "mom_12_1_resid", "sue_latest") } return { "signal_eval_side_by_side": side, "signal_eval_identical_sue_subset": side_restricted, "identical_subset_note": ( "Mom baselines re-scored only on (week, symbol) cells where SUE exists. " "Use this table when backfill is incomplete — full-universe mom N is not comparable." ), "full_signal_eval": signal_eval, "sue_grade": grade, "momentum_conditional_sue": mom_cond, "sue_coverage": { "symbols_with_sue": len(sue_map), "avg_weeks_with_sue": ( round( sum(len(v) for v in sue_collected.values()) / max(1, len(sue_collected)), 1, ) if sue_collected else 0 ), "weeks_with_min_cross_section": len(usable), }, } def _write_md(path: Path, payload: dict) -> None: pre = path.read_text(encoding="utf-8") if path.exists() else "" marker = "## Results" idx = pre.find(marker) header = pre[:idx] if idx >= 0 else pre.split("## Verdict")[0] lines = [ header.rstrip(), "", "## Results", "", f"Generated: `{payload.get('generated_at')}`", "", "### Data provenance", "", f"```json\n{json.dumps(payload.get('data_provenance') or {}, indent=2, default=str)}\n```", "", "### 2a — Earnings-gap risk (report-only)", "", ] a = payload.get("experiment_2a") if not a: lines.append("_Skipped or unavailable._") else: lines.append(f"```json\n{json.dumps(a, indent=2, default=str)}\n```") lines.extend(["", "### 2b — SUE / PEAD IC", ""]) b = payload.get("experiment_2b") if not b: lines.append("_Skipped or unavailable._") else: side = b.get("signal_eval_side_by_side") or {} lines.extend([ "| signal | mean_ic | ic_t_stat | weeks | avg_N | reliable |", "|---|---:|---:|---:|---:|---|", ]) for name in ( "mom_12_1", "mom_12_1_resid", "sue_latest", "fip_id", ): r = side.get(name) or {} lines.append( f"| {name} | {r.get('mean_ic', '')} | {r.get('ic_t_stat', '')} | " f"{r.get('weeks', '')} | {r.get('avg_cross_section', '')} | " f"{r.get('reliable', '')} |" ) lines.extend([ "", f"**SUE grade:** `{json.dumps(b.get('sue_grade') or {}, default=str)}`", "", f"**Momentum-conditional SUE:** `{json.dumps(b.get('momentum_conditional_sue') or {}, default=str)}`", "", ]) lines.extend([ "", "## Verdict", "", f"**{payload.get('verdict')}**", "", payload.get("verdict_detail") or "", "", "## What a human must decide next", "", payload.get("human_next") or "- Review; no auto-ship.", "", f"Artifacts: `{payload.get('report_path')}`", "", ]) path.write_text("\n".join(lines) + "\n", encoding="utf-8") async def _main() -> None: args = _parse_args() snapshot = Path(args.snapshot) if not snapshot.exists(): raise SystemExit(f"Missing snapshot {snapshot}") if args.allow_spawn: os.environ["BACKTEST_ALLOW_SPAWN"] = "1" events, meta = _load_earnings(snapshot) # Race guard lite on earnings completeness. provenance = { "snapshot": str(snapshot.resolve()), "n_earnings_events": len(events), "backfill_meta": meta, "announce_range": { "min": min((e["announce_date"] for e in events), default=None), "max": max((e["announce_date"] for e in events), default=None), }, "with_actual_and_estimate": sum( 1 for e in events if e.get("eps_actual") is not None and e.get("eps_estimate") is not None ), } print( f"Earnings events: {provenance['n_earnings_events']} " f"(with act+est={provenance['with_actual_and_estimate']}) meta={meta}" ) if meta and meta.get("done", 0) < 0.9 * (meta.get("universe_tickers") or 1): print( "WARNING: earnings backfill incomplete " f"({meta.get('done')}/{meta.get('universe_tickers')}). " "Results may be biased; resume backfill." ) exp_2a = None exp_2b = None if not args.skip_2a: print("Running 2a earnings-gap diagnostic…") exp_2a = await _run_2a( snapshot, events, quiet=args.quiet, workers=args.workers ) print( " 2a losses<-1R with earnings:", (exp_2a.get("q1_losses_worse_than_minus_1r") or {}), ) if not args.skip_2b: print("Running 2b SUE IC harness…") exp_2b = await _run_2b_ic( snapshot, events, quiet=args.quiet, workers=args.workers ) g = exp_2b.get("sue_grade") or {} print(f" 2b SUE green={g.get('green')} {g.get('reason')}") # Verdict if exp_2b and (exp_2b.get("sue_grade") or {}).get("green"): verdict = "PROMOTE (2b SUE) — STOP for human wire design" detail = ( "SUE cleared iron rule. No book integration without human approval. " "2a remains report-only." ) human = ( "- Design tilt vs second gate if desired.\n" "- Do not auto-filter from 2a without separate approval + tail review." ) else: sue_ic = None if exp_2b: sue_ic = ((exp_2b.get("sue_grade") or {}).get("row") or {}).get("mean_ic") if sue_ic is not None and abs(float(sue_ic)) >= 0.015: verdict = "PARK" detail = f"SUE IC={sue_ic} below iron bar or unreliable; keep data, no wire." else: verdict = "DEAD (2b) / REPORT-ONLY (2a)" detail = ( "SUE does not clear iron rule on this window. " "2a distributions for human risk review only — no filter." ) human = ( "- No SUE book change.\n" "- Read 2a tails before considering any earnings-avoid filter." ) stamp = datetime.now().strftime("%Y%m%d-%H%M%S") out = Path(args.out) if args.out else Path("reports") / f"earnings-gap-sue-{stamp}.json" payload = { "generated_at": datetime.now().isoformat(), "data_provenance": provenance, "experiment_2a": exp_2a, "experiment_2b": exp_2b, "verdict": verdict, "verdict_detail": detail, "human_next": human, "report_path": str(out.as_posix()), "fmp_note": ( "Bulk earnings-calendar is paid (402 on free tier). " "Backfill used per-symbol /stable/earnings; see earnings-backfill-status.json." ), } out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(payload, indent=2, default=str) + "\n", encoding="utf-8") md = Path("docs/research/earnings-gap-and-sue.md") _write_md(md, payload) out.with_suffix(".md").write_text(md.read_text(encoding="utf-8"), encoding="utf-8") print(f"Verdict: {verdict}") print(f"Wrote {out}") if __name__ == "__main__": asyncio.run(_main())